fix(heal): fence format repair during pool transitions

feat(connect): answer offline enrolment challenges without a network (#6335 )
2026-08-21 20:06:37 +00:00 · 2026-08-22 00:25:32 +08:00 · 2026-08-21 13:16:18 +00:00
40 changed files with 2685 additions and 38 deletions
@@ -3843,6 +3843,7 @@ dependencies = [
 "s3s",
 "serde",
 "serde_json",
+ "serial_test",
 "sha2 0.11.0",
 "suppaftp",
 "time",
@@ -9803,6 +9804,7 @@ dependencies = [
 "rustfs-replication",
 "rustfs-storage-api",
 "s3s",
+ "serial_test",
 "temp-env",
 "time",
 "tokio",
@@ -9918,6 +9920,7 @@ dependencies = [
 "rustfs-config",
 "rustfs-io-metrics",
 "rustfs-utils",
+ "serial_test",
 "temp-env",
 "tempfile",
 "tokio",
@@ -10290,6 +10293,7 @@ dependencies = [
 "s3s",
 "serde",
 "serde_json",
+ "serial_test",
 "sha2 0.11.0",
 "temp-env",
 "tempfile",
@@ -96,6 +96,7 @@ tokio-stream = { workspace = true }
 rustfs-madmin.workspace = true
 rustfs-filemeta.workspace = true
 bytes = { workspace = true, features = ["serde"] }
+serial_test = { workspace = true }
 aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
 aws-sdk-sts = { workspace = true, default-features = false, features = ["default-https-client", "rt-tokio"] }
 aws-config = { workspace = true }
@@ -55,6 +55,7 @@ mod tests {
    use aws_sdk_s3::Client;
    use aws_sdk_s3::primitives::ByteStream;
    use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+    use serial_test::serial;
    use sha2::{Digest, Sha256};
    use std::error::Error;
    use tokio::time::{Duration, timeout};
@@ -268,6 +269,7 @@ mod tests {
    /// stripes) and a multipart object (3 parts × 5 MiB) must GET back as a
    /// full, byte-identical body with the correct Content-Length. No early EOF.
    #[tokio::test]
+    #[serial]
    async fn degraded_read_large_objects_with_one_disk_offline_return_full_body() -> TestResult {
        init_logging();
        info!("dist-13 (a): large-object degraded read with one of four disks offline");
@@ -333,6 +335,7 @@ mod tests {
    /// mid-stream — the exact window the fixes had to reconstruct through rather
    /// than truncate.
    #[tokio::test]
+    #[serial]
    async fn degraded_read_reconstructs_through_midstream_bitrot_within_quorum() -> TestResult {
        init_logging();
        info!("dist-13 (b): mid-stream bitrot within quorum must reconstruct a full body");
@@ -390,6 +393,7 @@ mod tests {
    /// Content-Length. `get_checked` panics on that forbidden outcome, so this
    /// test fails loudly if the truncation bug ever returns.
    #[tokio::test]
+    #[serial]
    async fn beyond_quorum_degraded_read_never_silently_truncates() -> TestResult {
        init_logging();
        info!("dist-13 (c): beyond-quorum degraded read must fail, never 200+truncated");
@@ -51,6 +51,7 @@ mod tests {
    use aws_sdk_s3::Client;
    use aws_sdk_s3::primitives::ByteStream;
    use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+    use serial_test::serial;
    use std::error::Error;
    use tokio::time::{Duration, timeout};
    use tracing::info;
@@ -128,6 +129,7 @@ mod tests {
    /// the body — and assert the server log names the object, at the log level a
    /// default deployment actually runs with.
    #[tokio::test]
+    #[serial]
    async fn midstream_get_failure_is_logged_with_the_object_at_default_log_level() -> TestResult {
        init_logging();
        info!("rustfs#4784: a mid-stream GET failure must name its object in the source log");
@@ -46,6 +46,7 @@ use prost::Message;
 use rustfs_signer::constants::UNSIGNED_PAYLOAD;
 use rustfs_signer::sign_v4;
 use s3s::Body;
+use serial_test::serial;
 use std::collections::BTreeMap;
 use std::convert::Infallible;
 use std::error::Error;
@@ -1694,6 +1695,7 @@ fn assert_storage_layout(
 }

 #[tokio::test]
+#[serial]
 async fn four_node_inline_storage_and_get_boundaries() -> TestResult {
    init_logging();

@@ -1765,6 +1767,7 @@ async fn four_node_inline_storage_and_get_boundaries() -> TestResult {
 }

 #[tokio::test]
+#[serial]
 async fn four_node_empty_legacy_volumes_start_as_fresh() -> TestResult {
    init_logging();

@@ -1802,6 +1805,7 @@ async fn four_node_empty_legacy_volumes_start_as_fresh() -> TestResult {
 }

 #[tokio::test]
+#[serial]
 async fn four_node_inline_fallback_controls() -> TestResult {
    init_logging();

@@ -1866,6 +1870,7 @@ async fn four_node_inline_fallback_controls() -> TestResult {
 }

 #[tokio::test]
+#[serial]
 async fn four_node_compressed_inline_fallback() -> TestResult {
    init_logging();

@@ -1900,6 +1905,7 @@ async fn four_node_compressed_inline_fallback() -> TestResult {
 /// Multipart disk compression is live again, so a compression-enabled cluster classifies multipart objects as compressed and the roundtrip (full GET plus partNumber GET) must still return the original bytes.
 /// Reverting the multipart compression fix must fail this test.
 #[tokio::test]
+#[serial]
 async fn four_node_multipart_disk_compression_roundtrip() -> TestResult {
    init_logging();

@@ -1946,6 +1952,7 @@ async fn four_node_multipart_disk_compression_roundtrip() -> TestResult {
 /// read costs on the order of the covering part's block size against a ~5 MiB
 /// object.
 #[tokio::test]
+#[serial]
 async fn four_node_compressed_multipart_tail_range_reads_are_bounded() -> TestResult {
    init_logging();

@@ -2012,6 +2019,7 @@ async fn four_node_compressed_multipart_tail_range_reads_are_bounded() -> TestRe
 }

 #[tokio::test]
+#[serial]
 async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> TestResult {
    init_logging();

@@ -2115,6 +2123,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te
 }

 #[tokio::test]
+#[serial]
 async fn four_node_add_tier_converges() -> TestResult {
    init_logging();

@@ -2133,6 +2142,7 @@ async fn four_node_add_tier_converges() -> TestResult {
 }

 #[tokio::test]
+#[serial]
 async fn four_node_add_tier_converges_after_offline_node_restart_without_second_mutation() -> TestResult {
    init_logging();

@@ -2154,6 +2164,7 @@ async fn four_node_add_tier_converges_after_offline_node_restart_without_second_
 }

 #[tokio::test]
+#[serial]
 async fn four_node_manual_transition_job_status_survives_node_restart() -> TestResult {
    init_logging();

@@ -2228,6 +2239,7 @@ async fn four_node_manual_transition_job_status_survives_node_restart() -> TestR
 }

 #[tokio::test]
+#[serial]
 async fn four_node_manual_transition_distributed_admission_conflict_reports_status_and_backpressure() -> TestResult {
    init_logging();

@@ -2369,6 +2381,7 @@ async fn four_node_manual_transition_distributed_admission_conflict_reports_stat
 }

 #[tokio::test]
+#[serial]
 #[ignore = "manual #1508 evidence harness: starts a 4-node cluster, a remote tier, and an in-flight transition job"]
 async fn four_node_manual_transition_rollout_non_empty_restart_readback() -> TestResult {
    init_logging();
@@ -2473,6 +2486,7 @@ async fn four_node_manual_transition_rollout_non_empty_restart_readback() -> Tes
 }

 #[tokio::test]
+#[serial]
 async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_transition() -> TestResult {
    init_logging();

@@ -2584,6 +2598,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_
 }

 #[tokio::test]
+#[serial]
 async fn four_node_transitioned_inline_fallback() -> TestResult {
    init_logging();

@@ -19,6 +19,7 @@
 //! multipart upload behaviour.

 use crate::common::{TEST_BUCKET, init_logging};
+use serial_test::serial;
 use tokio::time::{Duration, sleep};
 use tracing::{error, info};

@@ -61,6 +62,7 @@ impl VaultKmsTestContext {
 }

 #[tokio::test]
+#[serial]
 async fn test_vault_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();
    if skip_if_kms_admin_tool_unavailable("test_vault_kms_end_to_end") {
@@ -116,6 +118,7 @@ async fn test_vault_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + S
 }

 #[tokio::test]
+#[serial]
 async fn test_vault_kms_key_isolation() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();
    if skip_if_kms_admin_tool_unavailable("test_vault_kms_key_isolation") {
@@ -202,6 +205,7 @@ async fn test_vault_kms_key_isolation() -> Result<(), Box<dyn std::error::Error
 }

 #[tokio::test]
+#[serial]
 async fn test_vault_kms_large_file() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();
    if skip_if_kms_admin_tool_unavailable("test_vault_kms_large_file") {
@@ -266,6 +270,7 @@ async fn test_vault_kms_large_file() -> Result<(), Box<dyn std::error::Error + S
 }

 #[tokio::test]
+#[serial]
 async fn test_vault_kms_multipart_upload() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();
    if skip_if_kms_admin_tool_unavailable("test_vault_kms_multipart_upload") {
@@ -296,6 +301,7 @@ async fn test_vault_kms_multipart_upload() -> Result<(), Box<dyn std::error::Err
 }

 #[tokio::test]
+#[serial]
 async fn test_vault_kms_key_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();
    if skip_if_kms_admin_tool_unavailable("test_vault_kms_key_operations") {
@@ -17,6 +17,7 @@
 use crate::common::{awscurl_delete, awscurl_put, init_logging};
 use crate::policy::test_env::PolicyTestEnvironment;
 use aws_sdk_s3::primitives::ByteStream;
+use serial_test::serial;
 use tracing::info;

 /// Helper function to create a regular user with given credentials
@@ -121,6 +122,7 @@ async fn cleanup_user_and_policy(env: &PolicyTestEnvironment, username: &str, po

 /// Test AWS policy variables with single-value scenarios
 #[tokio::test(flavor = "multi_thread")]
+#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_single_value() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_single_value_impl().await
@@ -273,6 +275,7 @@ pub async fn test_aws_policy_variables_single_value_impl_with_env(

 /// Test AWS policy variables with multi-value scenarios
 #[tokio::test(flavor = "multi_thread")]
+#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_multi_value() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_multi_value_impl().await
@@ -398,6 +401,7 @@ pub async fn test_aws_policy_variables_multi_value_impl_with_env(

 /// Test AWS policy variables with variable concatenation
 #[tokio::test(flavor = "multi_thread")]
+#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_concatenation() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_concatenation_impl().await
@@ -487,6 +491,7 @@ pub async fn test_aws_policy_variables_concatenation_impl_with_env(

 /// Test AWS policy variables with nested scenarios
 #[tokio::test(flavor = "multi_thread")]
+#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_nested() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_nested_impl().await
@@ -504,6 +509,7 @@ pub async fn test_aws_policy_variables_nested_impl() -> Result<(), Box<dyn std::

 /// Test AWS policy variables with STS temporary credentials
 #[tokio::test(flavor = "multi_thread")]
+#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_sts() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_sts_impl().await
@@ -699,6 +705,7 @@ pub async fn test_aws_policy_variables_sts_impl_with_env(

 /// Test AWS policy variables with deny scenarios
 #[tokio::test(flavor = "multi_thread")]
+#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_deny() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_deny_impl().await
@@ -14,6 +14,7 @@

 use crate::common::init_logging;
 use crate::policy::test_env::PolicyTestEnvironment;
+use serial_test::serial;
 use std::time::Instant;
 use tokio::time::{Duration, sleep};
 use tracing::{error, info};
@@ -212,6 +213,7 @@ impl PolicyTestSuite {

 /// Test suite
 #[tokio::test]
+#[serial]
 #[ignore = "Connects to existing rustfs server"]
 async fn test_policy_critical_suite() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let config = TestSuiteConfig {
@@ -41,6 +41,7 @@ use reqwest::Client;
 use rustfs_signer::constants::UNSIGNED_PAYLOAD;
 use rustfs_signer::sign_v4;
 use s3s::Body;
+use serial_test::serial;
 use tokio::process::Command;
 use tracing::info;

@@ -820,6 +821,7 @@ pub async fn test_webdav_core_operations() -> Result<()> {
 }

 #[tokio::test]
+#[serial]
 async fn test_webdav_core_operations_direct() -> Result<()> {
    test_webdav_core_operations().await
 }
@@ -27,6 +27,7 @@ mod tests {
    use aws_sdk_s3::Client;
    use aws_sdk_s3::primitives::ByteStream;
    use aws_sdk_s3::types::{BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, VersioningConfiguration};
+    use serial_test::serial;
    use sha2::{Digest, Sha256};
    use std::collections::HashSet;
    use std::error::Error;
@@ -156,6 +157,7 @@ mod tests {
    /// content, degraded writes must succeed, and everything must still
    /// verify after the disk returns.
    #[tokio::test]
+    #[serial]
    async fn test_degraded_read_write_with_one_disk_offline() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        init_logging();
        info!("Reliability: degraded read/write with one of four disks offline");
@@ -208,6 +210,7 @@ mod tests {
    /// bytes to a reader: per-shard bitrot checksums reject the bad shard and
    /// the object is reconstructed from the remaining shards.
    #[tokio::test]
+    #[serial]
    async fn test_bitrot_corrupted_shard_read_returns_correct_data() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        init_logging();
        info!("Reliability: GET must read through a bitrot-corrupted shard");
@@ -250,6 +253,7 @@ mod tests {
    /// heal, and require the replaced disk to be rebuilt and all content to
    /// verify against the sha256 manifest.
    #[tokio::test]
+    #[serial]
    async fn test_fresh_disk_replacement_heals_after_sigkill_restart() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        init_logging();
        info!("Reliability: fresh-disk replacement heals after SIGKILL restart");
@@ -323,6 +327,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_versioned_shard_census_selects_each_version_data_dir() -> Result<(), Box<dyn Error + Send + Sync>> {
        init_logging();
        info!("Reliability: physical shard census selects the requested object version");
@@ -29,6 +29,7 @@ mod tests {
    use aws_sdk_s3::primitives::ByteStream;
    use aws_sdk_s3::types::{BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, VersioningConfiguration};
    use http::Method;
+    use serial_test::serial;
    use sha2::{Digest, Sha256};
    use std::collections::BTreeSet;
    use std::error::Error;
@@ -1060,6 +1061,7 @@ mod tests {
    /// Linux mount namespaces are per-thread; keep mount setup and process
    /// spawning on one OS thread so child RustFS nodes inherit the test mounts.
    #[tokio::test(flavor = "current_thread")]
+    #[serial]
    #[ignore = "requires Linux root/CAP_SYS_ADMIN and RUSTFS_PRIVILEGED_REPLACEMENT_E2E=1"]
    async fn test_privileged_3x4_auto_replacement_rebuilds_ec8_plus_4_without_admin_heal()
    -> Result<(), Box<dyn Error + Send + Sync>> {
@@ -1073,6 +1075,7 @@ mod tests {
    /// Linux mount namespaces are per-thread; keep mount setup and process
    /// spawning on one OS thread so child RustFS nodes inherit the test mounts.
    #[tokio::test(flavor = "current_thread")]
+    #[serial]
    #[ignore = "requires Linux root/CAP_SYS_ADMIN and RUSTFS_PRIVILEGED_REPLACEMENT_E2E=1"]
    async fn test_privileged_3x4_auto_replacement_rebuilds_ec6_plus_6_without_admin_heal()
    -> Result<(), Box<dyn Error + Send + Sync>> {
@@ -1675,7 +1675,7 @@ impl PoolMeta {
        self.load_no_lock(pool).await
    }

-    async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
+    pub(crate) async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
    where
        S: EcstoreObjectIO,
    {
@@ -13,7 +13,12 @@
 // limitations under the License.

 use super::*;
+use crate::core::pools::POOL_META_NAME;
+use crate::services::rebalance::{REBAL_META_NAME, RebalStatus};
+use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::heal::HealOperations as _;
+use crate::storage_api_contracts::namespace::NamespaceLocking as _;
+use rustfs_lock::NamespaceLockGuard;
 use tracing::trace;

 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
@@ -30,7 +35,119 @@ fn invalid_heal_pool_index(pool_idx: usize, pool_count: usize) -> Error {
    )
 }

+#[derive(Debug, Clone, Copy)]
+enum HealFormatPoolSkip {
+    Completed,
+    Retryable,
+}
+
+fn classify_heal_format_pool(
+    pool_idx: usize,
+    pool_cmd_line: &str,
+    pool_meta: &PoolMeta,
+    rebalance_meta: Option<&RebalanceMeta>,
+) -> Option<HealFormatPoolSkip> {
+    let Some(pool) = pool_meta.pools.get(pool_idx) else {
+        return Some(HealFormatPoolSkip::Retryable);
+    };
+
+    if pool.id != pool_idx || pool_cmd_line.is_empty() || pool.cmd_line.is_empty() || pool.cmd_line != pool_cmd_line {
+        return Some(HealFormatPoolSkip::Retryable);
+    }
+
+    if let Some(decommission) = pool.decommission.as_ref() {
+        if decommission.complete {
+            return Some(HealFormatPoolSkip::Completed);
+        }
+        if decommission.failed || decommission.canceled || decommission.queued || pool_meta.is_suspended(pool_idx) {
+            return Some(HealFormatPoolSkip::Retryable);
+        }
+    }
+
+    if let Some(meta) = rebalance_meta {
+        let Some(pool_stats) = meta.pool_stats.get(pool_idx) else {
+            return Some(HealFormatPoolSkip::Retryable);
+        };
+        if pool_stats.info.stopping || (pool_stats.participating && pool_stats.info.status == RebalStatus::Started) {
+            return Some(HealFormatPoolSkip::Retryable);
+        }
+    }
+
+    None
+}
+
+fn heal_format_pool_skip_error(skip: HealFormatPoolSkip) -> Error {
+    match skip {
+        HealFormatPoolSkip::Completed => StorageError::NoHealRequired,
+        HealFormatPoolSkip::Retryable => StorageError::SlowDown,
+    }
+}
+
+fn heal_format_fence_lost_error() -> Error {
+    StorageError::SlowDown
+}
+
 impl ECStore {
+    async fn acquire_heal_format_fence(
+        &self,
+    ) -> Result<(NamespaceLockGuard, NamespaceLockGuard, PoolMeta, Option<RebalanceMeta>)> {
+        let metadata_pool = self
+            .pools
+            .first()
+            .cloned()
+            .ok_or_else(|| Error::other("heal format requires at least one storage pool"))?;
+
+        // Metadata fence order is part of the decommission/rebalance protocol:
+        // pool.bin must always be acquired before rebalance.bin.
+        let pool_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
+        let pool_guard = pool_lock.get_write_lock(get_lock_acquire_timeout()).await?;
+        let rebalance_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
+        let rebalance_guard = rebalance_lock.get_write_lock(get_lock_acquire_timeout()).await?;
+
+        if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
+            return Err(heal_format_fence_lost_error());
+        }
+
+        let mut pool_meta = PoolMeta::default();
+        pool_meta.load_no_lock(metadata_pool.clone()).await?;
+        if pool_meta.pools.len() != self.pools.len()
+            || pool_meta.pools.iter().enumerate().any(|(pool_idx, pool)| {
+                pool.id != pool_idx || pool.cmd_line.is_empty() || pool.cmd_line != self.pools[pool_idx].endpoints.cmd_line
+            })
+        {
+            return Err(heal_format_fence_lost_error());
+        }
+
+        let mut rebalance_meta = RebalanceMeta::new();
+        let rebalance_meta = match rebalance_meta
+            .load_with_opts(
+                metadata_pool,
+                ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+        {
+            Ok(()) => Some(rebalance_meta),
+            Err(Error::ConfigNotFound) => None,
+            Err(err) => return Err(err),
+        };
+
+        if rebalance_meta
+            .as_ref()
+            .is_some_and(|meta| meta.pool_stats.len() != self.pools.len())
+        {
+            return Err(heal_format_fence_lost_error());
+        }
+
+        if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
+            return Err(heal_format_fence_lost_error());
+        }
+
+        Ok((pool_guard, rebalance_guard, pool_meta, rebalance_meta))
+    }
+
    fn get_pools_for_heal_object(&self, opts: &HealOpts) -> Result<Vec<Arc<Sets>>> {
        match opts.pool {
            Some(pool_idx) => Ok(vec![
@@ -52,8 +169,24 @@ impl ECStore {
        };

        let mut count_no_heal = 0;
+        let mut count_completed = 0;
        let mut first_error = None;
-        for pool in self.pools.iter() {
+        for (pool_idx, pool) in self.pools.iter().enumerate() {
+            let (pool_guard, rebalance_guard, pool_meta, rebalance_meta) = self.acquire_heal_format_fence().await?;
+            if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
+                first_error.get_or_insert(heal_format_fence_lost_error());
+                break;
+            }
+            if let Some(skip) = classify_heal_format_pool(pool_idx, &pool.endpoints.cmd_line, &pool_meta, rebalance_meta.as_ref())
+            {
+                if matches!(skip, HealFormatPoolSkip::Completed) {
+                    count_completed += 1;
+                } else {
+                    first_error.get_or_insert(heal_format_pool_skip_error(skip));
+                }
+                continue;
+            }
+
            let (mut result, err) = pool.heal_format(dry_run).await?;
            if let Some(err) = err {
                match err {
@@ -69,11 +202,18 @@ impl ECStore {
            r.set_count += result.set_count;
            r.before.drives.append(&mut result.before.drives);
            r.after.drives.append(&mut result.after.drives);
+
+            // Sets::heal_format cannot observe this guard before each disk write;
+            // fail closed after the call if the lease was lost during format IO.
+            if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
+                first_error.get_or_insert(heal_format_fence_lost_error());
+                break;
+            }
        }
        if let Some(err) = first_error {
            return Ok((r, Some(err)));
        }
-        if count_no_heal == self.pools.len() {
+        if count_no_heal + count_completed == self.pools.len() {
            info!(
                event = EVENT_HEAL_FORMAT_COMPLETED,
                component = LOG_COMPONENT_ECSTORE,
@@ -300,6 +440,7 @@ mod tests {
    use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
    use crate::disk::{DiskOption, format::FormatV3, new_disk};
    use crate::layout::endpoints::{Endpoints, PoolEndpoints};
+    use crate::services::rebalance::{RebalanceInfo, RebalanceStats};
    use crate::store::init_format::{load_format_erasure, save_format_file};

    async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
@@ -347,6 +488,164 @@ mod tests {
        }
    }

+    fn pool_meta_with_decommission(info: PoolDecommissionInfo) -> PoolMeta {
+        PoolMeta {
+            pools: vec![PoolStatus {
+                id: 0,
+                cmd_line: "pool-0".to_string(),
+                last_update: OffsetDateTime::UNIX_EPOCH,
+                decommission: Some(info),
+            }],
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn heal_format_pool_state_barriers_are_classified() {
+        let active = pool_meta_with_decommission(PoolDecommissionInfo {
+            start_time: Some(OffsetDateTime::UNIX_EPOCH),
+            ..Default::default()
+        });
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &active, None),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        for info in [
+            PoolDecommissionInfo {
+                failed: true,
+                ..Default::default()
+            },
+            PoolDecommissionInfo {
+                canceled: true,
+                ..Default::default()
+            },
+        ] {
+            assert!(matches!(
+                classify_heal_format_pool(0, "pool-0", &pool_meta_with_decommission(info), None),
+                Some(HealFormatPoolSkip::Retryable)
+            ));
+        }
+
+        let completed = pool_meta_with_decommission(PoolDecommissionInfo {
+            complete: true,
+            ..Default::default()
+        });
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &completed, None),
+            Some(HealFormatPoolSkip::Completed)
+        ));
+    }
+
+    #[test]
+    fn heal_format_pool_rebalance_barriers_and_identity_are_fail_closed() {
+        let identity_meta = pool_meta_with_decommission(PoolDecommissionInfo::default());
+        let rebalance = RebalanceMeta {
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Started,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&rebalance)),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        let stopping = RebalanceMeta {
+            pool_stats: vec![RebalanceStats {
+                info: RebalanceInfo {
+                    stopping: true,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping)),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        let identity = pool_meta_with_decommission(PoolDecommissionInfo::default());
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-new", &identity, None),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        let identity_without_decommission = PoolMeta {
+            pools: vec![PoolStatus {
+                id: 0,
+                cmd_line: "pool-0".to_string(),
+                last_update: OffsetDateTime::UNIX_EPOCH,
+                decommission: None,
+            }],
+            ..Default::default()
+        };
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-new", &identity_without_decommission, None),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        assert!(matches!(
+            classify_heal_format_pool(0, "", &identity_meta, None),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &PoolMeta::default(), None),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        let stopped = RebalanceMeta {
+            stopped_at: Some(OffsetDateTime::UNIX_EPOCH),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Stopped,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        assert!(classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopped)).is_none());
+
+        let stopping_after_stop = RebalanceMeta {
+            stopped_at: Some(OffsetDateTime::UNIX_EPOCH),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Started,
+                    stopping: true,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping_after_stop)),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+    }
+
+    #[test]
+    fn skipped_heal_format_pool_is_never_reported_as_success() {
+        assert!(matches!(
+            heal_format_pool_skip_error(HealFormatPoolSkip::Retryable),
+            StorageError::SlowDown
+        ));
+        assert!(matches!(
+            heal_format_pool_skip_error(HealFormatPoolSkip::Completed),
+            StorageError::NoHealRequired
+        ));
+    }
+
    #[tokio::test]
    async fn heal_object_pool_scope_selects_only_requested_pool() {
        let store = minimal_heal_store().await;
@@ -615,6 +914,18 @@ mod tests {
            bucket_fence_registry: std::sync::Arc::default(),
        };

+        let err = store
+            .handle_heal_format(false)
+            .await
+            .expect_err("missing pool metadata must fail closed before format writes");
+        assert!(matches!(err, StorageError::SlowDown));
+
+        let pool_meta = PoolMeta::new(&store.pools, &PoolMeta::default());
+        pool_meta
+            .save(store.pools.clone())
+            .await
+            .expect("pool metadata should be persisted before format heal");
+
        let (result, err) = store
            .handle_heal_format(false)
            .await
@@ -628,5 +939,22 @@ mod tests {
            .await
            .expect("the later pool should be healed despite the first pool error");
        assert_eq!(healed.erasure.this, recoverable_format.erasure.sets[0][2]);
+
+        let mut completed_meta = PoolMeta::new(&store.pools, &PoolMeta::default());
+        for status in &mut completed_meta.pools {
+            status.decommission = Some(PoolDecommissionInfo {
+                complete: true,
+                ..Default::default()
+            });
+        }
+        completed_meta
+            .save(store.pools.clone())
+            .await
+            .expect("completed pool metadata should be persisted");
+        let (_, err) = store
+            .handle_heal_format(false)
+            .await
+            .expect("completed pools should be reported as a no-op");
+        assert!(matches!(err, Some(StorageError::NoHealRequired)));
    }
 }
@@ -231,6 +231,10 @@ impl HealTask {
                            "Heal erasure set format repair skipped because no format heal was required"
                        );
                    } else {
+                        let error = e;
+                        if error.is_recoverable_heal() {
+                            return Err(error);
+                        }
                        error!(
                            target: "rustfs::heal::task",
                            event = EVENT_HEAL_ERASURE_SET_RESULT,
@@ -239,7 +243,7 @@ impl HealTask {
                            task_id = %self.id,
                            set_disk_id,
                            result = "format_failed",
-                            error = %e,
+                            error = %error,
                            "Heal erasure set failed"
                        );
                        {
@@ -247,7 +251,7 @@ impl HealTask {
                            progress.update_progress(4, 4, 0, 0);
                        }
                        return Err(Error::TaskExecutionFailed {
-                            message: format!("Failed to heal disk format for {set_disk_id}: {e}"),
+                            message: format!("Failed to heal disk format for {set_disk_id}: {error}"),
                        });
                    }
                } else {
@@ -284,6 +288,9 @@ impl HealTask {
            Err(Error::TaskCancelled) => return Err(Error::TaskCancelled),
            Err(Error::TaskTimeout) => return Err(Error::TaskTimeout),
            Err(e) => {
+                if e.is_recoverable_heal() {
+                    return Err(e);
+                }
                error!(
                    target: "rustfs::heal::task",
                    event = EVENT_HEAL_ERASURE_SET_RESULT,
@@ -547,6 +547,7 @@ struct MockStorage {
    heal_object_outcome: Mutex<Option<MockHealObjectOutcome>>,
    heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
    format_no_heal_required: Mutex<bool>,
+    format_error: Mutex<Option<Error>>,
    global_format_calls: Mutex<u32>,
    replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
    replacement_targets_ready: Mutex<bool>,
@@ -867,6 +868,9 @@ impl HealStorageAPI for MockStorage {

    async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
        *self.global_format_calls.lock().unwrap() += 1;
+        if let Some(error) = self.format_error.lock().unwrap().take() {
+            return Err(error);
+        }
        let no_heal_required = *self.format_no_heal_required.lock().unwrap();
        if no_heal_required {
            Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::NoHealRequired))))
@@ -2052,6 +2056,30 @@ async fn test_erasure_set_heal_continues_after_format_no_heal_required() {
    );
 }

+#[tokio::test]
+async fn erasure_set_format_slowdown_is_propagated() {
+    let storage = Arc::new(MockStorage {
+        format_error: Mutex::new(Some(Error::Storage(EcstoreError::SlowDown))),
+        ..Default::default()
+    });
+    let request = HealRequest::new(
+        HealType::ErasureSet {
+            buckets: Vec::new(),
+            set_disk_id: "pool_0_set_0".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Normal,
+    );
+    let task = HealTask::from_request(request, storage);
+
+    let error = task
+        .execute()
+        .await
+        .expect_err("format SlowDown must remain recoverable for the task manager");
+
+    assert!(matches!(error, Error::Storage(EcstoreError::SlowDown)));
+}
+
 #[tokio::test]
 async fn erasure_set_bucket_prepass_failure_stops_before_object_heal() {
    let temp = TempDir::new().expect("temporary directory should be created");
@@ -69,7 +69,7 @@ uuid = { workspace = true, features = ["v4", "serde", "fast-rng", "macro-diagnos
 [dev-dependencies]
 metrics-util = { workspace = true, features = ["debugging"] }
 proptest = "1"
-serial_test = { workspace = true }
+serial_test.workspace = true
 temp-env.workspace = true
 tokio = { workspace = true, features = ["macros", "fs", "rt-multi-thread"] }

@@ -1564,6 +1564,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn abort_incomplete_multipart_upload_due_accepts_zero_days() {
        let initiated = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -1624,6 +1625,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn predict_expiration_selects_closest_expiry_for_put_object() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -1870,6 +1872,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn empty_transition_vectors_are_not_active_or_due() {
        let lc = BucketLifecycleConfiguration {
            expiry_updated_at: None,
@@ -1935,6 +1938,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_keeps_latest_object_before_days_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -1968,6 +1972,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_transitions_latest_object_after_days_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2005,6 +2010,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_transitions_latest_object_after_date_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let transition_date = base_time - Duration::days(1);
@@ -2044,6 +2050,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_selects_earliest_due_among_multiple_past_due_events() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        // Two enabled rules both yield a past-due DeleteAction and a third yields a
@@ -2157,6 +2164,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_expires_noncurrent_version_after_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2194,6 +2202,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_skips_noncurrent_expiration_without_successor() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).expect("valid fixed test timestamp");
        let lc = BucketLifecycleConfiguration {
@@ -2229,6 +2238,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_missing_successor_does_not_skip_noncurrent_transition() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).expect("valid fixed test timestamp");
        let lc = BucketLifecycleConfiguration {
@@ -2271,6 +2281,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_noncurrent_expiration_one_day_respects_due_boundary() {
        let successor_time = datetime!(2025-06-15 12:00:00 UTC);
        let due = expected_expiry_time(successor_time, 1);
@@ -2312,6 +2323,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_expires_noncurrent_version_immediately_when_zero_days() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2349,6 +2361,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_transitions_noncurrent_version_after_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2424,6 +2437,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn evaluator_honors_newer_noncurrent_versions_retention_count() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = Arc::new(BucketLifecycleConfiguration {
@@ -2712,6 +2726,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn expired_object_delete_marker_ignores_marker_with_noncurrent_versions_present() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2788,6 +2803,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn expired_object_delete_marker_deletes_only_delete_marker_immediately() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2865,6 +2881,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn expiration_days_deletes_only_expired_delete_marker_when_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2915,6 +2932,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn expiration_days_uses_earliest_due_rule_for_expired_delete_marker() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let make_rule = |id: &str, days| LifecycleRule {
@@ -3245,6 +3263,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn del_marker_expiration_deletes_marker_and_older_versions_when_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).expect("fixed timestamp should be valid");
        let lc = BucketLifecycleConfiguration {
@@ -3284,6 +3303,7 @@ mod tests {
    // --- TASK-003 tests: Round up to next UTC processing boundary ---

    #[test]
+    #[serial]
    fn expected_expiry_time_rounds_up_to_next_midnight_utc() {
        with_default_ilm_process_time(|| {
            // Object created at 2025-01-15T10:30:45Z, expire in 30 days
@@ -3299,6 +3319,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn expected_expiry_time_immediate_expiry_returns_epoch() {
        with_default_ilm_process_time(|| {
            let mod_time = datetime!(2025-06-01 12:00:00 UTC);
@@ -3308,6 +3329,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn expected_expiry_time_preserves_exact_midnight_boundary() {
        with_default_ilm_process_time(|| {
            let mod_time = datetime!(2025-03-01 00:00:00 UTC);
@@ -3317,6 +3339,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn expected_expiry_time_rounds_end_of_day_to_following_midnight() {
        with_default_ilm_process_time(|| {
            let mod_time = datetime!(2025-06-15 23:59:59 UTC);
@@ -3326,6 +3349,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn expected_expiry_time_uses_canonical_process_time_boundary() {
        let mod_time = datetime!(2025-01-15 10:30:45 UTC);

@@ -3338,6 +3362,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn expected_expiry_time_uses_deprecated_process_time_alias() {
        let mod_time = datetime!(2025-01-15 10:30:45 UTC);

@@ -3350,6 +3375,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn expected_expiry_time_uses_default_boundary_when_process_time_is_zero_or_invalid() {
        let mod_time = datetime!(2025-01-15 10:30:45 UTC);

@@ -3372,6 +3398,7 @@ mod tests {

    // (a) Default path (env unset) is byte-identical: one day == 86400s.
    #[test]
+    #[serial]
    fn ilm_day_secs_defaults_to_86400_when_unset() {
        temp_env::with_var_unset(ENV_ILM_DEBUG_DAY_SECS, || {
            assert_eq!(ilm_day_secs(), DEFAULT_ILM_DAY_SECS);
@@ -3400,6 +3427,7 @@ mod tests {

    // (b) End-to-end env read scales the day length.
    #[test]
+    #[serial]
    fn ilm_day_secs_scales_when_env_set() {
        temp_env::with_var(ENV_ILM_DEBUG_DAY_SECS, Some("2"), || {
            assert_eq!(ilm_day_secs(), 2);
@@ -3408,6 +3436,7 @@ mod tests {

    // (c) Invalid env value falls back to 86400.
    #[test]
+    #[serial]
    fn ilm_day_secs_falls_back_on_invalid_env() {
        temp_env::with_var(ENV_ILM_DEBUG_DAY_SECS, Some("bogus"), || {
            assert_eq!(ilm_day_secs(), DEFAULT_ILM_DAY_SECS);
@@ -3420,6 +3449,7 @@ mod tests {
    // Deadline math scales: with a 1s day and PROCESS_TIME unset, a Days=1 rule is
    // due 1s after mod_time (rounded up to the next 1s boundary => same instant).
    #[test]
+    #[serial]
    fn expected_expiry_time_scales_with_debug_day_secs() {
        let mod_time = datetime!(2025-01-15 10:30:45 UTC);
        temp_env::with_var(ENV_ILM_DEBUG_DAY_SECS, Some("1"), || {
@@ -3435,6 +3465,7 @@ mod tests {

    // days == 0 still yields the immediate-expiry sentinel regardless of the switch.
    #[test]
+    #[serial]
    fn expected_expiry_time_zero_days_ignores_debug_day_secs() {
        let mod_time = datetime!(2025-06-01 12:00:00 UTC);
        temp_env::with_var(ENV_ILM_DEBUG_DAY_SECS, Some("2"), || {
@@ -3445,6 +3476,7 @@ mod tests {
    // (③) Interaction with an explicit RUSTFS_ILM_PROCESS_TIME: the deadline offset
    // uses the accelerated day length, but the rounding boundary honors PROCESS_TIME.
    #[test]
+    #[serial]
    fn expected_expiry_time_debug_day_secs_respects_explicit_process_time() {
        let mod_time = datetime!(2025-01-15 10:30:00 UTC);
        // day == 10s, but round up to the next 60s (PROCESS_TIME) boundary.
@@ -3461,6 +3493,7 @@ mod tests {

    // (③) With the switch unset, an explicit PROCESS_TIME behaves exactly as before.
    #[test]
+    #[serial]
    fn expected_expiry_time_unset_debug_day_secs_matches_legacy_process_time() {
        let mod_time = datetime!(2025-01-15 10:30:45 UTC);
        temp_env::with_var_unset(ENV_ILM_DEBUG_DAY_SECS, || {
@@ -3488,6 +3521,7 @@ mod tests {

    // The abort-incomplete-multipart deadline path also scales through the switch.
    #[test]
+    #[serial]
    fn abort_incomplete_multipart_due_scales_with_debug_day_secs() {
        use s3s::dto::AbortIncompleteMultipartUpload;
        let initiated = datetime!(2025-01-15 10:30:45 UTC);
@@ -3532,6 +3566,7 @@ mod tests {
    // (⑤ evaluator seam) A Days=1 rule fires under RUSTFS_ILM_DEBUG_DAY_SECS=1 once
    // `now` advances a few seconds past a mod_time only ~seconds in the past.
    #[test]
+    #[serial]
    fn eval_inner_expires_days_one_rule_under_debug_day_secs() {
        let lc = BucketLifecycleConfiguration {
            expiry_updated_at: None,
@@ -3580,6 +3615,7 @@ mod tests {

    // Absolute Date-based rules must NOT scale with the switch (regression guard).
    #[test]
+    #[serial]
    fn eval_inner_date_rule_ignores_debug_day_secs() {
        let expiry_date = datetime!(2025-06-01 00:00:00 UTC);
        let lc = BucketLifecycleConfiguration {
@@ -3837,6 +3873,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_triggers_delete_all_versions_when_expired_object_all_versions_set() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -3875,6 +3912,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn expired_object_all_versions_does_not_apply_to_current_delete_marker() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).expect("fixed timestamp should be valid");
        let lc = BucketLifecycleConfiguration {
@@ -3904,6 +3942,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn eval_inner_uses_delete_action_when_all_versions_not_set() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -4022,6 +4061,7 @@ mod tests {
        use super::*;
        use proptest::prelude::*;
        use s3s::dto::{NoncurrentVersionExpiration, Tag};
+        use serial_test::serial;

        const DAY_SECS: i64 = 86400;

@@ -4252,6 +4292,7 @@ mod tests {
            /// combination, and must be deterministic: the same input
            /// evaluated twice yields an identical event.
            #[test]
+            #[serial]
            fn eval_inner_never_panics_and_is_deterministic(
                rules in prop::collection::vec(arb_rule(), 0..4),
                obj in arb_object_opts(),
@@ -4391,6 +4432,7 @@ mod tests {
            /// candidate set — earliest due wins, ties prefer delete-class —
            /// and must be `NoneAction` exactly when that set is empty.
            #[test]
+            #[serial]
            fn eval_inner_winner_matches_selection_oracle(
                rules in prop::collection::vec(arb_selection_rule(), 0..5),
                mod_off in 0i64..(2 * DAY_SECS),
@@ -4444,6 +4486,7 @@ mod tests {
            /// non-decreasing in `days` (days == 0 maps to UNIX_EPOCH, below
            /// any post-1970 deadline).
            #[test]
+            #[serial]
            fn expected_expiry_time_is_monotonic_in_days(
                mod_off in 0i64..(3650 * DAY_SECS),
                d1 in 0i32..2000,
@@ -4465,6 +4508,7 @@ mod tests {
            /// to the next whole-day boundary: the result is day-aligned, not
            /// before `mod_time + days`, and less than one boundary beyond it.
            #[test]
+            #[serial]
            fn expected_expiry_time_lands_on_default_day_boundary(
                mod_off in 0i64..(3650 * DAY_SECS),
                days in 1i32..2000,
@@ -4482,6 +4526,7 @@ mod tests {
            /// to that boundary instead: aligned to it, never early, and less
            /// than one boundary late.
            #[test]
+            #[serial]
            fn expected_expiry_time_lands_on_explicit_process_boundary(
                mod_off in 0i64..(365 * DAY_SECS),
                days in 1i32..400,
@@ -73,6 +73,7 @@ walkdir = { workspace = true }

 [dev-dependencies]
 criterion = { workspace = true, features = ["html_reports"] }
+serial_test = { workspace = true }
 temp-env = { workspace = true, features = ["async_closure"] }
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["test-util", "macros", "fs", "rt-multi-thread"] }
@@ -1484,6 +1484,7 @@ mod tests {
        ENV_CAPACITY_SAMPLE_RATE, ENV_CAPACITY_STAT_TIMEOUT, ENV_CAPACITY_WRITE_FREQUENCY_THRESHOLD,
        ENV_CAPACITY_WRITE_TRIGGER_DELAY,
    };
+    use serial_test::serial;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

@@ -1668,6 +1669,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn test_config_getter_defaults() {
        for (env_var, getter, default, _, _) in config_getter_cases() {
            temp_env::with_var(env_var, None::<&str>, || {
@@ -1677,6 +1679,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn test_config_getter_env_overrides() {
        for (env_var, getter, _, override_value, expected) in config_getter_cases() {
            temp_env::with_var(env_var, Some(override_value), || {
@@ -1686,6 +1689,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn test_zero_env_values_clamp_to_defaults() {
        // A zero threshold makes small disks report 0 bytes; a zero timeout
        // (with dynamic timeout off) makes every scan fail. Both must fall
@@ -1705,6 +1709,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_update_capacity_preserves_retrieval_metadata() {
        let manager = HybridCapacityManager::from_env();

@@ -1720,6 +1725,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_record_write_operation() {
        let manager = HybridCapacityManager::from_env();

@@ -1730,6 +1736,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_write_frequency_window() {
        let manager = HybridCapacityManager::from_env();

@@ -1817,6 +1824,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn test_recent_write_count_ignores_future_buckets() {
        let record = WriteRecord::new();
        record.write_buckets[0].store(120, 3);
@@ -1830,6 +1838,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_needs_fast_update() {
        let manager = HybridCapacityManager::from_env();

@@ -1846,6 +1855,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_cache_age_tracking() {
        let manager = HybridCapacityManager::from_env();

@@ -1865,6 +1875,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_data_source_tracking() {
        let manager = HybridCapacityManager::from_env();

@@ -1880,6 +1891,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_needs_fast_update_waits_for_write_trigger_delay() {
        let manager = create_isolated_manager(HybridStrategyConfig {
            scheduled_update_interval: Duration::from_secs(60),
@@ -1910,6 +1922,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_needs_fast_update_respects_enable_write_trigger() {
        let manager = create_isolated_manager(HybridStrategyConfig {
            scheduled_update_interval: Duration::from_secs(60),
@@ -1936,6 +1949,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_concurrent_access() {
        let manager = Arc::new(HybridCapacityManager::from_env());
        let mut handles = Vec::new();
@@ -1962,6 +1976,7 @@ mod tests {
    // exact under heavy same-second contention or the frequency window (and the
    // write-trigger decision) would undercount.
    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+    #[serial]
    async fn test_record_write_operation_lock_free_is_exact_under_contention() {
        let manager = Arc::new(HybridCapacityManager::from_env());
        let mut handles = Vec::new();
@@ -1986,6 +2001,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_performance_overhead() {
        let manager = Arc::new(HybridCapacityManager::from_env());
        let start = Instant::now();
@@ -2002,6 +2018,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_refresh_or_join_singleflight() {
        let manager = Arc::new(HybridCapacityManager::from_env());
        let calls = Arc::new(AtomicUsize::new(0));
@@ -2041,6 +2058,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_refresh_or_join_recovers_after_leader_cancellation() {
        let manager = Arc::new(HybridCapacityManager::from_env());

@@ -2069,6 +2087,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_refresh_or_join_cancelled_leader_unblocks_joiner() {
        let manager = Arc::new(HybridCapacityManager::from_env());

@@ -2096,6 +2115,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_spawn_refresh_if_needed_deduplicates_background_refresh() {
        let manager = Arc::new(HybridCapacityManager::from_env());
        let calls = Arc::new(AtomicUsize::new(0));
@@ -2133,6 +2153,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_record_write_operation_with_scope_token_marks_dirty_disks() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        let token = uuid::Uuid::new_v4();
@@ -2156,6 +2177,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_get_dirty_disks_drains_global_dirty_scope_registry() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        record_global_dirty_scope(CapacityScope {
@@ -2175,6 +2197,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_update_capacity_recomputes_total_from_disk_cache_for_subset_refresh() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2285,6 +2308,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_update_capacity_degraded_full_refresh_merges_cache_and_does_not_oscillate() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2330,6 +2354,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_update_capacity_degraded_with_empty_per_disk_serves_merged_cache() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        manager.update_capacity(full_two_disk_update(), DataSource::RealTime).await;
@@ -2359,6 +2384,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_update_capacity_degraded_without_complete_cache_keeps_partial_sum() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2399,6 +2425,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_commit_keeps_dirty_marks_recorded_after_scan_start() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        let disk = scope_disk("node-a", "/tmp/disk-a");
@@ -2429,6 +2456,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_commit_clears_dirty_marks_recorded_before_scan_start() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        let disk = scope_disk("node-a", "/tmp/disk-a");
@@ -2449,6 +2477,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_retain_dirty_disks_within_drops_ghost_entries() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        let local = scope_disk("node-a", "/tmp/disk-a");
@@ -2467,6 +2496,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_spawn_refresh_recovers_from_construction_panic() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2533,6 +2563,7 @@ mod tests {
    }

    #[tokio::test(start_paused = true)]
+    #[serial]
    async fn test_refresh_or_join_joiner_times_out_when_leader_wedges() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2560,6 +2591,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_refresh_or_join_returns_cluster_total_for_dirty_subset() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2641,6 +2673,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_config_from_env() {
        let config = HybridStrategyConfig::from_env();

@@ -2654,6 +2687,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_config_from_env_with_override() {
        temp_env::with_var(ENV_CAPACITY_SCHEDULED_INTERVAL, Some("600"), || {
            let config = HybridStrategyConfig::from_env();
@@ -1069,6 +1069,7 @@ mod tests {
    #[cfg(unix)]
    use rustfs_config::ENV_CAPACITY_FOLLOW_SYMLINKS;
    use rustfs_config::{ENV_CAPACITY_MAX_FILES_THRESHOLD, ENV_CAPACITY_SAMPLE_RATE};
+    use serial_test::serial;

    /// Reference implementation using unbounded `u128` arithmetic, clamped to
    /// `u64::MAX`, used as the source of truth for the sampling extrapolation.
@@ -1273,6 +1274,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_get_dir_size_async_nonexistent_directory() {
        let result = get_dir_size_async(Path::new("/nonexistent/path")).await;
        assert!(result.is_err());
@@ -1646,6 +1648,7 @@ mod tests {
    }

    #[tokio::test]
+    #[serial]
    async fn test_metadata_incomplete_aggregate_does_not_replace_disk_cache() {
        use std::fs::File;
        use std::io::Write;
@@ -1780,6 +1783,7 @@ mod tests {

    #[cfg(unix)]
    #[tokio::test]
+    #[serial]
    async fn test_get_dir_size_async_ignores_symlink_targets_when_follow_disabled() {
        use std::fs::File;
        use std::io::Write;
@@ -1805,6 +1809,7 @@ mod tests {

    #[cfg(unix)]
    #[tokio::test]
+    #[serial]
    async fn test_get_dir_size_async_counts_symlink_targets_when_follow_enabled() {
        use std::fs::File;
        use std::io::Write;
@@ -75,7 +75,7 @@ pub struct BucketReplicationBandwidthStats {
 }

 #[derive(Debug, Clone, Default)]
-pub struct BucketReplicationMetricsSnapshot {
+pub struct BucketReplicationStats {
    pub bucket: String,
    pub total_failed_bytes: u64,
    pub total_failed_count: u64,
@@ -107,7 +107,7 @@ pub struct BucketReplicationMetricsSnapshot {

 #[derive(Debug, Clone, Default)]
 pub(crate) struct BucketReplicationRuntimeStats {
-    pub(crate) stats: BucketReplicationMetricsSnapshot,
+    pub(crate) stats: BucketReplicationStats,
    pub(crate) target_flows: Vec<BucketReplicationTargetFlowStats>,
 }

@@ -182,7 +182,7 @@ fn push_proxy_request_result_metrics(
    }
 }

-pub fn collect_bucket_replication_metrics(stats: &[BucketReplicationMetricsSnapshot]) -> Vec<PrometheusMetric> {
+pub fn collect_bucket_replication_metrics(stats: &[BucketReplicationStats]) -> Vec<PrometheusMetric> {
    if stats.is_empty() {
        return Vec::new();
    }
@@ -572,7 +572,7 @@ mod tests {
    #[test]
    fn test_collect_bucket_replication_metrics() {
        let stats = vec![BucketReplicationRuntimeStats {
-            stats: BucketReplicationMetricsSnapshot {
+            stats: BucketReplicationStats {
                bucket: "b1".to_string(),
                total_failed_bytes: 64,
                total_failed_count: 2,
@@ -876,7 +876,7 @@ mod tests {

    #[test]
    fn test_collect_bucket_replication_metrics_empty() {
-        let stats: Vec<BucketReplicationMetricsSnapshot> = Vec::new();
+        let stats: Vec<BucketReplicationStats> = Vec::new();
        let metrics = collect_bucket_replication_metrics(&stats);
        assert!(metrics.is_empty());
    }
@@ -48,7 +48,7 @@ pub(crate) use bucket_replication::{
    BucketReplicationTargetFlowStats, collect_bucket_replication_backlog_metrics, collect_bucket_replication_runtime_metrics,
 };
 pub use bucket_replication::{
-    BucketReplicationBandwidthStats, BucketReplicationMetricsSnapshot, BucketReplicationTargetStats,
+    BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats,
    collect_bucket_replication_bandwidth_metrics, collect_bucket_replication_metrics,
 };
 pub use cluster::{ClusterStats, collect_cluster_metrics};
@@ -69,7 +69,7 @@ pub use notification::{NotificationStats, collect_notification_metrics};
 pub(crate) use notification_target::{NotificationTargetRuntimeStats, collect_notification_target_runtime_metrics};
 pub use notification_target::{NotificationTargetStats, collect_notification_target_metrics};
 pub(crate) use replication::{ReplicationRuntimeStats, collect_replication_runtime_metrics};
-pub use replication::{ReplicationMetricsSnapshot, collect_replication_metrics};
+pub use replication::{ReplicationStats, collect_replication_metrics};
 pub(crate) use request::{ApiRequestMetricSupport, ApiRequestStats, collect_request_metrics};
 pub use resource::{ResourceStats, collect_resource_metrics};
 pub(crate) use scanner::{ScannerRuntimeStats, collect_scanner_runtime_metrics};
@@ -22,7 +22,7 @@ use crate::metrics::schema::replication::*;

 /// Replication statistics.
 #[derive(Debug, Clone, Default)]
-pub struct ReplicationMetricsSnapshot {
+pub struct ReplicationStats {
    /// Average number of active replication workers
    pub average_active_workers: f64,
    /// Average queued bytes since server start
@@ -54,13 +54,13 @@ pub struct ReplicationMetricsSnapshot {
 #[derive(Debug, Clone, Default)]
 pub(crate) struct ReplicationRuntimeStats {
    pub(crate) server: String,
-    pub(crate) stats: ReplicationMetricsSnapshot,
+    pub(crate) stats: ReplicationStats,
 }

 /// Collects replication metrics from the given stats.
 ///
 /// Returns a vector of Prometheus metrics for replication statistics.
-pub fn collect_replication_metrics(stats: &ReplicationMetricsSnapshot) -> Vec<PrometheusMetric> {
+pub fn collect_replication_metrics(stats: &ReplicationStats) -> Vec<PrometheusMetric> {
    vec![
        PrometheusMetric::from_descriptor(&REPLICATION_AVERAGE_ACTIVE_WORKERS_MD, stats.average_active_workers),
        PrometheusMetric::from_descriptor(&REPLICATION_AVERAGE_QUEUED_BYTES_MD, stats.average_queued_bytes as f64),
@@ -120,7 +120,7 @@ mod tests {

    #[test]
    fn test_collect_replication_metrics() {
-        let stats = ReplicationMetricsSnapshot {
+        let stats = ReplicationStats {
            average_active_workers: 8.5,
            average_queued_bytes: 1024 * 1024 * 40,
            average_queued_count: 240,
@@ -182,7 +182,7 @@ mod tests {

    #[test]
    fn test_collect_replication_metrics_default() {
-        let stats = ReplicationMetricsSnapshot::default();
+        let stats = ReplicationStats::default();
        let metrics = collect_replication_metrics(&stats);

        assert_eq!(metrics.len(), 13);
@@ -194,7 +194,7 @@ mod tests {

    #[test]
    fn replication_stats_struct_literal_keeps_legacy_fields() {
-        let stats = ReplicationMetricsSnapshot {
+        let stats = ReplicationStats {
            average_active_workers: 1.0,
            average_queued_bytes: 2,
            average_queued_count: 3,
@@ -2811,14 +2811,14 @@ mod tests {
    #[test]
    fn replication_proxy_bucket_keys_detect_removed_buckets() {
        let previous = repl_proxy_bucket_live_keys(&[BucketReplicationRuntimeStats {
-            stats: crate::metrics::BucketReplicationMetricsSnapshot {
+            stats: crate::metrics::BucketReplicationStats {
                bucket: "photos".to_string(),
                ..Default::default()
            },
            ..Default::default()
        }]);
        let current = repl_proxy_bucket_live_keys(&[BucketReplicationRuntimeStats {
-            stats: crate::metrics::BucketReplicationMetricsSnapshot {
+            stats: crate::metrics::BucketReplicationStats {
                bucket: "logs".to_string(),
                ..Default::default()
            },
@@ -21,12 +21,12 @@
 use crate::metrics::collectors::scanner::{ScannerActiveBucketDriveStats, ScannerBucketDriveResultStats, ScannerSourceWorkStats};
 use crate::metrics::collectors::{
    ApiRequestMetricSupport, ApiRequestStats, BucketReplicationBacklogStats, BucketReplicationBandwidthStats,
-    BucketReplicationRuntimeStats, BucketReplicationMetricsSnapshot, BucketReplicationTargetBacklogStats, BucketReplicationTargetFlowStats,
+    BucketReplicationRuntimeStats, BucketReplicationStats, BucketReplicationTargetBacklogStats, BucketReplicationTargetFlowStats,
    BucketReplicationTargetStats, BucketStats, BucketUsageStats, ClusterConfigStats, ClusterHealthStats, ClusterStats,
    ClusterUsageStats, CompressionClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats,
    DriveRuntimeDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmActionTaskStats, IlmBackpressureStats,
    IlmQueueTaskStats, IlmRuntimeStats, IlmStats, IlmTaskEventStats, MemoryStats, NetworkStats, ProcessStats, ProcessStatusType,
-    ReplicationMetricsSnapshot, ResourceStats, ScannerRuntimeStats, ScannerStats,
+    ReplicationStats, ResourceStats, ScannerRuntimeStats, ScannerStats,
 };
 use crate::metrics::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot};
 use crate::metrics::{
@@ -266,7 +266,7 @@ fn bucket_replication_detail_from_snapshot(stats: ObsBucketReplicationStatsSnaps

    BucketReplicationRuntimeStats {
        target_flows,
-        stats: BucketReplicationMetricsSnapshot {
+        stats: BucketReplicationStats {
            bucket,
            total_failed_bytes: stats.total_failed_bytes,
            total_failed_count: stats.total_failed_count,
@@ -298,7 +298,7 @@ fn bucket_replication_detail_from_snapshot(stats: ObsBucketReplicationStatsSnaps
    }
 }

-async fn obs_site_replication_stats() -> ReplicationMetricsSnapshot {
+async fn obs_site_replication_stats() -> ReplicationStats {
    let current_data_transfer_rate = obs_bucket_replication_bandwidth_stats()
        .into_iter()
        .flatten()
@@ -306,7 +306,7 @@ async fn obs_site_replication_stats() -> ReplicationMetricsSnapshot {
        .sum::<f64>();
    let stats = obs_replication_site_stats_snapshot(current_data_transfer_rate).await;

-    ReplicationMetricsSnapshot {
+    ReplicationStats {
        average_active_workers: stats.average_active_workers,
        average_queued_bytes: stats.average_queued_bytes,
        average_queued_count: stats.average_queued_count,
@@ -648,7 +648,7 @@ pub fn collect_bucket_replication_bandwidth_stats() -> Vec<BucketReplicationBand
 }

 /// Collect bucket and target level replication stats from the global replication runtime.
-pub async fn collect_bucket_replication_detail_stats() -> Vec<BucketReplicationMetricsSnapshot> {
+pub async fn collect_bucket_replication_detail_stats() -> Vec<BucketReplicationStats> {
    obs_bucket_replication_stats_snapshot()
        .await
        .into_iter()
@@ -662,7 +662,7 @@ pub(crate) async fn collect_bucket_replication_stats_bundle()
 }

 /// Collect site-level replication stats from the global replication runtime.
-pub async fn collect_replication_stats() -> ReplicationMetricsSnapshot {
+pub async fn collect_replication_stats() -> ReplicationStats {
    obs_site_replication_stats().await
 }

@@ -103,7 +103,8 @@ hex-simd.workspace = true

 [dev-dependencies]
 tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] }
-temp-env = { workspace = true, features = ["async_closure"] }
+serial_test = { workspace = true }
+temp-env = { workspace = true }
 tempfile = { workspace = true }
 uuid = { workspace = true, features = ["v4", "serde", "fast-rng", "macro-diagnostics"] }
 tokio = { workspace = true, features = ["test-util", "fs", "rt-multi-thread"] }
@@ -599,8 +599,10 @@ impl ScannerConfigObjectDelete for ECStore {
 #[cfg(test)]
 mod tests {
    use super::*;
+    use serial_test::serial;

    #[tokio::test]
+    #[serial]
    async fn runtime_tier_names_serves_cached_arc_within_ttl() {
        reset_tier_name_cache_for_test();
        // The tier config manager is unconfigured in unit tests, so the
@@ -614,6 +616,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn foreground_read_guard_tracks_stream_lifetime() {
        reset_foreground_read_activity_for_test();
        assert_eq!(current_foreground_read_activity(), 0);
@@ -627,6 +630,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn foreground_read_activity_keeps_larger_signal() {
        reset_foreground_read_activity_for_test();
        let _guard = ForegroundReadGuard::new();
@@ -639,6 +643,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_guard_tracks_runtime_lifetime() {
        reset_scanner_runtime_instances_for_test();
        assert!(!scanner_runtime_initialized());
@@ -868,6 +868,7 @@ mod tests {
        SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE,
        SCANNER_SPEED, SCANNER_SUB_SYS, ScannerSpeed,
    };
+    use serial_test::serial;
    use std::collections::HashMap;
    use std::time::Duration;
    use temp_env::{with_var, with_var_unset};
@@ -915,6 +916,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_uses_persisted_values_when_env_is_unset() {
        let config = server_config_with_scanner(&[
            (SCANNER_SPEED, "slow"),
@@ -942,6 +944,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_normalizes_persisted_default_speed() {
        let config = server_config_with_scanner(&[(SCANNER_SPEED, "default")]);

@@ -957,6 +960,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_prefers_env_over_persisted_config() {
        let config = server_config_with_scanner(&[(SCANNER_SPEED, "slowest"), (SCANNER_CYCLE, "600")]);

@@ -973,6 +977,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_prefers_heal_bitrot_cycle_over_scanner_compat_config() {
        let config = server_config_with_scanner_and_heal(&[(SCANNER_BITROT_CYCLE, "3600")], &[(HEAL_BITROT_CYCLE, "off")]);

@@ -985,6 +990,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_marks_scanner_bitrot_cycle_as_compat_source() {
        let config = server_config_with_scanner(&[(SCANNER_BITROT_CYCLE, "3600")]);

@@ -1001,6 +1007,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_normalizes_persisted_default_bitrot_cycles() {
        let default_cycle = DEFAULT_HEAL_BITROT_CYCLE_SECS.to_string();
        for config in [
@@ -1025,6 +1032,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_validation_rejects_invalid_persisted_speed_with_env_override() {
        let config = server_config_with_scanner(&[(SCANNER_SPEED, "warp")]);

@@ -1058,6 +1066,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_uses_derived_delay_for_excessive_env_override() {
        let config = server_config_with_scanner(&[(SCANNER_SPEED, "slow")]);

@@ -1078,6 +1087,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_status_reports_value_sources() {
        let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_OBJECTS, "100"), (SCANNER_CACHE_SAVE_TIMEOUT, "5")]);

@@ -1098,6 +1108,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn applied_runtime_config_is_the_authoritative_scheduler_state() {
        let config = server_config_with_scanner(&[(SCANNER_CYCLE, "321")]);

@@ -1114,6 +1125,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_status_reports_persisted_pacing_overrides() {
        let config = server_config_with_scanner(&[("delay", "3.5"), ("max_wait", "7")]);

@@ -1135,6 +1147,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_status_prefers_env_pacing_overrides() {
        let config = server_config_with_scanner(&[("delay", "3.5"), ("max_wait", "7")]);

@@ -1156,6 +1169,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn scanner_runtime_config_status_preserves_subsecond_max_wait() {
        let config = server_config_with_scanner(&[(SCANNER_SPEED, "fast")]);

@@ -19,6 +19,7 @@ use crate::{
    ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader,
    init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx,
 };
+use serial_test::serial;
 use std::collections::HashMap;
 use std::io::Cursor;
 use std::task::Poll;
@@ -361,6 +362,7 @@ fn test_initial_scanner_delay_uses_configured_start_delay() {
 }

 #[test]
+#[serial]
 fn test_initial_scanner_delay_uses_cycle_without_explicit_start_delay() {
    with_var(ENV_SCANNER_CYCLE, Some("120"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -407,6 +409,7 @@ fn test_initial_scanner_delay_keeps_delay_for_replication_without_buckets() {
 }

 #[test]
+#[serial]
 fn test_scanner_cycle_max_duration_uses_env() {
    with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("42"), || {
        assert_eq!(scanner_cycle_max_duration(), Some(Duration::from_secs(42)));
@@ -414,6 +417,7 @@ fn test_scanner_cycle_max_duration_uses_env() {
 }

 #[test]
+#[serial]
 fn test_scanner_cycle_max_duration_default_is_disabled() {
    with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || {
        assert_eq!(scanner_cycle_max_duration(), None);
@@ -457,6 +461,7 @@ async fn test_scanner_cycle_budget_drop_cancels_child_without_elapsed() {
 }

 #[test]
+#[serial]
 fn test_scanner_cycle_budget_config_uses_work_budget_env() {
    with_var(ENV_SCANNER_CYCLE_MAX_OBJECTS, Some("100"), || {
        with_var(ENV_SCANNER_CYCLE_MAX_DIRECTORIES, Some("25"), || {
@@ -468,6 +473,7 @@ fn test_scanner_cycle_budget_config_uses_work_budget_env() {
 }

 #[test]
+#[serial]
 fn test_scanner_cycle_budget_config_disables_zero_work_budgets() {
    with_var(ENV_SCANNER_CYCLE_MAX_OBJECTS, Some("0"), || {
        with_var(ENV_SCANNER_CYCLE_MAX_DIRECTORIES, Some("0"), || {
@@ -510,6 +516,7 @@ fn test_scan_cycle_partial_source_maps_budget_reason() {
 }

 #[tokio::test]
+#[serial]
 async fn test_mark_scan_cycle_idle_clears_published_cycle_state() {
    let mut cycle_info = CurrentCycle {
        current: 12,
@@ -538,6 +545,7 @@ async fn test_mark_scan_cycle_idle_clears_published_cycle_state() {
 }

 #[tokio::test]
+#[serial]
 async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() {
    let cycle_started = Utc::now() - chrono::Duration::seconds(5);
    let mut cycle_info = CurrentCycle {
@@ -564,6 +572,7 @@ async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() {
 }

 #[tokio::test]
+#[serial]
 async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finalization() {
    let mut cycle_info = CurrentCycle {
        current: 12,
@@ -588,6 +597,7 @@ async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finaliz
 }

 #[tokio::test]
+#[serial]
 async fn scanner_cycle_metrics_guard_drop_clears_activity() {
    let guard = ScannerCycleMetricsGuard::new(CurrentCycle {
        current: 12,
@@ -605,6 +615,7 @@ async fn scanner_cycle_metrics_guard_drop_clears_activity() {
 }

 #[tokio::test]
+#[serial]
 async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() {
    let (_temp_dir, store) = setup_scanner_cycle_store().await;
    let ctx = CancellationToken::new();
@@ -655,6 +666,7 @@ async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() {
 }

 #[tokio::test]
+#[serial]
 async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -690,6 +702,7 @@ async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() {
 }

 #[tokio::test]
+#[serial]
 async fn scanner_cycle_recovers_to_newer_durable_cache_floor() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -729,6 +742,7 @@ async fn scanner_cycle_recovers_to_newer_durable_cache_floor() {
 }

 #[tokio::test]
+#[serial]
 async fn scanner_cycle_rejects_invalid_cache_floor() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -999,6 +1013,7 @@ async fn scanner_usage_floor_fails_closed_on_corrupt_or_exhausted_usage_state()
 }

 #[tokio::test]
+#[serial]
 async fn scanner_usage_backup_uses_durable_cycle_cadence_across_tasks() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -1072,6 +1087,7 @@ fn scanner_cycle_advance_fails_before_reserved_exhausted_value() {
 }

 #[tokio::test]
+#[serial]
 async fn test_finalize_partial_scan_cycle_reports_persist_failure() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -1095,6 +1111,7 @@ async fn test_finalize_partial_scan_cycle_reports_persist_failure() {
 }

 #[tokio::test]
+#[serial]
 async fn test_persist_scanner_cycle_state_reconciles_newer_winner() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -1517,6 +1534,7 @@ async fn test_usage_save_route_barrier_prevents_missing_snapshot_creation() {
 }

 #[tokio::test]
+#[serial]
 async fn test_usage_route_barrier_precedes_durable_reconciliation() {
    let store = Arc::new(MemoryConfigStore::default());
    let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
@@ -1544,6 +1562,7 @@ async fn test_usage_route_barrier_precedes_durable_reconciliation() {
 }

 #[tokio::test]
+#[serial]
 async fn test_deferred_usage_save_keeps_last_real_save_metric() {
    let metrics = global_metrics();
    metrics.record_scanner_usage_save_result(ScannerUsageSaveResult::Success);
@@ -2607,6 +2626,7 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
 }

 #[test]
+#[serial]
 fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
    crate::scanner_io::clear_dirty_usage_bucket("photos");
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -2633,6 +2653,7 @@ fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
 }

 #[test]
+#[serial]
 fn finalizing_a_deferred_usage_save_keeps_dirty_work_pending() {
    crate::scanner_io::clear_dirty_usage_bucket("photos");
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -2675,6 +2696,7 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
 }

 #[test]
+#[serial]
 fn finalizing_an_already_durable_cycle_acknowledges_its_exact_dirty_snapshot() {
    crate::scanner_io::clear_dirty_usage_bucket("photos");
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -2689,6 +2711,7 @@ fn finalizing_an_already_durable_cycle_acknowledges_its_exact_dirty_snapshot() {
 }

 #[test]
+#[serial]
 fn finalizing_a_prior_same_cycle_snapshot_keeps_new_dirty_work_pending() {
    crate::scanner_io::clear_dirty_usage_bucket("photos");
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -2704,6 +2727,7 @@ fn finalizing_a_prior_same_cycle_snapshot_keeps_new_dirty_work_pending() {
 }

 #[test]
+#[serial]
 fn finalizing_a_durable_superseded_snapshot_keeps_dirty_work_pending() {
    crate::scanner_io::clear_dirty_usage_bucket("photos");
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -2719,6 +2743,7 @@ fn finalizing_a_durable_superseded_snapshot_keeps_dirty_work_pending() {
 }

 #[test]
+#[serial]
 fn data_usage_persist_wait_covers_cache_retries_and_backup() {
    with_var(rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, Some("7"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -2771,6 +2796,7 @@ async fn maintenance_feature_inspection_preserves_base_cycle_after_timeout() {
 }

 #[tokio::test(start_paused = true)]
+#[serial]
 async fn stable_maintenance_detection_preserves_base_cycle_after_timeout() {
    let ctx = CancellationToken::new();

@@ -2836,6 +2862,7 @@ async fn maintenance_feature_inspection_stops_on_cancellation() {
 }

 #[test]
+#[serial]
 fn test_cycle_interval_prefers_explicit_cycle_override() {
    with_var(ENV_SCANNER_SPEED, Some("slowest"), || {
        with_var(ENV_SCANNER_CYCLE, Some("42"), || {
@@ -2845,6 +2872,7 @@ fn test_cycle_interval_prefers_explicit_cycle_override() {
 }

 #[test]
+#[serial]
 fn test_cycle_interval_prefers_explicit_cycle_over_default_cycle() {
    let _guard = ScannerDefaultCycleGuard::set(TEST_DEFAULT_SCANNER_CYCLE_SECS);

@@ -2854,6 +2882,7 @@ fn test_cycle_interval_prefers_explicit_cycle_over_default_cycle() {
 }

 #[test]
+#[serial]
 fn test_cycle_interval_uses_scanner_default_speed_override_when_unconfigured() {
    let _guard = ScannerDefaultSpeedGuard::set(ScannerSpeed::Slowest);

@@ -2863,6 +2892,7 @@ fn test_cycle_interval_uses_scanner_default_speed_override_when_unconfigured() {
 }

 #[test]
+#[serial]
 fn test_cycle_interval_prefers_explicit_speed_over_default_speed_override() {
    let _guard = ScannerDefaultSpeedGuard::set(ScannerSpeed::Slowest);

@@ -2880,6 +2910,7 @@ fn test_cycle_interval_prefers_explicit_speed_over_default_speed_override() {
 }

 #[test]
+#[serial]
 fn test_cycle_interval_uses_default_cycle_override_when_unconfigured() {
    let _guard = ScannerDefaultCycleGuard::set(TEST_DEFAULT_SCANNER_CYCLE_SECS);

@@ -3043,6 +3074,7 @@ fn scanner_cycle_wait_plan_drives_growth_resets_and_bitrot_cap() {
 }

 #[test]
+#[serial]
 fn scanner_cycle_schedule_status_reports_effective_backoff() {
    record_scanner_cycle_schedule(Duration::from_millis(86_400_001), true, 2_048, true, 7);

@@ -3322,6 +3354,7 @@ fn dirty_usage_wakes_are_disabled_for_explicit_cycle_policy() {
 }

 #[test]
+#[serial]
 fn clean_idle_cap_preserves_default_bitrot_coverage_window() {
    let config = ScannerRuntimeConfig {
        bitrot_cycle: Some(Duration::from_secs(30 * 24 * 60 * 60)),
@@ -3351,6 +3384,7 @@ fn clean_idle_cap_allows_policy_max_when_bitrot_is_disabled() {
 }

 #[test]
+#[serial]
 fn clean_idle_cap_never_shortens_the_base_cycle() {
    let config = ScannerRuntimeConfig {
        bitrot_cycle: Some(Duration::from_secs(60)),
@@ -3364,6 +3398,7 @@ fn clean_idle_cap_never_shortens_the_base_cycle() {
 }

 #[test]
+#[serial]
 fn test_cycle_interval_keeps_default_cycle_with_explicit_speed() {
    let _guard = ScannerDefaultCycleGuard::set(TEST_DEFAULT_SCANNER_CYCLE_SECS);

@@ -3381,6 +3416,7 @@ fn test_cycle_interval_keeps_default_cycle_with_explicit_speed() {
 }

 #[test]
+#[serial]
 fn test_cycle_interval_prefers_explicit_start_delay_over_default_cycle() {
    let _guard = ScannerDefaultCycleGuard::set(TEST_DEFAULT_SCANNER_CYCLE_SECS);

@@ -3394,6 +3430,7 @@ fn test_cycle_interval_prefers_explicit_start_delay_over_default_cycle() {
 }

 #[test]
+#[serial]
 fn test_cycle_interval_supports_minio_speed_alias() {
    with_var_unset(ENV_SCANNER_SPEED, || {
        with_var_unset(ENV_SCANNER_CYCLE, || {
@@ -3407,6 +3444,7 @@ fn test_cycle_interval_supports_minio_speed_alias() {
 }

 #[test]
+#[serial]
 fn test_cycle_interval_supports_minio_cycle_alias() {
    with_var_unset(ENV_SCANNER_CYCLE, || {
        with_var_unset(ENV_SCANNER_START_DELAY_SECS, || {
@@ -3426,6 +3464,7 @@ fn test_randomized_cycle_delay_handles_small_start_delay() {
 }

 #[tokio::test]
+#[serial]
 async fn test_wait_for_next_scanner_cycle_wakes_for_dirty_usage() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();

@@ -3451,6 +3490,7 @@ async fn test_wait_for_next_scanner_cycle_wakes_for_dirty_usage() {
 }

 #[tokio::test]
+#[serial]
 async fn test_wait_for_next_scanner_cycle_sees_unattempted_dirty_usage() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let dirty_generation = crate::scanner_io::dirty_usage_generation();
@@ -3472,6 +3512,7 @@ async fn test_wait_for_next_scanner_cycle_sees_unattempted_dirty_usage() {
 }

 #[tokio::test(start_paused = true)]
+#[serial]
 async fn test_wait_for_next_scanner_cycle_retries_stable_dirty_usage_on_timer() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -3493,6 +3534,7 @@ async fn test_wait_for_next_scanner_cycle_retries_stable_dirty_usage_on_timer()
 }

 #[tokio::test(start_paused = true)]
+#[serial]
 async fn test_wait_for_next_scanner_cycle_can_defer_dirty_wakes_until_timer() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3511,6 +3553,7 @@ async fn test_wait_for_next_scanner_cycle_can_defer_dirty_wakes_until_timer() {
 }

 #[tokio::test]
+#[serial]
 async fn test_wait_for_next_scanner_cycle_wakes_for_repeated_dirty_bucket() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -3536,6 +3579,7 @@ async fn test_wait_for_next_scanner_cycle_wakes_for_repeated_dirty_bucket() {
 }

 #[tokio::test]
+#[serial]
 async fn test_wait_for_next_scanner_cycle_reschedules_for_runtime_config() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let observed_generation = crate::runtime_config::scanner_runtime_config_generation();
@@ -3563,6 +3607,7 @@ async fn test_wait_for_next_scanner_cycle_reschedules_for_runtime_config() {
 }

 #[tokio::test]
+#[serial]
 async fn test_wait_for_next_scanner_cycle_reschedules_for_maintenance_change() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let observed_generation = crate::scanner_io::scanner_maintenance_generation();
@@ -3806,6 +3851,7 @@ fn scanner_activity_after_a_cycle_restores_the_base_interval() {
 }

 #[tokio::test(start_paused = true)]
+#[serial]
 async fn distributed_clean_idle_wait_wakes_at_base_interval_for_remote_activity() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3833,6 +3879,7 @@ async fn distributed_clean_idle_wait_wakes_at_base_interval_for_remote_activity(
 }

 #[tokio::test(start_paused = true)]
+#[serial]
 async fn superseded_retry_wait_defers_dirty_cluster_activity_until_timer() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3860,6 +3907,7 @@ async fn superseded_retry_wait_defers_dirty_cluster_activity_until_timer() {
 }

 #[tokio::test(start_paused = true)]
+#[serial]
 async fn distributed_clean_idle_wait_blocks_backoff_for_unpropagated_maintenance() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3886,6 +3934,7 @@ async fn distributed_clean_idle_wait_blocks_backoff_for_unpropagated_maintenance
 }

 #[tokio::test(start_paused = true)]
+#[serial]
 async fn distributed_clean_idle_wait_fails_closed_when_a_peer_is_unverifiable() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3912,6 +3961,7 @@ async fn distributed_clean_idle_wait_fails_closed_when_a_peer_is_unverifiable()
 }

 #[tokio::test(start_paused = true)]
+#[serial]
 async fn distributed_clean_idle_wait_keeps_the_extended_deadline_when_peers_are_clean() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3939,6 +3989,7 @@ async fn distributed_clean_idle_wait_keeps_the_extended_deadline_when_peers_are_
 }

 #[tokio::test(start_paused = true)]
+#[serial]
 async fn scanner_activity_probe_wait_is_cancellation_aware() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3969,6 +4020,7 @@ async fn scanner_activity_probe_wait_is_cancellation_aware() {
 }

 #[tokio::test(start_paused = true)]
+#[serial]
 async fn scanner_activity_probe_wait_stops_after_leader_lock_loss() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -4000,6 +4052,7 @@ async fn scanner_activity_probe_wait_stops_after_leader_lock_loss() {
 }

 #[test]
+#[serial]
 fn test_get_cycle_scan_mode_runs_deep_until_selection_window_completes() {
    with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("3600"), || {
        let mode = get_cycle_scan_mode(10, 0, Some(Utc::now()), bitrot_scan_cycle());
@@ -4008,6 +4061,7 @@ fn test_get_cycle_scan_mode_runs_deep_until_selection_window_completes() {
 }

 #[test]
+#[serial]
 fn test_get_cycle_scan_mode_respects_elapsed_bitrot_cycle() {
    with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("3600"), || {
        let recent = Utc::now() - chrono::Duration::minutes(30);
@@ -4019,6 +4073,7 @@ fn test_get_cycle_scan_mode_respects_elapsed_bitrot_cycle() {
 }

 #[test]
+#[serial]
 fn test_get_cycle_scan_mode_can_disable_periodic_deep_scan() {
    with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("off"), || {
        assert_eq!(get_cycle_scan_mode(1, 0, None, bitrot_scan_cycle()), HealScanMode::Normal);
@@ -4026,6 +4081,7 @@ fn test_get_cycle_scan_mode_can_disable_periodic_deep_scan() {
 }

 #[test]
+#[serial]
 fn test_background_heal_info_for_scan_start_marks_deep_active() {
    let now = Utc::now();
    let info =
@@ -4038,6 +4094,7 @@ fn test_background_heal_info_for_scan_start_marks_deep_active() {
 }

 #[test]
+#[serial]
 fn test_background_heal_info_for_scan_start_keeps_deep_window_start() {
    with_var_unset(ENV_SCANNER_BITROT_CYCLE_SECS, || {
        let started_at = Utc::now();
@@ -18,6 +18,7 @@ use super::*;
 use crate::storage_api::VersionPurgeStatusType;
 use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
 use rustfs_filemeta::{FileInfo, FileMeta};
+use serial_test::serial;
 use std::io::Write;
 #[cfg(unix)]
 use std::os::unix::fs::{PermissionsExt, symlink};
@@ -355,6 +356,7 @@ impl Drop for TestGuard {
 }

 #[tokio::test]
+#[serial]
 async fn test_should_skip_failed_respects_ttl() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
@@ -376,6 +378,7 @@ async fn test_should_skip_failed_respects_ttl() {
 }

 #[tokio::test]
+#[serial]
 async fn test_record_failed_ttl_zero_noop() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(0, 100, &mut scanner, temp_dir);
@@ -464,6 +467,7 @@ fn test_should_account_replication_stats_only_for_live_object_versions() {
 }

 #[tokio::test]
+#[serial]
 async fn test_heal_replication_only_queues_pending_null_deletes() {
    async fn replication_skipped_count() -> u64 {
        global_metrics()
@@ -712,6 +716,7 @@ async fn test_scanner_heal_admission_accounting_maps_deep_scan_to_bitrot() {
 }

 #[test]
+#[serial]
 fn test_excessive_version_alert_thresholds_use_env() {
    with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSIONS, Some("3"), || {
        with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, Some("100"), || {
@@ -726,6 +731,7 @@ fn test_excessive_version_alert_thresholds_use_env() {
 }

 #[test]
+#[serial]
 fn test_excessive_folders_threshold_uses_env() {
    with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_FOLDERS, Some("3"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -735,6 +741,7 @@ fn test_excessive_folders_threshold_uses_env() {
 }

 #[test]
+#[serial]
 fn test_excessive_folders_threshold_default_supports_pbs_layout() {
    with_var_unset(rustfs_config::ENV_SCANNER_ALERT_EXCESS_FOLDERS, || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -744,6 +751,7 @@ fn test_excessive_folders_threshold_default_supports_pbs_layout() {
 }

 #[test]
+#[serial]
 fn test_scanner_yield_every_n_objects_uses_env() {
    with_var(rustfs_config::ENV_SCANNER_YIELD_EVERY_N_OBJECTS, Some("32"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -753,6 +761,7 @@ fn test_scanner_yield_every_n_objects_uses_env() {
 }

 #[test]
+#[serial]
 fn test_scanner_yield_every_n_objects_uses_default() {
    with_var_unset(rustfs_config::ENV_SCANNER_YIELD_EVERY_N_OBJECTS, || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -879,6 +888,7 @@ fn test_order_folders_for_resume_reports_stale_hint() {
 }

 #[tokio::test]
+#[serial]
 async fn test_record_failed_prunes_to_max_entries() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(1000, 2, &mut scanner, temp_dir);
@@ -910,6 +920,7 @@ async fn test_record_failed_prunes_to_max_entries() {
 }

 #[tokio::test]
+#[serial]
 async fn test_prune_failed_objects_cache_drops_expired() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(5, 10, &mut scanner, temp_dir);
@@ -933,6 +944,7 @@ async fn test_prune_failed_objects_cache_drops_expired() {
 }

 #[tokio::test]
+#[serial]
 async fn test_prune_failed_objects_max_zero_keeps_fresh() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir);
@@ -1689,6 +1701,7 @@ async fn test_heal_actions_returns_actual_size_without_inline_heal() {
 }

 #[tokio::test]
+#[serial]
 #[cfg(unix)]
 async fn test_scan_folder_skips_unreadable_child_directory() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
@@ -1721,6 +1734,7 @@ async fn test_scan_folder_skips_unreadable_child_directory() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -1799,6 +1813,7 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_folder_xl_meta_named_directory_uses_namespace_descent() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -1844,6 +1859,7 @@ async fn test_scan_folder_xl_meta_named_directory_uses_namespace_descent() {
 }

 #[tokio::test(flavor = "current_thread")]
+#[serial]
 async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
    let logs = CapturedLogs::default();
    let subscriber = tracing_subscriber::fmt()
@@ -2005,6 +2021,7 @@ async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_folder_missing_xl_meta_stops_erasure_data_dir_descent() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2082,6 +2099,7 @@ async fn test_scan_folder_missing_xl_meta_stops_erasure_data_dir_descent() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_folder_uuid_namespace_part_name_directory_is_not_data_dir() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2143,6 +2161,7 @@ async fn test_scan_folder_uuid_namespace_part_name_directory_is_not_data_dir() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_folder_non_erasure_metadata_keeps_namespace_descent() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2184,6 +2203,7 @@ async fn test_scan_folder_non_erasure_metadata_keeps_namespace_descent() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_folder_compacted_parent_sends_partial_update() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2225,6 +2245,7 @@ async fn test_scan_folder_compacted_parent_sends_partial_update() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_data_folder_cancelled_before_scan_clears_current_path() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2269,6 +2290,7 @@ async fn test_scan_data_folder_cancelled_before_scan_clears_current_path() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_data_folder_returns_partial_cache_on_budget_cancel() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2324,6 +2346,7 @@ async fn test_scan_data_folder_returns_partial_cache_on_budget_cancel() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_data_folder_reports_invalid_checkpoint_ignored_once() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2368,6 +2391,7 @@ async fn test_scan_data_folder_reports_invalid_checkpoint_ignored_once() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_data_folder_resume_hint_prioritizes_next_existing_folder() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2441,6 +2465,7 @@ async fn test_scan_data_folder_resume_hint_prioritizes_next_existing_folder() {
 }

 #[tokio::test]
+#[serial]
 async fn scan_data_folder_missing_bucket_returns_partial() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2492,6 +2517,7 @@ async fn scan_data_folder_missing_bucket_returns_partial() {
 }

 #[tokio::test]
+#[serial]
 async fn scan_data_folder_missing_scan_root_returns_partial() {
    let (scanner, temp_dir) = build_test_scanner().await;
    tokio::fs::remove_dir_all(&temp_dir)
@@ -2537,6 +2563,7 @@ async fn scan_data_folder_missing_scan_root_returns_partial() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_data_folder_resume_hint_orders_across_new_and_existing_folders() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2605,6 +2632,7 @@ async fn test_scan_data_folder_resume_hint_orders_across_new_and_existing_folder
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_data_folder_partial_object_budget_accumulates_progress() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2687,6 +2715,7 @@ async fn test_scan_data_folder_partial_object_budget_accumulates_progress() {
 }

 #[tokio::test]
+#[serial]
 async fn test_partial_compacted_entry_does_not_carry_children() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2732,6 +2761,7 @@ async fn test_partial_compacted_entry_does_not_carry_children() {
 }

 #[tokio::test]
+#[serial]
 async fn test_partial_entry_does_not_carry_missing_old_child() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2764,6 +2794,7 @@ async fn test_partial_entry_does_not_carry_missing_old_child() {
 }

 #[tokio::test]
+#[serial]
 async fn test_legacy_windows_cache_rebuilds_and_round_trips_portable_keys() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2830,6 +2861,7 @@ async fn test_legacy_windows_cache_rebuilds_and_round_trips_portable_keys() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_data_folder_success_clears_resume_hint() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2872,6 +2904,7 @@ async fn test_scan_data_folder_success_clears_resume_hint() {
 }

 #[tokio::test]
+#[serial]
 async fn test_scan_data_folder_keeps_unresolved_objects_partial() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2918,6 +2951,7 @@ async fn test_scan_data_folder_keeps_unresolved_objects_partial() {
 }

 #[tokio::test]
+#[serial]
 #[cfg(unix)]
 async fn test_scan_folder_ignores_symlinked_child_directory() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
@@ -27,6 +27,7 @@ use crate::{
    init_local_disks_with_instance_ctx, new_disk, path2_bucket_object_with_base_path,
 };
 use rustfs_filemeta::FileInfo;
+use serial_test::serial;
 use temp_env::with_var;
 use time::OffsetDateTime;
 use uuid::Uuid;
@@ -102,6 +103,7 @@ async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc<ECStore>) {
 }

 #[tokio::test]
+#[serial]
 async fn scanner_cache_locks_block_same_source_workers() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    let set = &store.pools[0].disk_set[0];
@@ -128,6 +130,7 @@ async fn scanner_cache_locks_block_same_source_workers() {
 }

 #[tokio::test]
+#[serial]
 async fn scanner_cache_locks_allow_cross_source_workers() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    let first_set = &store.pools[0].disk_set[0];
@@ -146,6 +149,7 @@ async fn scanner_cache_locks_allow_cross_source_workers() {
 }

 #[tokio::test]
+#[serial]
 async fn scanner_cycle_is_deferred_while_rebalance_is_active() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    let mut pool_stats = vec![EcstoreRebalanceStats::default(); store.pools.len()];
@@ -181,6 +185,7 @@ async fn scanner_cycle_is_deferred_while_rebalance_is_active() {
 }

 #[tokio::test]
+#[serial]
 async fn scanner_cycle_is_deferred_while_terminal_decommission_is_blocked() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    for decommission in [
@@ -225,6 +230,7 @@ async fn data_usage_publish_fails_when_receiver_is_closed() {
 }

 #[tokio::test]
+#[serial]
 async fn multi_pool_scanner_cycle_publishes_combined_usage() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    let bucket = format!("scanner-union-{}", Uuid::new_v4().simple());
@@ -272,6 +278,7 @@ async fn multi_pool_scanner_cycle_publishes_combined_usage() {
 }

 #[tokio::test]
+#[serial]
 async fn multi_pool_scanner_cycle_zero_fills_bucket_absent_from_first_pool() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    let bucket = format!("scanner-second-pool-{}", Uuid::new_v4().simple());
@@ -359,6 +366,7 @@ fn object_lock_config_enabled_accepts_enabled_only() {
 }

 #[test]
+#[serial]
 fn dirty_usage_snapshot_clear_preserves_newer_generation() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -373,6 +381,7 @@ fn dirty_usage_snapshot_clear_preserves_newer_generation() {
 }

 #[test]
+#[serial]
 fn dirty_usage_generation_acknowledgement_preserves_newer_mutations() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -398,6 +407,7 @@ fn dirty_usage_generation_acknowledgement_preserves_newer_mutations() {
 }

 #[test]
+#[serial]
 fn dirty_usage_generation_acknowledgement_rejects_stale_process_and_future_generation() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -427,6 +437,7 @@ fn dirty_usage_generation_acknowledgement_rejects_stale_process_and_future_gener
 }

 #[test]
+#[serial]
 fn dirty_usage_snapshot_detects_uncovered_generation() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -451,6 +462,7 @@ fn generation_saturates_instead_of_wrapping() {
 }

 #[test]
+#[serial]
 fn dirty_usage_snapshot_clears_a_stably_absent_bucket_after_durable_save() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -472,6 +484,7 @@ fn dirty_usage_snapshot_clears_a_stably_absent_bucket_after_durable_save() {
 }

 #[test]
+#[serial]
 fn dirty_usage_snapshot_preserves_an_absent_bucket_recorded_after_listing_started() {
    clear_dirty_usage_buckets_for_tests();
    let generation_before_bucket_list = dirty_usage_generation();
@@ -486,6 +499,7 @@ fn dirty_usage_snapshot_preserves_an_absent_bucket_recorded_after_listing_starte
 }

 #[test]
+#[serial]
 fn deleting_a_clean_bucket_invalidates_an_inflight_usage_snapshot() {
    clear_dirty_usage_buckets_for_tests();
    let snapshot = snapshot_dirty_usage_buckets(&[bucket_info("photos")], dirty_usage_generation());
@@ -499,6 +513,7 @@ fn deleting_a_clean_bucket_invalidates_an_inflight_usage_snapshot() {
 }

 #[test]
+#[serial]
 fn deleting_a_bucket_during_listing_invalidates_the_resulting_usage_snapshot() {
    clear_dirty_usage_buckets_for_tests();
    let generation_before_bucket_list = dirty_usage_generation();
@@ -512,6 +527,7 @@ fn deleting_a_bucket_during_listing_invalidates_the_resulting_usage_snapshot() {
 }

 #[test]
+#[serial]
 fn scanner_maintenance_change_advances_generation_and_marks_usage_dirty() {
    clear_dirty_usage_buckets_for_tests();
    let generation = scanner_maintenance_generation();
@@ -524,6 +540,7 @@ fn scanner_maintenance_change_advances_generation_and_marks_usage_dirty() {
 }

 #[test]
+#[serial]
 fn dirty_usage_clear_excludes_failed_buckets() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -555,6 +572,7 @@ fn dirty_usage_clear_plan_excludes_cache_save_failures() {
 }

 #[test]
+#[serial]
 fn dirty_usage_is_acknowledged_only_after_durable_usage_confirmation() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -572,6 +590,7 @@ fn dirty_usage_is_acknowledged_only_after_durable_usage_confirmation() {
 }

 #[test]
+#[serial]
 fn clear_dirty_usage_bucket_removes_deleted_bucket_marker() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -898,30 +917,35 @@ async fn bucket_cache_pending_heal_reaches_cycle_maintenance_state() {
 }

 #[test]
+#[serial]
 fn scanner_concurrency_limit_preserves_available_when_unconfigured() {
    crate::reset_foreground_read_activity_for_test();
    assert_eq!(scanner_concurrency_limit(0, 4), 4);
 }

 #[test]
+#[serial]
 fn scanner_concurrency_limit_caps_to_configured_value() {
    crate::reset_foreground_read_activity_for_test();
    assert_eq!(scanner_concurrency_limit(2, 4), 2);
 }

 #[test]
+#[serial]
 fn scanner_concurrency_limit_never_exceeds_available_work() {
    crate::reset_foreground_read_activity_for_test();
    assert_eq!(scanner_concurrency_limit(8, 4), 4);
 }

 #[test]
+#[serial]
 fn scanner_concurrency_limit_handles_no_available_work() {
    crate::reset_foreground_read_activity_for_test();
    assert_eq!(scanner_concurrency_limit(2, 0), 0);
 }

 #[test]
+#[serial]
 fn scanner_concurrency_limit_yields_to_foreground_reads() {
    crate::reset_foreground_read_activity_for_test();
    crate::set_foreground_read_activity(8);
@@ -931,6 +955,7 @@ fn scanner_concurrency_limit_yields_to_foreground_reads() {
 }

 #[test]
+#[serial]
 fn scanner_concurrency_limit_yields_to_streaming_reads() {
    crate::reset_foreground_read_activity_for_test();
    let _guard = crate::ForegroundReadGuard::new();
@@ -954,6 +979,7 @@ fn increment_atomic_usize_saturates_at_max() {
 }

 #[test]
+#[serial]
 fn scanner_max_concurrent_set_scans_uses_env_cap() {
    with_var(ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, Some("2"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -963,6 +989,7 @@ fn scanner_max_concurrent_set_scans_uses_env_cap() {
 }

 #[test]
+#[serial]
 fn scanner_max_concurrent_disk_scans_uses_env_cap() {
    with_var(ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, Some("1"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -258,6 +258,7 @@ impl SleepTimer {
 #[cfg(test)]
 mod tests {
    use super::*;
+    use serial_test::serial;
    use temp_env::{with_var, with_var_unset};

    struct ScannerDefaultSpeedGuard;
@@ -325,6 +326,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn test_refresh_from_env_applies_speed_and_idle_mode_for_next_cycle() {
        let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
        SCANNER_IDLE_MODE.store(true, Ordering::Relaxed);
@@ -344,6 +346,7 @@ mod tests {
    }

    #[test]
+    #[serial]
    fn test_refresh_from_env_uses_default_speed_override_when_speed_unset() {
        let _guard = ScannerDefaultSpeedGuard::set(ScannerSpeed::Slowest);
        let s = DynamicSleeper::new(ScannerSpeed::Default);
@@ -359,6 +362,7 @@ mod tests {
    }

    #[tokio::test(start_paused = true)]
+    #[serial]
    async fn test_fastest_never_sleeps() {
        let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
        SCANNER_IDLE_MODE.store(true, Ordering::Relaxed);
@@ -372,6 +376,7 @@ mod tests {
    }

    #[tokio::test(start_paused = true)]
+    #[serial]
    async fn test_idle_mode_off_skips_sleep() {
        let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
        SCANNER_IDLE_MODE.store(false, Ordering::Relaxed);
@@ -14,6 +14,7 @@

 #![recursion_limit = "256"]

+use futures::FutureExt;
 use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT;
 use rustfs_scanner::scanner_folder::ScannerItem;
 use rustfs_scanner::scanner_io::ScannerIODisk;
@@ -22,8 +23,10 @@ use rustfs_scanner::{
    scanner::init_data_scanner,
 };
 use s3s::dto::RestoreRequest;
+use serial_test::serial;
 use std::{
    collections::HashMap,
+    env,
    path::{Path, PathBuf},
    sync::{Arc, Once, OnceLock},
    time::Duration,
@@ -532,19 +535,31 @@ async fn wait_for_transition(ecstore: &Arc<ECStore>, bucket: &str, object: &str,
    }
 }

-// Run `test_fn` with `ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT`
-// set to `"1"` for its duration. `temp_env` serializes environment mutations
-// globally, preventing data races when multiple tests run in parallel.
+// SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio
+// runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the
+// `env::set_var` / `env::remove_var` window.
+#[allow(unsafe_code)]
 async fn with_forced_immediate_enqueue_timeout<F, Fut>(test_fn: F)
 where
    F: FnOnce() -> Fut,
    Fut: std::future::Future<Output = ()>,
 {
-    temp_env::async_with_vars(
-        [(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, Some("1"))],
-        test_fn(),
-    )
-    .await;
+    let original = env::var_os(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT);
+    unsafe {
+        env::set_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, "1");
+    }
+    let result = std::panic::AssertUnwindSafe(test_fn()).catch_unwind().await;
+    match original {
+        Some(value) => unsafe {
+            env::set_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, value);
+        },
+        None => unsafe {
+            env::remove_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT);
+        },
+    }
+    if let Err(err) = result {
+        std::panic::resume_unwind(err);
+    }
 }

 mod serial_tests {
@@ -577,6 +592,7 @@ mod serial_tests {
    ///     body (GET won) or a clean object/version-not-found (expiry won). A
    ///     tier-fetch failure -- the #3491 symptom -- is never tolerated.
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-2)"]
    async fn test_expire_transitioned_object_never_races_concurrent_get() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -722,6 +738,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
    async fn rejected_transition_candidate_is_recovered_from_persisted_delete_journal() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -808,6 +825,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
    async fn cancelled_before_cleanup_store_resolution_persists_journal() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -901,6 +919,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
    async fn rejected_transition_cleanup_durability_matrix() {
        #[derive(Clone, Copy)]
@@ -1040,6 +1059,7 @@ mod serial_tests {
    }

    #[test]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    fn test_transition_and_restore_flows() {
        std::thread::Builder::new()
@@ -1365,6 +1385,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_scanner_enqueues_free_version_cleanup_for_stale_transitioned_object() {
        let (disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -1425,6 +1446,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_scanner_cleanup_still_works_after_immediate_compensation_transition() {
        let (disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -1482,6 +1504,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_existing_object_backfill_is_idempotent_after_immediate_compensation_transition() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -1524,6 +1547,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "FAILING on main: excluded from the serial ILM lane pending a fix, see rustfs/backlog#1148 (ilm-1 partial)"]
    async fn test_noncurrent_expiry_still_works_after_immediate_compensation_transition() {
        let (disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -1607,6 +1631,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "FAILING on main: excluded from the serial ILM lane pending a fix, see rustfs/backlog#1148 (ilm-1 partial)"]
    async fn test_noncurrent_transition_still_works_after_immediate_compensation_transition() {
        let (disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -1689,6 +1714,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_modeled_versioned_delete_creates_delete_marker_after_immediate_compensation_transition() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -1736,6 +1762,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_modeled_delete_marker_cleanup_after_immediate_compensation_transition() {
        let (disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -1812,6 +1839,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_scanner_expires_zero_day_current_version() {
        let (disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -1838,6 +1866,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_put_object_immediately_enqueues_zero_day_current_expiry() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -1875,6 +1904,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_scanner_expires_zero_day_noncurrent_version() {
        let (disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -1941,6 +1971,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_put_object_immediately_enqueues_zero_day_noncurrent_expiry() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -2001,6 +2032,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    async fn test_background_scanner_expires_zero_day_current_version() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(true).await;

@@ -2024,6 +2056,7 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_background_scanner_expires_zero_day_current_version_for_exact_key_prefix() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -2089,6 +2122,7 @@ mod serial_tests {
    /// tier object is untouched (zero `remove` calls) -> GET streams from the
    /// tier again -> a second restore succeeds.
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8)"]
    async fn test_restore_chain_local_read_expiry_keeps_remote_and_allows_re_restore() {
        let (_disk_paths, ecstore) = setup_test_env().await;
@@ -2220,6 +2254,7 @@ mod serial_tests {
    /// parts) must reassemble the exact part layout: part count and sizes,
    /// the multipart ETag, and byte-identical content across part boundaries.
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8)"]
    async fn test_multipart_restore_preserves_parts_and_etag() {
        let (_disk_paths, ecstore) = setup_test_env().await;
@@ -245,6 +245,18 @@ impl TestECStoreEnvBuilder {
            .await
            .expect("build test ECStore");

+        // The production bootstrap only persists pool.bin from the elected
+        // first cluster node.  Test stores intentionally have no cluster
+        // election, but heal-format still requires that durable fence before
+        // it can write any disk format.  Materialize the validated topology
+        // here so the shared fixture models a ready single-node store.
+        let mut pool_meta = ecstore.pool_meta.read().await.clone();
+        pool_meta.dont_save = false;
+        pool_meta
+            .save(ecstore.pools.clone())
+            .await
+            .expect("persist test pool metadata");
+
        if self.init_bucket_metadata {
            let buckets_list = ecstore
                .list_bucket(&BucketOptions {
@@ -17,6 +17,220 @@
 //! This binary shares RustFS's existing subcommand dispatcher and provides the
 //! documented entry point for offline tooling such as `inspect bucket-meta`.

-fn main() {
+use std::fs;
+use std::io::{Read as _, Write as _};
+use std::path::{Path, PathBuf};
+use std::process::ExitCode;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use rustfs::connect::offline::{OfflineEnrollment, OfflineKeyStore};
+
+/// Owner read/write only. The response names the key being enrolled and the
+/// challenge it answers; neither belongs to anyone else on the machine.
+#[cfg(unix)]
+const RESPONSE_MODE: u32 = 0o600;
+
+const USAGE: &str = "\
+Usage: rustfs-cli connect offline enroll --challenge <path|-> --output <path> [--key-dir <path>]
+
+Answers a Connect offline enrolment challenge without a network. Reads the
+challenge from a file or from stdin when the path is `-`, verifies it against the
+enrolment root compiled into this binary, mints the key being enrolled on first
+use, and writes the signed response.
+
+No secret is ever accepted on the command line.
+";
+
+fn main() -> ExitCode {
+    let arguments: Vec<String> = std::env::args().skip(1).collect();
+
+    // Offline enrolment is handled before the server dispatcher is reached, and
+    // the reason is the surface's whole point: `run_process` builds a Tokio
+    // runtime and enters the server's async main. An air-gapped enrolment must
+    // not start a runtime, a task, or anything that could open a socket, so the
+    // two paths cannot share an entry.
+    if matches!(
+        arguments.first().map(String::as_str),
+        Some("connect") if matches!(arguments.get(1).map(String::as_str), Some("offline"))
+    ) {
+        return match run_offline(&arguments[2..]) {
+            Ok(()) => ExitCode::SUCCESS,
+            Err(message) => {
+                eprintln!("rustfs-cli: {message}");
+                ExitCode::FAILURE
+            }
+        };
+    }
+
    rustfs::startup_entrypoint::run_process();
+
+    ExitCode::SUCCESS
+}
+
+fn run_offline(arguments: &[String]) -> Result<(), String> {
+    match arguments.first().map(String::as_str) {
+        Some("enroll") => enroll(&arguments[1..]),
+        Some(other) => Err(format!("unknown offline subcommand `{other}`\n\n{USAGE}")),
+        None => Err(format!("missing offline subcommand\n\n{USAGE}")),
+    }
+}
+
+fn enroll(arguments: &[String]) -> Result<(), String> {
+    let mut challenge_path: Option<String> = None;
+    let mut output_path: Option<String> = None;
+    let mut key_directory: Option<String> = None;
+
+    let mut index = 0;
+    while index < arguments.len() {
+        let flag = arguments[index].as_str();
+        let take_value = |name: &str| -> Result<String, String> {
+            arguments
+                .get(index + 1)
+                .cloned()
+                .ok_or_else(|| format!("`{name}` needs a value\n\n{USAGE}"))
+        };
+
+        match flag {
+            "--challenge" => challenge_path = Some(take_value("--challenge")?),
+            "--output" => output_path = Some(take_value("--output")?),
+            "--key-dir" => key_directory = Some(take_value("--key-dir")?),
+            "-h" | "--help" => {
+                println!("{USAGE}");
+                return Ok(());
+            }
+            other => return Err(format!("unknown option `{other}`\n\n{USAGE}")),
+        }
+
+        index += 2;
+    }
+
+    let challenge_path = challenge_path.ok_or_else(|| format!("`--challenge` is required\n\n{USAGE}"))?;
+    let output_path = output_path.ok_or_else(|| format!("`--output` is required\n\n{USAGE}"))?;
+    let key_directory = key_directory.unwrap_or_else(|| ".".to_string());
+
+    let challenge = read_challenge(&challenge_path)?;
+
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(|_| "the system clock is before the Unix epoch".to_string())?
+        .as_secs() as i64;
+
+    let verified = OfflineEnrollment::verify_challenge(&challenge, now).map_err(|error| error.to_string())?;
+
+    // First use mints the key; a retry answers with the one already enrolled,
+    // because the operator may already be carrying a response naming it.
+    let key = OfflineKeyStore::new(&key_directory)
+        .load_or_create()
+        .map_err(|error| error.to_string())?;
+
+    let mut device_nonce = [0u8; 32];
+    getrandom(&mut device_nonce)?;
+
+    let response = OfflineEnrollment::build_response(&verified, &key, &device_nonce, now).map_err(|error| error.to_string())?;
+
+    write_response(Path::new(&output_path), &response)?;
+
+    Ok(())
+}
+
+/// Reads the challenge from a file, or from stdin when the path is `-`.
+///
+/// A challenge is not a secret — it is signed, public, and carried in by hand —
+/// so accepting a path is safe. The response's key never arrives this way.
+fn read_challenge(path: &str) -> Result<Vec<u8>, String> {
+    if path == "-" {
+        let mut buffer = Vec::new();
+        std::io::stdin()
+            .read_to_end(&mut buffer)
+            .map_err(|error| format!("cannot read the challenge from stdin: {error}"))?;
+        return Ok(buffer);
+    }
+
+    fs::read(path).map_err(|error| format!("cannot read the challenge at {path}: {error}"))
+}
+
+/// Writes the response durably and atomically at mode 0600.
+///
+/// Not the no-clobber publish `IdentityStore` performs for a key: an operator
+/// who reruns an enrolment expects the response file to be replaced, whereas a
+/// second key would strand the first. Same durability, deliberately different
+/// publication rule.
+fn write_response(path: &Path, response: &[u8]) -> Result<(), String> {
+    let parent = path.parent().filter(|parent| !parent.as_os_str().is_empty());
+    let temporary: PathBuf = match parent {
+        Some(parent) => parent.join(format!(".{}.tmp", file_name(path))),
+        None => PathBuf::from(format!(".{}.tmp", file_name(path))),
+    };
+
+    let mut options = fs::OpenOptions::new();
+    options.write(true).create(true).truncate(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt as _;
+        options.mode(RESPONSE_MODE);
+    }
+
+    let write = (|| -> std::io::Result<()> {
+        let mut file = options.open(&temporary)?;
+        file.write_all(response)?;
+
+        // The umask can only narrow the creation mode, so set the exact mode
+        // before the bytes become durable.
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt as _;
+            file.set_permissions(fs::Permissions::from_mode(RESPONSE_MODE))?;
+        }
+
+        file.sync_all()
+    })();
+
+    if let Err(error) = write {
+        let _ = fs::remove_file(&temporary);
+        return Err(format!("cannot write the response to {}: {error}", path.display()));
+    }
+
+    fs::rename(&temporary, path).map_err(|error| {
+        let _ = fs::remove_file(&temporary);
+        format!("cannot publish the response at {}: {error}", path.display())
+    })?;
+
+    if let Some(parent) = parent {
+        sync_directory(parent);
+    }
+
+    Ok(())
+}
+
+fn file_name(path: &Path) -> String {
+    path.file_name()
+        .map(|name| name.to_string_lossy().into_owned())
+        .unwrap_or_else(|| "response".to_string())
+}
+
+/// Fsync the directory so the renamed entry survives power loss. Directories
+/// cannot be opened for syncing on Windows, where this is a no-op.
+fn sync_directory(directory: &Path) {
+    #[cfg(unix)]
+    {
+        if let Ok(handle) = fs::File::open(directory) {
+            let _ = handle.sync_all();
+        }
+    }
+    #[cfg(not(unix))]
+    let _ = directory;
+}
+
+/// Fills `buffer` with operating-system randomness.
+///
+/// The device nonce must be unpredictable: it is what stops a captured response
+/// being replayed as a fresh one. Sourced through p256's pinned rand_core 0.6
+/// rather than the workspace `rand` 0.10, matching `identity.rs`; the two are
+/// different crate versions and only the pinned one is on p256's own path.
+fn getrandom(buffer: &mut [u8]) -> Result<(), String> {
+    use p256::elliptic_curve::rand_core::{OsRng, RngCore as _};
+
+    OsRng
+        .try_fill_bytes(buffer)
+        .map_err(|error| format!("the operating system random source failed: {error}"))
 }
@@ -27,6 +27,8 @@

 pub mod identity;
 pub mod identity_store;
+pub mod offline;

 pub use identity::{DeviceIdentity, IdentityError, RegistrationProof, RegistrationTranscript};
 pub use identity_store::{IdentityStore, StoreError};
+pub use offline::{EnrollmentError, OfflineEnrollment, OfflineKeyStore, VerifiedChallenge};
@@ -0,0 +1,684 @@
+// 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.
+
+//! Challenge verification and response production for offline enrolment.
+//!
+//! Two invariants carry the security of this surface and both are easy to break
+//! by accident:
+//!
+//! - Every signature is checked over the octets that arrived, never over a
+//!   re-serialised document. Parsing happens only to route the verification, and
+//!   nothing a parse yields is believed until the signature over those same
+//!   octets has verified.
+//! - The enrolment root is the constant in this file. It is never taken from a
+//!   challenge, a configuration file, or an operator prompt, so there is no
+//!   trust-on-first-use path an operator could be talked into.
+//!
+//! The order of the checks in [`OfflineEnrollment::verify_challenge`] is frozen
+//! by `verificationOrder.enrollmentChallenge` in
+//! `protocol/agent/v1/fixtures/offline-enrollment/trust-model.json`, and the
+//! signature encoding, the domain separation tags, and every rejection reason
+//! are frozen beside it. Reordering the checks changes which reason a given
+//! artifact produces, which is itself part of the contract.
+
+use base64::Engine as _;
+use base64::engine::general_purpose::{STANDARD as BASE64_STANDARD, URL_SAFE_NO_PAD as BASE64_URL_NO_PAD};
+use p256::ecdsa::signature::{Signer as _, Verifier as _};
+use p256::ecdsa::{Signature, SigningKey, VerifyingKey};
+use p256::pkcs8::DecodePrivateKey as _;
+use serde::{Deserialize, Serialize};
+use sha2::{Digest as _, Sha256};
+use time::{Date, Month, OffsetDateTime, PrimitiveDateTime, Time};
+
+use crate::connect::identity::DeviceIdentity;
+
+/// The hosted enrolment root, compiled in. Both halves are pinned: the
+/// fingerprint identifies the root, and the point is what actually verifies the
+/// first link, so a build cannot be pointed at a different key by supplying one.
+const PINNED_ROOT_KEY_ID: &str = "df22e2806112debbe953672aafa186d699af0e97dd3fd2b09fa8359005fe348f";
+const PINNED_ROOT_PUBLIC_KEY: &str = "BFfx-K-FfEA5nK_Rz3IHacvRCkJyQ7JOd1geLyU6HKRZDgNezmVuKhvJ22VhemyjV__Gshk8JGGqOBzYPMD0p6s";
+
+/// Domain separation tags. A document that verifies under one of these must not
+/// be accepted for another artifact type, so the tag is part of the signature
+/// input rather than a property of the caller.
+const TAG_TRUST_LINK: &[u8] = b"rustfs-offline-trust-link-v1";
+const TAG_CHALLENGE: &[u8] = b"rustfs-offline-enrollment-challenge-v1";
+const TAG_RESPONSE: &[u8] = b"rustfs-offline-enrollment-response-v1";
+
+/// The single octet between the tag and the signed document.
+const DOMAIN_SEPARATOR: u8 = 0x00;
+
+const SIGNATURE_ALGORITHM: &str = "ES256";
+const PROTOCOL_VERSION: &str = "v1";
+const FORMAT_TRUST_LINK: &str = "rustfs.connect.offline.trustLink/1";
+const FORMAT_CHALLENGE: &str = "rustfs.connect.offline.enrollmentChallenge/1";
+const FORMAT_RESPONSE: &str = "rustfs.connect.offline.enrollmentResponse/1";
+
+/// DER SubjectPublicKeyInfo header for an uncompressed P-256 point. A keyId is
+/// the SHA-256 of this prefix followed by the 65 octet point, so the prefix is
+/// also how a device public key is recovered from its own DER encoding.
+const SPKI_PREFIX: [u8; 26] = [
+    0x30, 0x59, 0x30, 0x13, 0x06, 0x07, 0x2a, 0x86, 0x48, 0xce, 0x3d, 0x02, 0x01, 0x06, 0x08, 0x2a, 0x86, 0x48, 0xce, 0x3d, 0x03,
+    0x01, 0x07, 0x03, 0x42, 0x00,
+];
+
+/// Order of the P-256 group, and half of it. `r` and `s` must lie in `[1, n)`,
+/// and `s` additionally in `[1, n/2]`: ECDSA admits both `s` and `n - s`, and a
+/// signature with two spellings cannot serve as an artifact identity. Every
+/// ECDSA library accepts the malleated form, so the encoding layer rejects it.
+const GROUP_ORDER: [u8; 32] = [
+    0xff, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xbc, 0xe6, 0xfa, 0xad, 0xa7,
+    0x17, 0x9e, 0x84, 0xf3, 0xb9, 0xca, 0xc2, 0xfc, 0x63, 0x25, 0x51,
+];
+const MAX_S: [u8; 32] = [
+    0x7f, 0xff, 0xff, 0xff, 0x80, 0x00, 0x00, 0x00, 0x7f, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xde, 0x73, 0x7d, 0x56, 0xd3,
+    0x8b, 0xcf, 0x42, 0x79, 0xdc, 0xe5, 0x61, 0x7e, 0x31, 0x92, 0xa8,
+];
+
+const SCALAR_OCTETS: usize = 32;
+const SIGNATURE_OCTETS: usize = 64;
+/// 64 octets as unpadded base64url. The length is checked before decoding so
+/// that `=` padding, the standard alphabet, DER, and a truncated value are all
+/// refused rather than repaired.
+const SIGNATURE_VALUE_CHARS: usize = 86;
+
+const PUBLIC_KEY_OCTETS: usize = 65;
+const PUBLIC_KEY_CHARS: usize = 87;
+/// SEC1 tag of an uncompressed point. Compressed and hybrid forms are refused.
+const UNCOMPRESSED_POINT: u8 = 0x04;
+
+const TIMESTAMP_CHARS: usize = 20;
+
+/// The chain is exactly two links: a pinned root issues the intermediate, and
+/// the intermediate issues the signing key. Roles are positional and the
+/// enumeration is closed.
+const CHAIN_LINK_COUNT: usize = 2;
+const CHAIN_ROLES: [&str; CHAIN_LINK_COUNT] = ["intermediate", "signing"];
+
+/// Skew allowed on the challenge window. A device may have no synchronised
+/// clock at all, so its own reading of "now" is advisory.
+const CLOCK_SKEW_TOLERANCE: i64 = 300;
+
+/// Longest life a challenge may claim. The issuer sets both ends of its own
+/// window, so the protocol bound is applied on top of the declared expiry
+/// rather than trusted from it.
+const MAX_CHALLENGE_LIFETIME: i64 = 604_800;
+
+/// A challenge that verified, with the fields the response has to echo.
+///
+/// Construction is the proof: a value of this type only exists after the chain
+/// closed on the pinned root and the challenge signature verified over the
+/// received octets.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct VerifiedChallenge {
+    pub challenge_id: String,
+    pub organization_name: String,
+    pub cluster_name: String,
+    pub nonce: String,
+    pub issued_at: String,
+    pub expires_at: String,
+    pub connect_key_id: String,
+    /// The signature value of the challenge, verbatim. It binds a response to
+    /// the one challenge it answers, so it is carried rather than recomputed.
+    pub challenge_proof: String,
+}
+
+/// Why an offline enrolment artifact was refused.
+///
+/// The variants are the frozen `reason` vocabulary of
+/// `fixtures/offline-enrollment/error-codes.json`, which spans both halves of
+/// the exchange. The device half implemented here produces the encoding, chain,
+/// version, and freshness reasons; the reasons that describe a response being
+/// evaluated against stored state — [`Self::ChallengeUnknown`],
+/// [`Self::ChallengeProofInvalid`], [`Self::DeviceProofInvalid`],
+/// [`Self::EnrollmentReplayed`], [`Self::OrganizationMismatch`], and
+/// [`Self::ClusterMismatch`] — are Connect's to raise and are named here so the
+/// two sides share one vocabulary.
+///
+/// No variant carries a payload: a rejection must never disclose key material,
+/// signature octets, nonces, or document bytes.
+#[derive(Debug, PartialEq, Eq, thiserror::Error)]
+pub enum EnrollmentError {
+    #[error("protocolVersion is missing, malformed, or names an unsupported major version")]
+    UnsupportedProtocol,
+
+    #[error("formatVersion is not a supported offline enrollment format")]
+    UnsupportedFormat,
+
+    #[error("the signature is not 64 octets of fixed-width r||s in unpadded base64url")]
+    SignatureMalformed,
+
+    #[error("the signature is not in its canonical low-S form")]
+    SignatureNotCanonical,
+
+    #[error("the signature does not verify over the received octets")]
+    SignatureInvalid,
+
+    #[error("the trust chain is not issued by a root pinned in this build")]
+    EnrollmentRootUnknown,
+
+    #[error("a trust link is invalid, misordered, or outside its validity at the challenge issuedAt")]
+    TrustChainInvalid,
+
+    #[error("connectKeyId is not the subject of the last trust link")]
+    ConnectKeyUnchained,
+
+    #[error("no issued challenge matches this challengeId")]
+    ChallengeUnknown,
+
+    #[error("the challenge is not yet valid at the evaluation time")]
+    ChallengeNotYetValid,
+
+    #[error("the challenge has expired at the evaluation time")]
+    ChallengeExpired,
+
+    #[error("the response nonce or challengeProof is not the one issued for this challenge")]
+    ChallengeProofInvalid,
+
+    #[error("the response does not prove possession of the device key it presents")]
+    DeviceProofInvalid,
+
+    #[error("the challenge was already consumed")]
+    EnrollmentReplayed,
+
+    #[error("the response names a different organization than the challenge it answers")]
+    OrganizationMismatch,
+
+    #[error("the response names a different cluster than the challenge it answers")]
+    ClusterMismatch,
+
+    /// The artifact could not be read as a signed enrolment document at all: the
+    /// envelope, the base64 of the signed octets, or a field the frozen order
+    /// reads before the signature verifies did not parse. The frozen reason set
+    /// has no code for a structurally unreadable document, so this variant maps
+    /// to none of them.
+    #[error("the offline enrollment document is not well formed")]
+    MalformedDocument,
+
+    /// A fault on this side of the exchange rather than in the artifact: the
+    /// device key did not round-trip through its own PKCS#8 encoding, or the
+    /// caller named an instant outside the representable calendar. Fails closed
+    /// because a half-produced response must never reach removable media.
+    #[error("the enrollment response could not be produced on this device")]
+    ResponseNotProduced,
+}
+
+impl EnrollmentError {
+    /// The frozen `reason` an operator and Connect both branch on.
+    ///
+    /// The `Display` message is prose and may be reworded; this is the stable
+    /// identifier, so nothing should parse the message instead. The two
+    /// variants with no frozen counterpart deliberately return codes outside
+    /// the frozen set rather than borrowing the nearest one, so a document that
+    /// simply failed to parse can never be reported as a signature or freshness
+    /// failure.
+    pub fn reason(&self) -> &'static str {
+        match self {
+            Self::UnsupportedProtocol => "UNSUPPORTED_PROTOCOL",
+            Self::UnsupportedFormat => "UNSUPPORTED_FORMAT",
+            Self::SignatureMalformed => "SIGNATURE_MALFORMED",
+            Self::SignatureNotCanonical => "SIGNATURE_NOT_CANONICAL",
+            Self::SignatureInvalid => "SIGNATURE_INVALID",
+            Self::EnrollmentRootUnknown => "ENROLLMENT_ROOT_UNKNOWN",
+            Self::TrustChainInvalid => "TRUST_CHAIN_INVALID",
+            Self::ConnectKeyUnchained => "CONNECT_KEY_UNCHAINED",
+            Self::ChallengeUnknown => "CHALLENGE_UNKNOWN",
+            Self::ChallengeNotYetValid => "CHALLENGE_NOT_YET_VALID",
+            Self::ChallengeExpired => "CHALLENGE_EXPIRED",
+            Self::ChallengeProofInvalid => "CHALLENGE_PROOF_INVALID",
+            Self::DeviceProofInvalid => "DEVICE_PROOF_INVALID",
+            Self::EnrollmentReplayed => "ENROLLMENT_REPLAYED",
+            Self::OrganizationMismatch => "ORGANIZATION_MISMATCH",
+            Self::ClusterMismatch => "CLUSTER_MISMATCH",
+            Self::MalformedDocument => "MALFORMED_DOCUMENT",
+            Self::ResponseNotProduced => "RESPONSE_NOT_PRODUCED",
+        }
+    }
+}
+
+/// A signed document, in the shape both directions carry it. `bytes` is
+/// standard padded base64 of the exact octets that were signed; nothing else is
+/// ever used as the signature input.
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct SignedDocument {
+    bytes: String,
+    signature: DocumentSignature,
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct DocumentSignature {
+    algorithm: String,
+    key_id: String,
+    value: String,
+}
+
+/// The three fields the frozen order permits reading before anything verifies.
+/// They route the verification and are not facts until it has.
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ChallengeRouting {
+    connect_key_id: String,
+    issued_at: String,
+    trust_chain: Vec<SignedDocument>,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ChallengeDocument {
+    format_version: String,
+    protocol_version: String,
+    challenge_id: String,
+    organization_name: String,
+    cluster_name: String,
+    nonce: String,
+    issued_at: String,
+    expires_at: String,
+    connect_key_id: String,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct TrustLink {
+    format_version: String,
+    protocol_version: String,
+    role: String,
+    issuer_key_id: String,
+    subject_key_id: String,
+    subject_public_key: String,
+    not_before: String,
+    not_after: String,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct ResponseDocument<'a> {
+    format_version: &'a str,
+    protocol_version: &'a str,
+    challenge_id: &'a str,
+    organization_name: &'a str,
+    cluster_name: &'a str,
+    challenge_nonce: &'a str,
+    challenge_proof: &'a str,
+    device_key_id: String,
+    device_public_key: String,
+    device_nonce: String,
+    produced_at: String,
+}
+
+/// The device half of the offline enrolment exchange: bytes in, bytes out.
+pub struct OfflineEnrollment;
+
+impl OfflineEnrollment {
+    /// Verify an enrolment challenge and return what a response must echo.
+    ///
+    /// `now_unix` is the device's reading of the current time, which the clock
+    /// skew tolerance treats as advisory.
+    pub fn verify_challenge(document: &[u8], now_unix: i64) -> Result<VerifiedChallenge, EnrollmentError> {
+        let envelope: SignedDocument = serde_json::from_slice(document).map_err(|_| EnrollmentError::MalformedDocument)?;
+
+        // Step 1: the encoding is checked before anything is decoded from it, so
+        // a DER, padded, truncated, out-of-range, or high-S signature is refused
+        // on its spelling rather than handed to a library that would accept it.
+        let signature = decode_signature(&envelope.signature)?;
+
+        // The octets that were transmitted. They are never re-serialised: every
+        // later step signs and parses this same buffer.
+        let bytes = BASE64_STANDARD
+            .decode(envelope.bytes.as_bytes())
+            .map_err(|_| EnrollmentError::MalformedDocument)?;
+
+        // Step 2: routing only.
+        let routing: ChallengeRouting = serde_json::from_slice(&bytes).map_err(|_| EnrollmentError::MalformedDocument)?;
+        let issued_at = parse_timestamp(&routing.issued_at)?;
+
+        // Steps 3 to 5.
+        let connect_key = verify_trust_chain(&routing.trust_chain, &routing.connect_key_id, issued_at)?;
+
+        // Step 6. The verification key comes from the chain, so `signature.keyId`
+        // is a label rather than an input: a value naming some other key simply
+        // fails to verify here.
+        if !verifies(&connect_key, TAG_CHALLENGE, &bytes, &signature) {
+            return Err(EnrollmentError::SignatureInvalid);
+        }
+
+        // Step 7: only now is the document read as a fact.
+        let challenge: ChallengeDocument = serde_json::from_slice(&bytes).map_err(|_| EnrollmentError::MalformedDocument)?;
+        if challenge.protocol_version != PROTOCOL_VERSION {
+            return Err(EnrollmentError::UnsupportedProtocol);
+        }
+        if challenge.format_version != FORMAT_CHALLENGE {
+            return Err(EnrollmentError::UnsupportedFormat);
+        }
+
+        // Step 8.
+        let expires_at = parse_timestamp(&challenge.expires_at)?;
+        check_challenge_window(issued_at, expires_at, now_unix)?;
+
+        Ok(VerifiedChallenge {
+            challenge_id: challenge.challenge_id,
+            organization_name: challenge.organization_name,
+            cluster_name: challenge.cluster_name,
+            nonce: challenge.nonce,
+            issued_at: challenge.issued_at,
+            expires_at: challenge.expires_at,
+            connect_key_id: challenge.connect_key_id,
+            challenge_proof: envelope.signature.value,
+        })
+    }
+
+    /// Build the signed response an operator carries back to Connect.
+    ///
+    /// `device_nonce` is the response's own replay value and must come from a
+    /// cryptographic source. The private key never appears in the result: only
+    /// the public point, its fingerprint, and a signature over the document
+    /// that presents them, which is what makes presenting the key safe.
+    pub fn build_response(
+        challenge: &VerifiedChallenge,
+        key: &DeviceIdentity,
+        device_nonce: &[u8; 32],
+        produced_at_unix: i64,
+    ) -> Result<Vec<u8>, EnrollmentError> {
+        let issued_at = parse_timestamp(&challenge.issued_at)?;
+        let expires_at = parse_timestamp(&challenge.expires_at)?;
+        // Connect re-checks producedAt against the same window, so a response
+        // outside it is refused here rather than written to media and rejected
+        // after the operator has carried it out.
+        check_challenge_window(issued_at, expires_at, produced_at_unix)?;
+
+        let point = device_public_point(key)?;
+        let produced_at = format_timestamp(produced_at_unix)?;
+
+        let document = ResponseDocument {
+            format_version: FORMAT_RESPONSE,
+            protocol_version: PROTOCOL_VERSION,
+            challenge_id: &challenge.challenge_id,
+            organization_name: &challenge.organization_name,
+            cluster_name: &challenge.cluster_name,
+            challenge_nonce: &challenge.nonce,
+            challenge_proof: &challenge.challenge_proof,
+            device_key_id: key_id(&point),
+            device_public_key: BASE64_URL_NO_PAD.encode(point),
+            device_nonce: BASE64_URL_NO_PAD.encode(device_nonce),
+            produced_at,
+        };
+
+        // Serialised once. These octets are what is signed and what is carried,
+        // so no second serialisation can disagree with the signature.
+        let bytes = serde_json::to_vec(&document).map_err(|_| EnrollmentError::ResponseNotProduced)?;
+        let signature = sign(key, TAG_RESPONSE, &bytes)?;
+
+        let envelope = SignedDocument {
+            bytes: BASE64_STANDARD.encode(&bytes),
+            signature: DocumentSignature {
+                algorithm: SIGNATURE_ALGORITHM.to_owned(),
+                key_id: document.device_key_id,
+                value: signature,
+            },
+        };
+
+        serde_json::to_vec(&envelope).map_err(|_| EnrollmentError::ResponseNotProduced)
+    }
+}
+
+/// Walk the chain from the pinned root to the signing key, returning the key
+/// `connect_key_id` names once the chain vouches for it.
+fn verify_trust_chain(
+    chain: &[SignedDocument],
+    connect_key_id: &str,
+    challenge_issued_at: i64,
+) -> Result<VerifyingKey, EnrollmentError> {
+    // The pinned root gate runs before the chain's shape is examined, so a
+    // chain that is internally consistent under a foreign root — exactly what
+    // trust on first use would have accepted — is refused for its root rather
+    // than for its length.
+    let first = chain.first().ok_or(EnrollmentError::EnrollmentRootUnknown)?;
+    let root = decode_trust_link(first)?;
+    if root.0.issuer_key_id != PINNED_ROOT_KEY_ID {
+        return Err(EnrollmentError::EnrollmentRootUnknown);
+    }
+
+    let [_, second] = chain else {
+        return Err(EnrollmentError::TrustChainInvalid);
+    };
+    let links = [root, decode_trust_link(second)?];
+
+    let mut issuer_key_id = PINNED_ROOT_KEY_ID.to_owned();
+    let (mut issuer_key, _) = decode_public_key(PINNED_ROOT_PUBLIC_KEY).ok_or(EnrollmentError::EnrollmentRootUnknown)?;
+
+    for (index, ((link, link_bytes), entry)) in links.iter().zip(chain).enumerate() {
+        if link.format_version != FORMAT_TRUST_LINK
+            || link.protocol_version != PROTOCOL_VERSION
+            || link.role != CHAIN_ROLES[index]
+            || link.issuer_key_id != issuer_key_id
+            // A link that names itself as its own issuer would let a stolen
+            // intermediate mint its own root.
+            || link.subject_key_id == link.issuer_key_id
+        {
+            return Err(EnrollmentError::TrustChainInvalid);
+        }
+
+        let (subject_key, subject_point) =
+            decode_public_key(&link.subject_public_key).ok_or(EnrollmentError::TrustChainInvalid)?;
+        if key_id(&subject_point) != link.subject_key_id {
+            return Err(EnrollmentError::TrustChainInvalid);
+        }
+
+        let signature = decode_signature(&entry.signature)?;
+        if !verifies(&issuer_key, TAG_TRUST_LINK, link_bytes, &signature) {
+            return Err(EnrollmentError::TrustChainInvalid);
+        }
+
+        // The issuer controls both ends of a link's window, so it is evaluated
+        // with no skew tolerance, and against the challenge's issuedAt rather
+        // than against the device clock: a challenge carries the chain that was
+        // valid when it was issued.
+        let not_before = parse_timestamp(&link.not_before)?;
+        let not_after = parse_timestamp(&link.not_after)?;
+        if challenge_issued_at < not_before || challenge_issued_at > not_after {
+            return Err(EnrollmentError::TrustChainInvalid);
+        }
+
+        issuer_key_id = link.subject_key_id.clone();
+        issuer_key = subject_key;
+    }
+
+    if issuer_key_id != connect_key_id {
+        return Err(EnrollmentError::ConnectKeyUnchained);
+    }
+
+    Ok(issuer_key)
+}
+
+/// Decode a link and keep the octets it was signed over: the signature is
+/// checked against these, never against a re-encoding of the parsed link.
+fn decode_trust_link(entry: &SignedDocument) -> Result<(TrustLink, Vec<u8>), EnrollmentError> {
+    let bytes = BASE64_STANDARD
+        .decode(entry.bytes.as_bytes())
+        .map_err(|_| EnrollmentError::MalformedDocument)?;
+    let link = serde_json::from_slice(&bytes).map_err(|_| EnrollmentError::TrustChainInvalid)?;
+    Ok((link, bytes))
+}
+
+/// Check a signature's spelling and range, then admit it.
+///
+/// `r` and `s` are compared against the group order here rather than left to
+/// the ECDSA library, because a library that accepts high-S — every library
+/// does — would let a malleated copy of an artifact pass as a second artifact.
+fn decode_signature(signature: &DocumentSignature) -> Result<Signature, EnrollmentError> {
+    if signature.algorithm != SIGNATURE_ALGORITHM {
+        return Err(EnrollmentError::SignatureMalformed);
+    }
+
+    let value = signature.value.as_bytes();
+    if value.len() != SIGNATURE_VALUE_CHARS || !value.iter().all(|byte| is_base64url(*byte)) {
+        return Err(EnrollmentError::SignatureMalformed);
+    }
+
+    let decoded = BASE64_URL_NO_PAD
+        .decode(value)
+        .map_err(|_| EnrollmentError::SignatureMalformed)?;
+    let octets: [u8; SIGNATURE_OCTETS] = decoded
+        .as_slice()
+        .try_into()
+        .map_err(|_| EnrollmentError::SignatureMalformed)?;
+
+    // Big-endian octets of equal length order lexicographically exactly as the
+    // integers they spell, so a slice comparison is the range check.
+    let (r, s) = octets.split_at(SCALAR_OCTETS);
+    let out_of_range = |scalar: &[u8]| scalar.iter().all(|byte| *byte == 0) || scalar >= &GROUP_ORDER[..];
+    if out_of_range(r) || out_of_range(s) {
+        return Err(EnrollmentError::SignatureMalformed);
+    }
+    if s > &MAX_S[..] {
+        return Err(EnrollmentError::SignatureNotCanonical);
+    }
+
+    Signature::from_slice(&octets).map_err(|_| EnrollmentError::SignatureMalformed)
+}
+
+fn verifies(key: &VerifyingKey, tag: &[u8], bytes: &[u8], signature: &Signature) -> bool {
+    key.verify(&signature_input(tag, bytes), signature).is_ok()
+}
+
+fn signature_input(tag: &[u8], bytes: &[u8]) -> Vec<u8> {
+    let mut input = Vec::with_capacity(tag.len() + 1 + bytes.len());
+    input.extend_from_slice(tag);
+    input.push(DOMAIN_SEPARATOR);
+    input.extend_from_slice(bytes);
+    input
+}
+
+fn sign(key: &DeviceIdentity, tag: &[u8], bytes: &[u8]) -> Result<String, EnrollmentError> {
+    // `DeviceIdentity` publishes no general signing operation, so the key is
+    // rebuilt from its own PKCS#8 encoding; the encoding is wiped when the
+    // wrapper drops.
+    let pkcs8 = key.to_pkcs8_der().map_err(|_| EnrollmentError::ResponseNotProduced)?;
+    let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| EnrollmentError::ResponseNotProduced)?;
+
+    let signature: Signature = signing_key.sign(&signature_input(tag, bytes));
+    let canonical = signature.normalize_s().unwrap_or(signature);
+
+    Ok(BASE64_URL_NO_PAD.encode(canonical.to_bytes()))
+}
+
+/// The device's public point, recovered from the DER encoding the identity
+/// publishes so that one prefix constant governs both the fingerprint and the
+/// wire form.
+fn device_public_point(key: &DeviceIdentity) -> Result<[u8; PUBLIC_KEY_OCTETS], EnrollmentError> {
+    key.public_key_der()
+        .strip_prefix(&SPKI_PREFIX)
+        .and_then(|point| <[u8; PUBLIC_KEY_OCTETS]>::try_from(point).ok())
+        .ok_or(EnrollmentError::ResponseNotProduced)
+}
+
+/// Decode an uncompressed SEC1 point and check that it is on the curve.
+///
+/// The length and alphabet are checked before decoding so that a padded or
+/// standard-alphabet spelling is refused, and the point tag is checked so that
+/// the compressed and hybrid forms — which no keyId would match — cannot be
+/// spelled at all.
+fn decode_public_key(value: &str) -> Option<(VerifyingKey, [u8; PUBLIC_KEY_OCTETS])> {
+    let value = value.as_bytes();
+    if value.len() != PUBLIC_KEY_CHARS || !value.iter().all(|byte| is_base64url(*byte)) {
+        return None;
+    }
+
+    let point: [u8; PUBLIC_KEY_OCTETS] = BASE64_URL_NO_PAD.decode(value).ok()?.try_into().ok()?;
+    if point[0] != UNCOMPRESSED_POINT {
+        return None;
+    }
+
+    VerifyingKey::from_sec1_bytes(&point).ok().map(|key| (key, point))
+}
+
+/// Lowercase SHA-256 hex of the DER SubjectPublicKeyInfo built from a 65 octet
+/// uncompressed point.
+fn key_id(point: &[u8]) -> String {
+    let mut digest = Sha256::new();
+    digest.update(SPKI_PREFIX);
+    digest.update(point);
+    hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower)
+}
+
+fn is_base64url(byte: u8) -> bool {
+    byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_'
+}
+
+/// Parse `YYYY-MM-DDTHH:MM:SSZ` into a Unix instant.
+///
+/// The shape is checked before the fields are read: offsets other than `Z` and
+/// fractional seconds are refused rather than normalised, so two producers
+/// cannot spell the same instant two ways.
+fn parse_timestamp(value: &str) -> Result<i64, EnrollmentError> {
+    let octets = value.as_bytes();
+    if octets.len() != TIMESTAMP_CHARS
+        || octets[4] != b'-'
+        || octets[7] != b'-'
+        || octets[10] != b'T'
+        || octets[13] != b':'
+        || octets[16] != b':'
+        || octets[19] != b'Z'
+    {
+        return Err(EnrollmentError::MalformedDocument);
+    }
+
+    let field = |range: std::ops::Range<usize>| -> Result<u32, EnrollmentError> {
+        let text = &value[range];
+        if !text.bytes().all(|byte| byte.is_ascii_digit()) {
+            return Err(EnrollmentError::MalformedDocument);
+        }
+        text.parse().map_err(|_| EnrollmentError::MalformedDocument)
+    };
+
+    let month = Month::try_from(field(5..7)? as u8).map_err(|_| EnrollmentError::MalformedDocument)?;
+    let date = Date::from_calendar_date(field(0..4)? as i32, month, field(8..10)? as u8)
+        .map_err(|_| EnrollmentError::MalformedDocument)?;
+    let clock = Time::from_hms(field(11..13)? as u8, field(14..16)? as u8, field(17..19)? as u8)
+        .map_err(|_| EnrollmentError::MalformedDocument)?;
+
+    Ok(PrimitiveDateTime::new(date, clock).assume_utc().unix_timestamp())
+}
+
+fn format_timestamp(unix: i64) -> Result<String, EnrollmentError> {
+    let moment = OffsetDateTime::from_unix_timestamp(unix).map_err(|_| EnrollmentError::ResponseNotProduced)?;
+    Ok(format!(
+        "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z",
+        moment.year(),
+        u8::from(moment.month()),
+        moment.day(),
+        moment.hour(),
+        moment.minute(),
+        moment.second()
+    ))
+}
+
+/// `at` must fall within `[issuedAt - 300, expiresAt + 300]`.
+///
+/// The declared expiry is capped at the protocol's maximum challenge lifetime
+/// because the issuer sets both ends of its own window; a challenge claiming a
+/// longer life expires at the bound.
+fn check_challenge_window(issued_at: i64, expires_at: i64, at: i64) -> Result<(), EnrollmentError> {
+    if at < issued_at.saturating_sub(CLOCK_SKEW_TOLERANCE) {
+        return Err(EnrollmentError::ChallengeNotYetValid);
+    }
+
+    let effective_expiry = expires_at.min(issued_at.saturating_add(MAX_CHALLENGE_LIFETIME));
+    if at > effective_expiry.saturating_add(CLOCK_SKEW_TOLERANCE) {
+        return Err(EnrollmentError::ChallengeExpired);
+    }
+
+    Ok(())
+}
@@ -0,0 +1,72 @@
+// 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.
+
+//! On-disk home of the offline enrollment key.
+//!
+//! An air-gapped device enrols with a key that is not its online device
+//! identity: the online key is minted during a registration exchange this
+//! device cannot perform, and an operator who carries an enrolment response out
+//! on removable media is enrolling exactly one key that Connect will pin. Losing
+//! it means asking for a fresh challenge, so it is written durably and published
+//! exactly once.
+//!
+//! The durability protocol is not reimplemented here. [`IdentityStore`] already
+//! seals a P-256 key at mode 0600, fsyncs it, and publishes it through a
+//! no-clobber link so a retry or a concurrent start converges on one key; it is
+//! pointed at a directory of this key's own rather than generalised into a
+//! key-store abstraction that would have to describe both lifecycles.
+
+use std::path::{Path, PathBuf};
+
+use super::super::identity::DeviceIdentity;
+use super::super::identity_store::{IdentityStore, StoreError};
+
+/// Subdirectory holding the offline enrolment key, kept apart from the online
+/// device identity so neither can be read in place of the other.
+const OFFLINE_DIRECTORY: &str = "offline";
+
+/// The offline enrolment key of one deployment.
+#[derive(Clone, Debug)]
+pub struct OfflineKeyStore {
+    inner: IdentityStore,
+}
+
+impl OfflineKeyStore {
+    pub fn new(directory: impl AsRef<Path>) -> Self {
+        Self {
+            inner: IdentityStore::new(directory.as_ref().join(OFFLINE_DIRECTORY)),
+        }
+    }
+
+    pub fn key_path(&self) -> PathBuf {
+        self.inner.key_path()
+    }
+
+    /// Return the stored key, or `None` when this deployment has never enrolled
+    /// offline. Reading never creates one, so a deployment that only ever
+    /// registers online holds no offline key.
+    pub fn load(&self) -> Result<Option<DeviceIdentity>, StoreError> {
+        self.inner.load()
+    }
+
+    /// Return the stored key, generating and publishing one the first time.
+    ///
+    /// A second enrolment attempt returns the original key rather than minting a
+    /// replacement: the operator may already be carrying a response for it, and
+    /// two keys would mean the response and the device disagree about which one
+    /// Connect pinned.
+    pub fn load_or_create(&self) -> Result<DeviceIdentity, StoreError> {
+        self.inner.load_or_create()
+    }
+}
@@ -0,0 +1,35 @@
+// 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.
+
+//! Offline enrolment: joining a Connect tenant without a network.
+//!
+//! An air-gapped cluster cannot perform the registration exchange, so an
+//! operator carries a signed challenge in and a signed response out. The device
+//! half of that exchange lives here: verifying the challenge against a root
+//! whose fingerprint is compiled into this binary, minting the key being
+//! enrolled, and signing the response.
+//!
+//! Nothing here opens a socket. That is the point of the surface, and it is
+//! asserted rather than assumed: the enrolment path takes bytes and returns
+//! bytes.
+//!
+//! The trust model, the signing convention, and every rejection reason are
+//! frozen by `protocol/agent/v1/fixtures/offline-enrollment/` and by
+//! `docs/adr/0009-offline-signing.md` on the Connect side.
+
+pub mod enrollment;
+pub mod key_store;
+
+pub use enrollment::{EnrollmentError, OfflineEnrollment, VerifiedChallenge};
+pub use key_store::OfflineKeyStore;
@@ -0,0 +1,951 @@
+// 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.
+
+//! Offline enrollment conformance against the frozen Connect fixtures.
+//!
+//! The device half of the air-gapped exchange verifies a challenge Connect
+//! signed and produces a response Connect will verify. Neither side can talk to
+//! the other while it does so, which means every disagreement about encoding,
+//! trust, or clock windows surfaces as a failed enrollment in the field rather
+//! than as an error at development time. The fixtures under
+//! `protocol/agent/v1/fixtures/offline-enrollment/` are the shared statement of
+//! what both sides must do, so this suite replays them rather than restating
+//! them: accept vectors must be accepted with the fields the document carries,
+//! reject vectors must fail with the single reason `error-codes.json` freezes,
+//! and the signature encoding rules in `trust-model.json` must hold even where
+//! the underlying ECDSA library is happy.
+
+use std::fs;
+use std::path::PathBuf;
+
+use base64::Engine as _;
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
+use base64::engine::general_purpose::URL_SAFE_NO_PAD as BASE64_URL_NO_PAD;
+use rustfs::connect::identity::DeviceIdentity;
+use rustfs::connect::offline::{EnrollmentError, OfflineEnrollment, VerifiedChallenge};
+use serde_json::Value;
+use sha2::{Digest as _, Sha256};
+
+/// DER prefix of a P-256 `SubjectPublicKeyInfo`, frozen by
+/// `trust-model.json` as `signature.subjectPublicKeyInfoDerPrefix`. The 65
+/// octet uncompressed point follows it, so a SEC1 point published in a fixture
+/// becomes a decodable public key by concatenation.
+const SPKI_PREFIX_HEX: &str = "3059301306072a8648ce3d020106082a8648ce3d030107034200";
+
+/// `clockSkew.toleranceSeconds` in `trust-model.json`.
+const SKEW_TOLERANCE_SECONDS: i64 = 300;
+
+// ---------------------------------------------------------------------------
+// Fixture access
+// ---------------------------------------------------------------------------
+
+fn fixture_dir() -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../protocol/agent/v1/fixtures/offline-enrollment")
+}
+
+fn sha256_hex(bytes: &[u8]) -> String {
+    Sha256::digest(bytes).iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
+/// Read one fixture file and refuse it unless its bytes match the digest
+/// `MANIFEST.sha256` freezes.
+///
+/// Every vector in this suite arrives through here. A fixture edited on this
+/// side therefore fails the tests that depend on it instead of quietly
+/// redefining what conformance means, which is the failure mode a
+/// fixture-driven suite is otherwise blind to.
+fn read_fixture(name: &str) -> Vec<u8> {
+    let dir = fixture_dir();
+    let manifest = fs::read_to_string(dir.join("MANIFEST.sha256")).expect("read MANIFEST.sha256");
+
+    let expected = manifest
+        .lines()
+        .filter(|line| !line.trim().is_empty())
+        .find_map(|line| {
+            let (digest, file) = line
+                .split_once("  ")
+                .unwrap_or_else(|| panic!("malformed manifest line: {line}"));
+            (file == name).then(|| digest.to_string())
+        })
+        .unwrap_or_else(|| panic!("{name} is not listed in MANIFEST.sha256"));
+
+    let bytes = fs::read(dir.join(name)).unwrap_or_else(|error| panic!("read {name}: {error}"));
+    assert_eq!(sha256_hex(&bytes), expected, "{name} does not match the digest MANIFEST.sha256 freezes");
+    bytes
+}
+
+fn fixture_json(name: &str) -> Value {
+    serde_json::from_slice(&read_fixture(name)).unwrap_or_else(|error| panic!("{name} parses: {error}"))
+}
+
+fn accept_vectors() -> Value {
+    fixture_json("accept-vectors.json")
+}
+
+fn reject_vectors() -> Value {
+    fixture_json("reject-vectors.json")
+}
+
+fn trust_model() -> Value {
+    fixture_json("trust-model.json")
+}
+
+fn vector_list(fixture: &Value) -> Vec<Value> {
+    fixture["vectors"].as_array().expect("fixture carries a vector list").clone()
+}
+
+fn field<'a>(value: &'a Value, key: &str) -> &'a str {
+    value[key]
+        .as_str()
+        .unwrap_or_else(|| panic!("expected a string at '{key}' in {value}"))
+}
+
+/// The octets an operator carries in on removable media.
+///
+/// The fixture's `document` object *is* the transmitted artifact: a padded
+/// base64 `bytes` field holding the raw signed octets, plus the detached
+/// signature over them. Only `bytes` is covered by the signature, so
+/// re-serialising the surrounding envelope here cannot change what a verifier
+/// checks.
+fn envelope(document: &Value) -> Vec<u8> {
+    serde_json::to_vec(document).expect("envelope serialises")
+}
+
+/// The raw octets the signature covers, exactly as transmitted.
+fn signed_octets(document: &Value) -> Vec<u8> {
+    BASE64_STANDARD
+        .decode(field(document, "bytes"))
+        .expect("document bytes are padded base64")
+}
+
+/// The parsed signed document. Parsing is a convenience for the assertions
+/// below; the implementation under test is required to verify before it parses.
+fn signed_document(document: &Value) -> Value {
+    serde_json::from_slice(&signed_octets(document)).expect("signed document parses")
+}
+
+fn unix(rfc3339: &str) -> i64 {
+    chrono::DateTime::parse_from_rfc3339(rfc3339)
+        .unwrap_or_else(|error| panic!("'{rfc3339}' is not RFC 3339: {error}"))
+        .timestamp()
+}
+
+fn hex_to_bytes(hex: &str) -> Vec<u8> {
+    (0..hex.len())
+        .step_by(2)
+        .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).expect("valid hex"))
+        .collect()
+}
+
+/// Turn a fixture's unpadded-base64url SEC1 point into a usable verifying key.
+fn verifying_key(sec1_base64url: &str) -> p256::ecdsa::VerifyingKey {
+    let point = BASE64_URL_NO_PAD.decode(sec1_base64url).expect("public key is base64url");
+    assert_eq!(point.len(), 65, "the protocol freezes a 65 octet uncompressed SEC1 point");
+
+    let mut der = hex_to_bytes(SPKI_PREFIX_HEX);
+    der.extend_from_slice(&point);
+    <p256::ecdsa::VerifyingKey as p256::pkcs8::DecodePublicKey>::from_public_key_der(&der).expect("public key decodes")
+}
+
+fn published_key(role_or_name: &str) -> Value {
+    fixture_json("trust-chain.json")["keys"]
+        .as_array()
+        .expect("trust chain publishes keys")
+        .iter()
+        .find(|key| field(key, "name") == role_or_name)
+        .unwrap_or_else(|| panic!("trust-chain.json publishes no key named '{role_or_name}'"))
+        .clone()
+}
+
+/// `signatureInput = domainSeparationTag || 0x00 || the received octets`, the
+/// rule `trust-model.json` freezes under `domainSeparation`.
+fn signing_input(artifact_tag: &str, received: &[u8]) -> Vec<u8> {
+    let mut input = artifact_tag.as_bytes().to_vec();
+    input.push(0x00);
+    input.extend_from_slice(received);
+    input
+}
+
+fn domain_tag(artifact: &str) -> String {
+    let model = trust_model();
+    assert_eq!(
+        field(&model["domainSeparation"], "separatorByte"),
+        "0x00",
+        "the separator byte this suite encodes is the one the trust model freezes"
+    );
+    field(&model["domainSeparation"]["tags"], artifact).to_string()
+}
+
+/// Locate an accept vector by the name other vectors reference it by.
+fn accept_vector_named(name: &str) -> Value {
+    vector_list(&accept_vectors())
+        .into_iter()
+        .find(|vector| field(vector, "name") == name)
+        .unwrap_or_else(|| panic!("accept-vectors.json carries no vector named '{name}'"))
+}
+
+/// Verify the challenge a response vector answers, at that challenge's own
+/// evaluation time.
+fn answered_challenge(response_vector: &Value) -> (Value, VerifiedChallenge) {
+    let challenge_vector = accept_vector_named(field(response_vector, "answersChallenge"));
+    let now = unix(field(&challenge_vector, "evaluationTime"));
+    let verified = OfflineEnrollment::verify_challenge(&envelope(&challenge_vector["document"]), now)
+        .expect("the answered challenge is an accept vector and must verify");
+    (challenge_vector, verified)
+}
+
+fn device_nonce_of(document: &Value) -> [u8; 32] {
+    let raw = BASE64_URL_NO_PAD
+        .decode(field(&signed_document(document), "deviceNonce"))
+        .expect("deviceNonce is base64url");
+    raw.try_into().expect("replay.nonceLengthBytes freezes a 32 octet nonce")
+}
+
+// ---------------------------------------------------------------------------
+// Accept vectors
+// ---------------------------------------------------------------------------
+
+/// Every challenge accept vector must verify at its own evaluation time and
+/// expose exactly what the signed document says.
+///
+/// Two of these vectors sit on the skew boundary — 120 seconds before
+/// `issuedAt` and 300 seconds after `expiresAt` — so a verifier that compares
+/// against the raw window instead of the tolerated one fails here rather than
+/// in an air-gapped data centre. `challenge_proof` is pinned to the challenge's
+/// own detached signature value because that is what the response has to echo;
+/// deriving it from anything else would silently break the binding.
+#[test]
+fn every_challenge_accept_vector_verifies_and_exposes_the_signed_fields() {
+    let mut verified_count = 0usize;
+
+    for vector in vector_list(&accept_vectors()) {
+        if field(&vector, "artifact") != "challenge" {
+            continue;
+        }
+
+        let name = field(&vector, "name");
+        let document = &vector["document"];
+        let now = unix(field(&vector, "evaluationTime"));
+
+        let verified = OfflineEnrollment::verify_challenge(&envelope(document), now)
+            .unwrap_or_else(|error| panic!("accept vector '{name}' must verify: {}", error.reason()));
+
+        let signed = signed_document(document);
+        assert_eq!(verified.challenge_id, field(&signed, "challengeId"), "vector '{name}' challengeId");
+        assert_eq!(
+            verified.organization_name,
+            field(&signed, "organizationName"),
+            "vector '{name}' organizationName"
+        );
+        assert_eq!(verified.cluster_name, field(&signed, "clusterName"), "vector '{name}' clusterName");
+        assert_eq!(verified.nonce, field(&signed, "nonce"), "vector '{name}' nonce");
+        assert_eq!(verified.issued_at, field(&signed, "issuedAt"), "vector '{name}' issuedAt");
+        assert_eq!(verified.expires_at, field(&signed, "expiresAt"), "vector '{name}' expiresAt");
+        assert_eq!(verified.connect_key_id, field(&signed, "connectKeyId"), "vector '{name}' connectKeyId");
+        assert_eq!(
+            verified.challenge_proof,
+            field(&document["signature"], "value"),
+            "vector '{name}' must carry the challenge's own signature as the proof a response echoes"
+        );
+
+        verified_count += 1;
+    }
+
+    assert_eq!(
+        verified_count, 3,
+        "accept-vectors.json publishes three challenge vectors; a fourth is a protocol change"
+    );
+}
+
+/// Connect's own producer wrote the response accept vectors. Rebuilding them
+/// from the challenge they answer, with the device nonce and production time
+/// they used, must reproduce every field that does not depend on which device
+/// key signed — including the discarded-unknown-field vector, whose extra
+/// `telemetryHint` must not survive into anything this side produces.
+#[test]
+fn response_accept_vectors_are_reproduced_field_for_field_by_build_response() {
+    let key = DeviceIdentity::generate();
+    let mut reproduced = 0usize;
+
+    for vector in vector_list(&accept_vectors()) {
+        if field(&vector, "artifact") != "response" {
+            continue;
+        }
+
+        let name = field(&vector, "name");
+        let published = signed_document(&vector["document"]);
+        let (_, challenge) = answered_challenge(&vector);
+
+        let produced_at = unix(field(&published, "producedAt"));
+        let nonce = device_nonce_of(&vector["document"]);
+
+        let built_envelope: Value = serde_json::from_slice(
+            &OfflineEnrollment::build_response(&challenge, &key, &nonce, produced_at)
+                .unwrap_or_else(|error| panic!("vector '{name}' must be reproducible: {}", error.reason())),
+        )
+        .expect("the built response is JSON");
+        let built = signed_document(&built_envelope);
+
+        for shared in [
+            "formatVersion",
+            "protocolVersion",
+            "challengeId",
+            "organizationName",
+            "clusterName",
+            "challengeNonce",
+            "challengeProof",
+            "deviceNonce",
+        ] {
+            assert_eq!(
+                field(&built, shared),
+                field(&published, shared),
+                "vector '{name}' field {shared} must match the response Connect published"
+            );
+        }
+        assert_eq!(
+            unix(field(&built, "producedAt")),
+            produced_at,
+            "vector '{name}' producedAt must be the instant it was given"
+        );
+
+        // `versioning.additive` says an unknown optional field is discarded and
+        // never echoed back; a producer that copied the challenge or a previous
+        // response wholesale would carry it forward.
+        assert!(
+            built.get("telemetryHint").is_none(),
+            "vector '{name}' must not echo an unknown optional field"
+        );
+
+        reproduced += 1;
+    }
+
+    assert_eq!(
+        reproduced, 2,
+        "accept-vectors.json publishes two response vectors; a third is a protocol change"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// Reject vectors
+// ---------------------------------------------------------------------------
+
+/// Every challenge reject vector must fail, and fail for the one reason
+/// `error-codes.json` freezes.
+///
+/// Asserting only that verification failed would pass for an implementation
+/// that rejects everything, and would let a tampered document be reported as an
+/// expiry — a rejection reason is what an operator acts on, so it is part of the
+/// contract rather than a diagnostic detail.
+#[test]
+fn every_challenge_reject_vector_fails_with_its_frozen_reason() {
+    let known_reasons: Vec<String> = fixture_json("error-codes.json")["reasons"]
+        .as_array()
+        .expect("error-codes.json carries reasons")
+        .iter()
+        .map(|entry| field(entry, "reason").to_string())
+        .collect();
+
+    let mut rejected = 0usize;
+
+    for vector in vector_list(&reject_vectors()) {
+        if field(&vector, "artifact") != "challenge" {
+            continue;
+        }
+
+        let name = field(&vector, "name");
+        let expected = field(&vector["expected"], "reason");
+        assert!(
+            known_reasons.iter().any(|reason| reason == expected),
+            "vector '{name}' names reason {expected}, which error-codes.json does not freeze"
+        );
+
+        let now = unix(field(&vector, "evaluationTime"));
+        let error = OfflineEnrollment::verify_challenge(&envelope(&vector["document"]), now)
+            .expect_err(&format!("reject vector '{name}' must not verify"));
+
+        assert_eq!(error.reason(), expected, "vector '{name}' must fail as {expected}");
+        rejected += 1;
+    }
+
+    assert_eq!(
+        rejected, 8,
+        "reject-vectors.json publishes eight challenge vectors; losing one silently narrows the suite"
+    );
+}
+
+/// The response reject vectors are artifacts Connect refuses. This side never
+/// verifies a response, so the device-side statement is the stronger one: given
+/// the challenge each vector answers, `build_response` must not be capable of
+/// emitting that artifact in the first place.
+///
+/// Each arm pins the specific field a compromised or careless producer would
+/// have to get wrong, so an implementation that copied values out of the wrong
+/// place — the response's own document, an operator-supplied argument, a
+/// previous exchange — fails here.
+#[test]
+fn response_reject_vectors_are_artifacts_build_response_cannot_emit() {
+    let key = DeviceIdentity::generate();
+    let mut covered = 0usize;
+
+    for vector in vector_list(&reject_vectors()) {
+        if field(&vector, "artifact") != "response" {
+            continue;
+        }
+
+        let name = field(&vector, "name");
+        let refused = signed_document(&vector["document"]);
+        let (_, challenge) = answered_challenge(&vector);
+        let produced_at = unix(field(&refused, "producedAt"));
+        let nonce = device_nonce_of(&vector["document"]);
+
+        let outcome = OfflineEnrollment::build_response(&challenge, &key, &nonce, produced_at);
+
+        match field(&vector["expected"], "reason") {
+            // `responseWindow` in trust-model.json: a device that emits a
+            // response outside the tolerated challenge window has produced an
+            // artifact Connect will refuse, so the refusal belongs here rather
+            // than at the far end of a courier run.
+            "CHALLENGE_EXPIRED" => {
+                let error = outcome.expect_err(&format!("vector '{name}': producing this response must be refused"));
+                assert_eq!(error.reason(), "CHALLENGE_EXPIRED", "vector '{name}' must refuse as CHALLENGE_EXPIRED");
+                covered += 1;
+                continue;
+            }
+            reason => {
+                let built_envelope: Value = serde_json::from_slice(
+                    &outcome.unwrap_or_else(|error| panic!("vector '{name}' baseline must build: {}", error.reason())),
+                )
+                .expect("the built response is JSON");
+                let built = signed_document(&built_envelope);
+
+                match reason {
+                    "ORGANIZATION_MISMATCH" => {
+                        assert_ne!(
+                            field(&refused, "organizationName"),
+                            challenge.organization_name,
+                            "vector '{name}' is only a mismatch if it names another organization"
+                        );
+                        assert_eq!(
+                            field(&built, "organizationName"),
+                            challenge.organization_name,
+                            "vector '{name}': the organization must come from the challenge, never from elsewhere"
+                        );
+                    }
+                    "CLUSTER_MISMATCH" => {
+                        assert_ne!(
+                            field(&refused, "clusterName"),
+                            challenge.cluster_name,
+                            "vector '{name}' is only a mismatch if it names another cluster"
+                        );
+                        assert_eq!(
+                            field(&built, "clusterName"),
+                            challenge.cluster_name,
+                            "vector '{name}': the cluster must come from the challenge, never from elsewhere"
+                        );
+                    }
+                    "CHALLENGE_PROOF_INVALID" => {
+                        // Two distinct vectors land here: a nonce the challenge
+                        // never carried, and a proof lifted from another
+                        // challenge. Both must be impossible to produce.
+                        assert_eq!(
+                            field(&built, "challengeNonce"),
+                            challenge.nonce,
+                            "vector '{name}': the echoed nonce must be the challenge's own"
+                        );
+                        assert_eq!(
+                            field(&built, "challengeProof"),
+                            challenge.challenge_proof,
+                            "vector '{name}': the proof must be the answered challenge's signature"
+                        );
+                        assert!(
+                            field(&refused, "challengeNonce") != challenge.nonce
+                                || field(&refused, "challengeProof") != challenge.challenge_proof,
+                            "vector '{name}' must differ from the challenge in nonce or proof to be rejectable"
+                        );
+                    }
+                    "DEVICE_PROOF_INVALID" => {
+                        // The refused vector presents one key and is signed by
+                        // another; hold the fixture to that claim, then require
+                        // the built response to be the opposite. Proof of
+                        // possession is the only thing that makes presenting a
+                        // key in an unauthenticated document safe.
+                        use p256::ecdsa::signature::Verifier as _;
+
+                        let presented = verifying_key(field(&refused, "devicePublicKey"));
+                        let raw = BASE64_URL_NO_PAD
+                            .decode(field(&vector["document"]["signature"], "value"))
+                            .expect("signature is base64url");
+                        let signature = p256::ecdsa::Signature::from_slice(&raw).expect("signature parses");
+                        assert!(
+                            presented
+                                .verify(
+                                    &signing_input(&domain_tag("enrollmentResponse"), &signed_octets(&vector["document"])),
+                                    &signature
+                                )
+                                .is_err(),
+                            "vector '{name}' is only a possession failure if it does not verify under the key it presents"
+                        );
+
+                        assert_response_proves_possession(&built_envelope, name);
+                    }
+                    "UNSUPPORTED_FORMAT" => {
+                        assert_ne!(
+                            field(&refused, "formatVersion"),
+                            field(&built, "formatVersion"),
+                            "vector '{name}' is only unsupported if it names another format version"
+                        );
+                        assert_eq!(
+                            field(&built, "formatVersion"),
+                            "rustfs.connect.offline.enrollmentResponse/1",
+                            "vector '{name}': the format version is frozen"
+                        );
+                    }
+                    "UNSUPPORTED_PROTOCOL" => {
+                        assert_ne!(
+                            field(&refused, "protocolVersion"),
+                            field(&built, "protocolVersion"),
+                            "vector '{name}' is only unsupported if it names another protocol major"
+                        );
+                        assert_eq!(field(&built, "protocolVersion"), "v1", "vector '{name}': the protocol major is frozen");
+                    }
+                    "ENROLLMENT_REPLAYED" => {
+                        // The vector claims to be a byte-identical replay of an
+                        // accepted response; hold it to that, because a replay
+                        // vector that is not byte identical proves nothing about
+                        // single use.
+                        let accepted = accept_vector_named("response binding the device public key and the challenge proof");
+                        assert_eq!(
+                            signed_octets(&vector["document"]),
+                            signed_octets(&accepted["document"]),
+                            "vector '{name}' must be the accepted response octet for octet"
+                        );
+                        assert_eq!(
+                            field(&vector["document"]["signature"], "value"),
+                            field(&accepted["document"]["signature"], "value"),
+                            "vector '{name}' must carry the accepted response's signature"
+                        );
+
+                        // A fresh device nonce is a different artifact, so a
+                        // second enrollment is never mistaken for a replay of
+                        // the first.
+                        let other = OfflineEnrollment::build_response(&challenge, &key, &[0x5a; 32], produced_at)
+                            .expect("a second response builds");
+                        assert_ne!(
+                            signed_octets(&built_envelope),
+                            signed_octets(&serde_json::from_slice::<Value>(&other).expect("JSON")),
+                            "vector '{name}': a different device nonce must yield a different artifact"
+                        );
+                    }
+                    other => panic!("vector '{name}' names an unhandled reason {other}; extend this test"),
+                }
+            }
+        }
+
+        covered += 1;
+    }
+
+    assert_eq!(
+        covered, 9,
+        "reject-vectors.json publishes nine response vectors; losing one silently narrows the suite"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// Signature encoding
+// ---------------------------------------------------------------------------
+
+/// The high-S malleation is the rejection the whole encoding rule exists for.
+///
+/// `(r, n - s)` is a second valid signature over the same document under the
+/// same key. Every mainstream ECDSA library verifies it, so an implementation
+/// that hands the decoded octets straight to `p256` accepts a forged-looking
+/// duplicate of a genuine challenge — and because the 64 octets differ, that
+/// duplicate is a distinct artifact identity that slips past any deduplication
+/// keyed on the signature. This test proves the rejection came from the
+/// encoding rule and not from a failed verification: it first shows the
+/// malleated signature verifying mathematically, then requires
+/// `verify_challenge` to refuse it as SIGNATURE_NOT_CANONICAL.
+#[test]
+fn malleated_high_s_signature_is_refused_although_it_verifies_mathematically() {
+    use p256::ecdsa::signature::Verifier as _;
+
+    let model = trust_model();
+    let malleated = model["rejectedSignatureEncodings"]
+        .as_array()
+        .expect("trust-model.json publishes rejected encodings")
+        .iter()
+        .find(|entry| field(entry, "reason") == "SIGNATURE_NOT_CANONICAL")
+        .expect("trust-model.json publishes the high-S malleation")
+        .clone();
+    assert!(
+        malleated["acceptedByALenientVerifier"].as_bool() == Some(true),
+        "this vector is only interesting because a lenient verifier accepts it"
+    );
+
+    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
+    let genuine_value = field(&vector["document"]["signature"], "value").to_string();
+    let malleated_value = field(&malleated, "value").to_string();
+    assert_ne!(genuine_value, malleated_value, "the malleation must be a different encoding");
+
+    let genuine = BASE64_URL_NO_PAD.decode(&genuine_value).expect("signature is base64url");
+    let raw = BASE64_URL_NO_PAD.decode(&malleated_value).expect("signature is base64url");
+    assert_eq!(raw.len(), 64, "the malleation is well formed at 64 octets");
+    assert_eq!(raw[..32], genuine[..32], "the malleation shares r with the genuine signature");
+    assert_ne!(raw[32..], genuine[32..], "the malleation replaces s with n - s");
+
+    // Step one: the malleated pair really does verify under the signing key, so
+    // a verifier cannot be excused for accepting it on mathematical grounds.
+    let signature = p256::ecdsa::Signature::from_slice(&raw).expect("the malleated signature parses");
+    assert!(signature.normalize_s().is_some(), "the malleated signature must be the high-S form");
+    let key = verifying_key(field(&published_key("signing"), "publicKey"));
+    let input = signing_input(&domain_tag("enrollmentChallenge"), &signed_octets(&vector["document"]));
+    key.verify(&input, &signature)
+        .expect("the malleated signature must verify mathematically, or this test proves nothing");
+
+    // Step two: the implementation must refuse it anyway, and say why.
+    let mut tampered = vector["document"].clone();
+    tampered["signature"]["value"] = Value::String(malleated_value);
+
+    let now = unix(field(&vector, "evaluationTime"));
+    let error = OfflineEnrollment::verify_challenge(&envelope(&tampered), now)
+        .expect_err("a high-S signature must be refused even though it verifies");
+    assert_eq!(
+        error.reason(),
+        "SIGNATURE_NOT_CANONICAL",
+        "a malleated signature is a canonicality failure, not a verification failure"
+    );
+}
+
+/// Every encoding `trust-model.json` names as rejected must fail with the
+/// reason it names — DER, padded base64url, truncation, and out-of-range
+/// scalars alongside the malleation. Three of the five are accepted by a
+/// lenient verifier, so a single blanket "signature did not verify" answer would
+/// be both wrong and undiagnosable.
+#[test]
+fn every_rejected_signature_encoding_fails_with_its_frozen_reason() {
+    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
+    let now = unix(field(&vector, "evaluationTime"));
+    let model = trust_model();
+    let encodings = model["rejectedSignatureEncodings"]
+        .as_array()
+        .expect("trust-model.json publishes rejected encodings");
+
+    for entry in encodings {
+        let name = field(entry, "name");
+        let mut tampered = vector["document"].clone();
+        tampered["signature"]["value"] = Value::String(field(entry, "value").to_string());
+
+        let error: EnrollmentError = OfflineEnrollment::verify_challenge(&envelope(&tampered), now)
+            .err()
+            .unwrap_or_else(|| panic!("rejected encoding '{name}' must not verify"));
+
+        assert_eq!(error.reason(), field(entry, "reason"), "rejected encoding '{name}'");
+    }
+
+    assert_eq!(encodings.len(), 5, "trust-model.json freezes five rejected encodings");
+}
+
+// ---------------------------------------------------------------------------
+// Clock window
+// ---------------------------------------------------------------------------
+
+/// The tolerated window is `[issuedAt - 300, expiresAt + 300]`, inclusive at
+/// both ends. An air-gapped device has no synchronised clock, so an
+/// off-by-one here either strands a legitimate enrollment or widens the window
+/// a stolen challenge stays usable in. Both ends are checked at the exact bound
+/// and one second past it, and the reason distinguishes the two directions.
+#[test]
+fn challenge_is_accepted_at_the_exact_skew_bound_and_refused_one_second_past_it() {
+    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
+    let document = envelope(&vector["document"]);
+    let signed = signed_document(&vector["document"]);
+
+    let issued_at = unix(field(&signed, "issuedAt"));
+    let expires_at = unix(field(&signed, "expiresAt"));
+
+    let earliest = issued_at - SKEW_TOLERANCE_SECONDS;
+    OfflineEnrollment::verify_challenge(&document, earliest).expect("the earliest tolerated instant is inside the window");
+    let error =
+        OfflineEnrollment::verify_challenge(&document, earliest - 1).expect_err("one second earlier is outside the window");
+    assert_eq!(error.reason(), "CHALLENGE_NOT_YET_VALID");
+
+    let latest = expires_at + SKEW_TOLERANCE_SECONDS;
+    OfflineEnrollment::verify_challenge(&document, latest).expect("the latest tolerated instant is inside the window");
+    let error = OfflineEnrollment::verify_challenge(&document, latest + 1).expect_err("one second later is outside the window");
+    assert_eq!(error.reason(), "CHALLENGE_EXPIRED");
+}
+
+// ---------------------------------------------------------------------------
+// Response production
+// ---------------------------------------------------------------------------
+
+/// Assert a built response proves possession of the key it presents: the
+/// fingerprint matches the presented key, and the detached signature is a
+/// canonical low-S ES256 signature that verifies under that key over the exact
+/// octets transmitted.
+fn assert_response_proves_possession(built_envelope: &Value, label: &str) {
+    use p256::ecdsa::signature::Verifier as _;
+
+    let raw = signed_octets(built_envelope);
+    let built = signed_document(built_envelope);
+    let signature_block = &built_envelope["signature"];
+
+    assert_eq!(field(signature_block, "algorithm"), "ES256", "{label}: the algorithm is frozen");
+
+    let value = field(signature_block, "value");
+    assert_eq!(value.len(), 86, "{label}: the transfer encoding is 86 unpadded base64url characters");
+    assert!(
+        value.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_'),
+        "{label}: the signature must use the base64url alphabet with no padding"
+    );
+
+    let bytes = BASE64_URL_NO_PAD.decode(value).expect("signature is base64url");
+    assert_eq!(bytes.len(), 64, "{label}: the signature is a fixed-width r || s");
+    let signature = p256::ecdsa::Signature::from_slice(&bytes).expect("signature parses");
+    assert!(
+        signature.normalize_s().is_none(),
+        "{label}: this side must never emit the malleated high-S form it refuses to accept"
+    );
+
+    let presented = field(&built, "devicePublicKey");
+    let key = verifying_key(presented);
+    key.verify(&signing_input(&domain_tag("enrollmentResponse"), &raw), &signature)
+        .unwrap_or_else(|error| panic!("{label}: the response must verify under the key it presents: {error}"));
+
+    // `signature.keyIdAlgorithm`: the lowercase SHA-256 of the DER
+    // SubjectPublicKeyInfo, not of the bare point and not of the transfer
+    // encoding.
+    let mut spki = hex_to_bytes(SPKI_PREFIX_HEX);
+    spki.extend_from_slice(&BASE64_URL_NO_PAD.decode(presented).expect("public key is base64url"));
+    let fingerprint = sha256_hex(&spki);
+    assert_eq!(
+        field(&built, "deviceKeyId"),
+        fingerprint,
+        "{label}: deviceKeyId must be the fingerprint of the key the document presents"
+    );
+    assert_eq!(
+        field(signature_block, "keyId"),
+        fingerprint,
+        "{label}: the detached signature must name the same key"
+    );
+}
+
+/// A response is the only thing Connect will ever see from this device, so it
+/// has to carry the whole binding on its own: the challenge it answers, the
+/// proof that challenge was genuine, the key being enrolled, and possession of
+/// that key.
+#[test]
+fn built_response_binds_the_challenge_proof_and_proves_possession_of_the_device_key() {
+    let vector = accept_vector_named("response binding the device public key and the challenge proof");
+    let (challenge_vector, challenge) = answered_challenge(&vector);
+    let key = DeviceIdentity::generate();
+    let produced_at = unix(field(&signed_document(&vector["document"]), "producedAt"));
+
+    let bytes = OfflineEnrollment::build_response(&challenge, &key, &[0x11; 32], produced_at).expect("the response builds");
+    let built_envelope: Value = serde_json::from_slice(&bytes).expect("the response is JSON");
+    let built = signed_document(&built_envelope);
+
+    assert_response_proves_possession(&built_envelope, "built response");
+
+    // The proof is the challenge's own detached signature. A producer that
+    // echoed the nonce alone, or hashed something, would let a response be
+    // built from an unverified challenge.
+    assert_eq!(
+        field(&built, "challengeProof"),
+        field(&challenge_vector["document"]["signature"], "value"),
+        "the proof must be the signature of the challenge being answered"
+    );
+    assert_eq!(field(&built, "challengeNonce"), challenge.nonce);
+    assert_eq!(field(&built, "challengeId"), challenge.challenge_id);
+
+    assert_eq!(
+        field(&built, "devicePublicKey"),
+        BASE64_URL_NO_PAD.encode(&key.public_key_der()[hex_to_bytes(SPKI_PREFIX_HEX).len()..]),
+        "the presented key must be the key that was passed in"
+    );
+    assert_eq!(
+        field(&built, "deviceNonce"),
+        BASE64_URL_NO_PAD.encode([0x11; 32]),
+        "the device nonce must be the one that was passed in"
+    );
+    assert!(field(&built, "producedAt").ends_with('Z'), "producedAt is a UTC RFC 3339 instant");
+}
+
+/// The response leaves the air gap on removable media and is read by anyone who
+/// handles it. A producer that serialised the key pair instead of the public
+/// key, or logged a debug rendering into the document, would put the enrolled
+/// private key on that medium — and the enrollment would still succeed, so
+/// nothing else in this suite would notice.
+#[test]
+fn built_response_carries_no_private_key_material() {
+    let vector = accept_vector_named("response binding the device public key and the challenge proof");
+    let (_, challenge) = answered_challenge(&vector);
+    let key = DeviceIdentity::generate();
+    let produced_at = unix(field(&signed_document(&vector["document"]), "producedAt"));
+
+    let response = OfflineEnrollment::build_response(&challenge, &key, &[0x22; 32], produced_at).expect("the response builds");
+
+    // The envelope carries the signed document base64-encoded, so a needle
+    // present in the document is not present in the envelope octets. Both
+    // layers are searched: an operator handling the medium can read either.
+    let envelope_value: Value = serde_json::from_slice(&response).expect("the response is JSON");
+    let mut haystack = response;
+    haystack.extend_from_slice(&signed_octets(&envelope_value));
+
+    let pkcs8 = key.to_pkcs8_der().expect("serialise the key");
+    let secret = <p256::SecretKey as p256::pkcs8::DecodePrivateKey>::from_pkcs8_der(&pkcs8).expect("the key parses");
+    let scalar = secret.to_bytes();
+
+    // Every spelling the scalar could plausibly reach a document in: raw, and
+    // the three encodings this protocol already uses elsewhere.
+    let scalar_hex: String = scalar.iter().map(|byte| format!("{byte:02x}")).collect();
+    for (description, needle) in [
+        ("the PKCS#8 encoding", pkcs8.to_vec()),
+        ("the raw private scalar", scalar.to_vec()),
+        ("the scalar in base64url", BASE64_URL_NO_PAD.encode(scalar).into_bytes()),
+        ("the scalar in standard base64", BASE64_STANDARD.encode(scalar).into_bytes()),
+        ("the scalar in hex", scalar_hex.into_bytes()),
+    ] {
+        assert!(
+            !haystack.windows(needle.len()).any(|window| window == needle.as_slice()),
+            "the response must not contain {description}"
+        );
+    }
+
+    // The public half must be there, so the absence above is a statement about
+    // what was excluded rather than about a haystack that would not have found
+    // the private half either.
+    let point = BASE64_URL_NO_PAD.encode(&key.public_key_der()[hex_to_bytes(SPKI_PREFIX_HEX).len()..]);
+    assert!(
+        haystack.windows(point.len()).any(|window| window == point.as_bytes()),
+        "the response must still present the public key"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// The offline invariant
+// ---------------------------------------------------------------------------
+
+/// The whole surface exists because there is no network. This asserts that
+/// three different ways, because no single one of them is conclusive on its own.
+///
+/// 1. The process opens no descriptor across a full verify-and-respond cycle. A
+///    socket, a DNS resolver, a pooled HTTP client, or a revocation-list fetch
+///    all show up here — including one that is opened and cached rather than
+///    opened and closed, which is what a lazily built client does.
+/// 2. The cycle is a pure byte transform: the same inputs produce the same
+///    verified fields, and the evaluation instant is an argument rather than an
+///    ambient read, so nothing about the outcome can depend on reachability.
+/// 3. Repeating the cycle changes nothing observable, so a first call cannot be
+///    quietly initialising shared state that a later one reuses.
+#[cfg(unix)]
+#[test]
+fn enrollment_opens_no_descriptor_and_is_a_pure_byte_transform() {
+    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
+    let document = envelope(&vector["document"]);
+    let now = unix(field(&vector, "evaluationTime"));
+    let key = DeviceIdentity::generate();
+
+    // Warm anything the test harness itself lazily opens before the baseline.
+    let _ = open_descriptors();
+    let baseline = open_descriptors();
+    assert!(
+        !baseline.is_empty(),
+        "the descriptor table must be readable for this test to mean anything"
+    );
+
+    let mut fields = Vec::new();
+    for _ in 0..2 {
+        let challenge = OfflineEnrollment::verify_challenge(&document, now).expect("the challenge verifies");
+        let response = OfflineEnrollment::build_response(&challenge, &key, &[0x33; 32], now).expect("the response builds");
+        fields.push((
+            challenge.challenge_id.clone(),
+            challenge.nonce.clone(),
+            challenge.challenge_proof.clone(),
+            signed_octets(&serde_json::from_slice::<Value>(&response).expect("JSON")),
+        ));
+    }
+
+    assert_eq!(
+        open_descriptors(),
+        baseline,
+        "the enrollment path must not open a descriptor: no socket, no resolver, no cached client"
+    );
+
+    let (first, second) = (&fields[0], &fields[1]);
+    assert_eq!(first.0, second.0, "verification must be deterministic");
+    assert_eq!(first.1, second.1, "verification must be deterministic");
+    assert_eq!(first.2, second.2, "verification must be deterministic");
+    assert_eq!(
+        first.3, second.3,
+        "the signed response octets are a function of the challenge, the key, the nonce, and the instant"
+    );
+}
+
+#[cfg(unix)]
+fn open_descriptors() -> Vec<String> {
+    // Linux publishes the table at /proc/self/fd; the BSDs and macOS at /dev/fd.
+    let path = if PathBuf::from("/proc/self/fd").is_dir() {
+        "/proc/self/fd"
+    } else {
+        "/dev/fd"
+    };
+
+    let mut entries: Vec<String> = fs::read_dir(path)
+        .unwrap_or_else(|error| panic!("read {path}: {error}"))
+        .map(|entry| entry.expect("read dir entry").file_name().to_string_lossy().into_owned())
+        .collect();
+    entries.sort();
+    entries
+}
+
+/// A descriptor count taken around a call cannot see a socket that was opened
+/// and closed inside it, so the invariant is also asserted where it can be
+/// stated absolutely: the implementation names no network API at all.
+///
+/// This is the shape the regression actually takes — someone adds a
+/// revocation-list fetch, a time-server check, or a "just confirm the challenge
+/// with Connect" call — and it is caught at the source rather than by observing
+/// its effects.
+#[test]
+fn enrollment_implementation_names_no_network_api() {
+    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("src/connect/offline/enrollment.rs");
+    let source = fs::read_to_string(&path).unwrap_or_else(|error| panic!("read {}: {error}", path.display()));
+
+    // Prose is allowed to discuss the invariant it is documenting, so only code
+    // is scanned.
+    let code: String = source
+        .lines()
+        .filter(|line| !line.trim_start().starts_with("//"))
+        .collect::<Vec<_>>()
+        .join("\n");
+
+    for forbidden in [
+        "std::net",
+        "tokio::net",
+        "TcpStream",
+        "TcpListener",
+        "UdpSocket",
+        "UnixStream",
+        "ToSocketAddrs",
+        "reqwest",
+        "hyper",
+        "tonic",
+    ] {
+        assert!(
+            !code.contains(forbidden),
+            "offline enrollment must not reach the network, but the implementation names {forbidden}"
+        );
+    }
+}
Author	SHA1	Message	Date
马登山	eeab9d201b	fix(heal): fence format repair during pool transitions	2026-08-22 00:25:32 +08:00
Zhengchao An	bce5922aef	feat(connect): answer offline enrolment challenges without a network (#6335 )	2026-08-21 13:16:18 +00:00