style: cargo fmt

refactor(data-usage): ReplicationStats -> ReplicationTargetUsage
Rename the data-usage crate's ReplicationStats to ReplicationTargetUsage. Serde field names are byte-identical (only the Rust type name changed; field identifiers that rmp encodes are untouched). An rmp round-trip test guards against future drift. Scanner test imports updated to match.
2026-08-22 12:26:37 +00:00 · 2026-08-22 18:48:01 +08:00 · 2026-08-22 18:48:00 +08:00
17 changed files with 72 additions and 1731 deletions
@@ -585,9 +585,12 @@ impl VersionsHistogram {
    }
 }

-/// Replication statistics for a single target
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
-pub struct ReplicationStats {
+/// Replication statistics for a single target.
+///
+/// Renamed from `ReplicationStats`; serde field names are preserved
+/// byte-identically to maintain wire compatibility with existing snapshots.
+#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct ReplicationTargetUsage {
    pub pending_size: u64,
    pub replicated_size: u64,
    pub failed_size: u64,
@@ -600,7 +603,7 @@ pub struct ReplicationStats {
    pub replicated_count: u64,
 }

-impl ReplicationStats {
+impl ReplicationTargetUsage {
    pub fn is_empty(&self) -> bool {
        let Self {
            pending_size,
@@ -636,7 +639,7 @@ impl ReplicationStats {
 /// Replication statistics for all targets
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]
 pub struct ReplicationAllStats {
-    pub targets: HashMap<String, ReplicationStats>,
+    pub targets: HashMap<String, ReplicationTargetUsage>,
    pub replica_size: u64,
    pub replica_count: u64,
 }
@@ -649,7 +652,7 @@ impl ReplicationAllStats {
            targets,
        } = self;

-        *replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationStats::is_empty)
+        *replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationTargetUsage::is_empty)
    }

    #[deprecated(note = "use is_empty instead")]
@@ -2466,7 +2469,7 @@ mod tests {

    #[test]
    fn replication_stats_empty_checks_every_field() {
-        type SetField = fn(&mut ReplicationStats);
+        type SetField = fn(&mut ReplicationTargetUsage);

        let cases: [(&str, SetField); 10] = [
            ("pending_size", |stats| stats.pending_size = 1),
@@ -2481,9 +2484,9 @@ mod tests {
            ("replicated_count", |stats| stats.replicated_count = 1),
        ];

-        assert!(ReplicationStats::default().is_empty());
+        assert!(ReplicationTargetUsage::default().is_empty());
        for (field, set_nonzero) in cases {
-            let mut stats = ReplicationStats::default();
+            let mut stats = ReplicationTargetUsage::default();
            set_nonzero(&mut stats);
            assert!(!stats.is_empty(), "{field} must make replication stats non-empty");
        }
@@ -2514,17 +2517,17 @@ mod tests {
        }

        let empty_targets = ReplicationAllStats {
-            targets: HashMap::from([("arn:test:empty".to_string(), ReplicationStats::default())]),
+            targets: HashMap::from([("arn:test:empty".to_string(), ReplicationTargetUsage::default())]),
            ..Default::default()
        };
        assert!(empty_targets.is_empty(), "all-empty targets must keep aggregate stats empty");

        let stats = ReplicationAllStats {
            targets: HashMap::from([
-                ("arn:test:empty".to_string(), ReplicationStats::default()),
+                ("arn:test:empty".to_string(), ReplicationTargetUsage::default()),
                (
                    "arn:test:non-empty".to_string(),
-                    ReplicationStats {
+                    ReplicationTargetUsage {
                        pending_count: 1,
                        ..Default::default()
                    },
@@ -2565,7 +2568,7 @@ mod tests {
                replication_stats: Some(ReplicationAllStats {
                    targets: HashMap::from([(
                        "arn:test:pending".to_string(),
-                        ReplicationStats {
+                        ReplicationTargetUsage {
                            pending_count: 1,
                            ..Default::default()
                        },
@@ -2714,7 +2717,7 @@ mod tests {
                targets: HashMap::from([
                    (
                        "arn:self-only".to_string(),
-                        ReplicationStats {
+                        ReplicationTargetUsage {
                            pending_size: 7,
                            pending_count: 1,
                            ..Default::default()
@@ -2722,7 +2725,7 @@ mod tests {
                    ),
                    (
                        "arn:shared".to_string(),
-                        ReplicationStats {
+                        ReplicationTargetUsage {
                            failed_size: 3,
                            failed_count: 1,
                            missed_threshold_size: 2,
@@ -2741,7 +2744,7 @@ mod tests {
                targets: HashMap::from([
                    (
                        "arn:shared".to_string(),
-                        ReplicationStats {
+                        ReplicationTargetUsage {
                            failed_size: 5,
                            failed_count: 2,
                            after_threshold_size: 4,
@@ -2751,7 +2754,7 @@ mod tests {
                    ),
                    (
                        "arn:other-only".to_string(),
-                        ReplicationStats {
+                        ReplicationTargetUsage {
                            replicated_size: 11,
                            replicated_count: 3,
                            ..Default::default()
@@ -2993,7 +2996,9 @@ mod tests {
    fn replication_target_deserialization_preserves_large_historical_maps() {
        let mut stats = ReplicationAllStats::default();
        for index in 0..=1024 {
-            stats.targets.insert(format!("target-{index}"), ReplicationStats::default());
+            stats
+                .targets
+                .insert(format!("target-{index}"), ReplicationTargetUsage::default());
        }
        let encoded = rmp_serde::to_vec_named(&stats).expect("large replication target fixture should encode");
        let decoded = rmp_serde::from_slice::<ReplicationAllStats>(&encoded)
@@ -3002,6 +3007,47 @@ mod tests {
        assert_eq!(decoded.targets.len(), stats.targets.len());
    }

+    /// Round-trip test: encoding a [`ReplicationTargetUsage`] and decoding it back
+    /// must produce the exact same value.  This guards against accidental serde
+    /// field-name drift during the `ReplicationStats` -> `ReplicationTargetUsage`
+    /// rename.  Wire-level field names are the serialized Rust field identifiers,
+    /// which must remain byte-identical.
+    #[test]
+    fn replication_target_usage_rmp_round_trip() {
+        let original = ReplicationTargetUsage {
+            pending_size: 100,
+            replicated_size: 2_000,
+            failed_size: 50,
+            failed_count: 3,
+            pending_count: 7,
+            missed_threshold_size: 11,
+            after_threshold_size: 22,
+            missed_threshold_count: 1,
+            after_threshold_count: 2,
+            replicated_count: 99,
+        };
+
+        let buf = rmp_serde::to_vec_named(&original).expect("encode ReplicationTargetUsage to msgpack");
+        let decoded: ReplicationTargetUsage = rmp_serde::from_slice(&buf).expect("decode ReplicationTargetUsage from msgpack");
+        assert_eq!(original, decoded, "round-trip through rmp must preserve every field");
+
+        // Also verify that encoding as an unnamed sequence and then decoding
+        // with named fields produces the correct mapping (this catches reordering).
+        let named_buf = rmp_serde::to_vec_named(&original).expect("re-encode for field-name pinning");
+        // Spot-check that known field names appear in the named encoding.
+        let named_str = String::from_utf8_lossy(&named_buf);
+        assert!(named_str.contains("pending_size"), "field 'pending_size' must survive the rename");
+        assert!(named_str.contains("replicated_size"), "field 'replicated_size' must survive the rename");
+        assert!(
+            named_str.contains("missed_threshold_size"),
+            "field 'missed_threshold_size' must survive the rename"
+        );
+        assert!(
+            named_str.contains("after_threshold_count"),
+            "field 'after_threshold_count' must survive the rename"
+        );
+    }
+
    #[test]
    fn checked_merge_rejects_noncanonical_histograms_without_mutation() {
        let mut entry = DataUsageEntry {
@@ -16,7 +16,7 @@ use super::persistence::DataUsageCacheLoadAttempt;
 use super::*;
 use crate::storage_api::scanner_io::{HTTPRangeSpec, ObjectIO};
 use crate::{ScannerGetObjectReader, ScannerPutObjReader};
-use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
+use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
 use serde_json::Value;
 use std::io::Cursor;
 use std::pin::Pin;
@@ -1636,7 +1636,7 @@ fn size_recursive_prunes_empty_and_preserves_threshold_replication_stats() {
            replication_stats: Some(ReplicationAllStats {
                targets: HashMap::from([(
                    "arn:test:threshold".to_string(),
-                    ReplicationStats {
+                    ReplicationTargetUsage {
                        after_threshold_count: 1,
                        ..Default::default()
                    },
@@ -13,7 +13,7 @@
 // limitations under the License.

 use super::*;
-use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
+use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};

 const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]);

@@ -271,7 +271,7 @@ fn completed_data_usage_info_flattens_nested_bucket_entries() {
        replication_stats: Some(ReplicationAllStats {
            targets: HashMap::from([(
                "arn:target".to_string(),
-                ReplicationStats {
+                ReplicationTargetUsage {
                    replicated_size: 2048,
                    replicated_count: 2,
                    ..Default::default()
@@ -25,7 +25,7 @@
    },
    {
      "name": "heartbeat",
-      "status": "populated",
+      "status": "reserved",
      "purpose": "Heartbeat payloads, Connect receive time, and freshness window behavior."
    },
    {
@@ -1,5 +0,0 @@
-975c1ca53eefeef6766a6fc0b3d3281f7408255342b0686e5e2aee5ad055414c  duplicate.json
-963529a38a02849c6c2acc6d72668dca9f63218b49c89fae41a451b584850411  overflow.json
-e3adeee1c8a19aa17e70894896fb79c072e3785bea3611b93c11e79f039ed5af  stale.json
-35b9cebd8525389a701e8fe69fbe96407bcb31aa28392fe95babf4a4886985ad  unknown.json
-37941735dbd6ad3d238258a7b2cae6f0b3aa0ecaae1d8817817c3d718d11d633  valid.json
@@ -1,9 +0,0 @@
-{
-  "protocolVersion": "v1",
-  "fixtureSet": "heartbeat",
-  "fixture": "duplicate",
-  "description": "An exact requestId replay returns the first result and creates no second heartbeat.",
-  "first": {"requestId": "550e8400-e29b-41d4-a716-446655440000", "sequence": 42},
-  "replay": {"requestId": "550e8400-e29b-41d4-a716-446655440000", "sequence": 42},
-  "expected": {"decision": "DUPLICATE", "heartbeatWrites": 1, "events": 1, "sameResponse": true}
-}
@@ -1,11 +0,0 @@
-{
-  "protocolVersion": "v1",
-  "fixtureSet": "heartbeat",
-  "fixture": "overflow",
-  "description": "Values beyond frozen bounds are rejected before persistence.",
-  "vectors": [
-    {"field": "sequence", "value": 9007199254740992, "maximum": 9007199254740991},
-    {"field": "coarseNodeSummary.total", "value": 4097, "maximum": 4096}
-  ],
-  "expected": {"decision": "REJECT", "httpStatus": 422, "status": "INVALID_ARGUMENT"}
-}
@@ -1,9 +0,0 @@
-{
-  "protocolVersion": "v1",
-  "fixtureSet": "heartbeat",
-  "fixture": "stale",
-  "description": "A lower heartbeat sequence is retained as history and cannot replace the current projection.",
-  "head": {"requestId": "550e8400-e29b-41d4-a716-446655440000", "sequence": 42},
-  "late": {"requestId": "7c4d2e10-9f83-4a5b-b6c7-d8e9f0a1b2c3", "sequence": 9},
-  "expected": {"decision": "ACCEPT_HISTORY", "currentSequence": 42, "historySequence": 9}
-}
@@ -1,18 +0,0 @@
-{
-  "protocolVersion": "v1",
-  "fixtureSet": "heartbeat",
-  "fixture": "unknown",
-  "description": "Unknown optional members and capabilities are accepted, discarded before hashing, and never stored or echoed.",
-  "requestAdditions": {
-    "telemetryProfile": "extended",
-    "authorization": "Bearer non-functional-example",
-    "capabilities": ["heartbeat", "future.capability"],
-    "coarseNodeSummary": {"rackNames": ["customer-rack"]}
-  },
-  "expected": {
-    "decision": "ACCEPT",
-    "storedCapabilities": ["heartbeat"],
-    "discarded": ["authorization", "future.capability", "telemetryProfile", "coarseNodeSummary.rackNames"],
-    "echoed": []
-  }
-}
@@ -1,21 +0,0 @@
-{
-  "protocolVersion": "v1",
-  "fixtureSet": "heartbeat",
-  "fixture": "valid",
-  "description": "A bounded L0 heartbeat. clientTime is advisory; Connect's receivedAt is online authority.",
-  "request": {
-    "protocolVersion": "v1",
-    "requestId": "550e8400-e29b-41d4-a716-446655440000",
-    "agentVersion": "rustfs-agent/1.19.4",
-    "capabilities": ["heartbeat", "inventory"],
-    "sequence": 42,
-    "clientTime": "2026-08-22T01:02:03Z",
-    "coarseNodeSummary": {"total": 8, "healthy": 7, "degraded": 1}
-  },
-  "expected": {
-    "decision": "ACCEPT",
-    "acceptedVersion": "v1",
-    "responseFields": ["serverTime", "acceptedVersion", "capabilityHints"],
-    "onlineAuthority": "serverTime"
-  }
-}
@@ -1,173 +0,0 @@
-// 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.
-
-use std::env;
-use std::ffi::OsString;
-use std::fs;
-use std::path::PathBuf;
-use std::time::Duration;
-
-use super::{CredentialStore, IdentityStore};
-
-pub const ENV_CONNECT_ENDPOINT: &str = "RUSTFS_CONNECT_ENDPOINT";
-pub const ENV_CONNECT_ROOT_CA_FILE: &str = "RUSTFS_CONNECT_ROOT_CA_FILE";
-pub const ENV_CONNECT_STATE_DIR: &str = "RUSTFS_CONNECT_STATE_DIR";
-
-#[derive(Clone, Copy, Debug)]
-pub struct HeartbeatSchedule {
-    pub cadence: Duration,
-    pub jitter: Duration,
-    pub timeout: Duration,
-    pub initial_backoff: Duration,
-    pub max_backoff: Duration,
-}
-
-impl Default for HeartbeatSchedule {
-    fn default() -> Self {
-        Self {
-            cadence: Duration::from_secs(30),
-            jitter: Duration::from_secs(3),
-            timeout: Duration::from_secs(5),
-            initial_backoff: Duration::from_secs(1),
-            max_backoff: Duration::from_secs(5 * 60),
-        }
-    }
-}
-
-#[derive(Clone, Debug)]
-pub struct HeartbeatConfig {
-    pub endpoint: String,
-    pub root_ca_pem: Vec<u8>,
-    pub identity_store: IdentityStore,
-    pub credential_store: CredentialStore,
-    pub state_path: PathBuf,
-    pub schedule: HeartbeatSchedule,
-}
-
-impl HeartbeatConfig {
-    pub fn new(
-        endpoint: impl Into<String>,
-        root_ca_pem: impl Into<Vec<u8>>,
-        identity_store: IdentityStore,
-        credential_store: CredentialStore,
-        state_path: impl Into<PathBuf>,
-    ) -> Self {
-        Self {
-            endpoint: endpoint.into(),
-            root_ca_pem: root_ca_pem.into(),
-            identity_store,
-            credential_store,
-            state_path: state_path.into(),
-            schedule: HeartbeatSchedule::default(),
-        }
-    }
-
-    pub fn from_env() -> Result<Option<Self>, HeartbeatConfigError> {
-        Self::from_env_values(
-            env::var_os(ENV_CONNECT_ENDPOINT),
-            env::var_os(ENV_CONNECT_ROOT_CA_FILE),
-            env::var_os(ENV_CONNECT_STATE_DIR),
-        )
-    }
-
-    fn from_env_values(
-        endpoint: Option<OsString>,
-        root_ca_file: Option<OsString>,
-        state_dir: Option<OsString>,
-    ) -> Result<Option<Self>, HeartbeatConfigError> {
-        let configured = endpoint.is_some() || root_ca_file.is_some() || state_dir.is_some();
-        if !configured {
-            return Ok(None);
-        }
-        let (Some(endpoint), Some(root_ca_file), Some(state_dir)) = (endpoint, root_ca_file, state_dir) else {
-            return Err(HeartbeatConfigError::Partial);
-        };
-        let endpoint = endpoint.into_string().map_err(|_| HeartbeatConfigError::EndpointEncoding)?;
-        let root_ca_file = PathBuf::from(root_ca_file);
-        let state_dir = PathBuf::from(state_dir);
-        if endpoint.is_empty() || root_ca_file.as_os_str().is_empty() || state_dir.as_os_str().is_empty() {
-            return Err(HeartbeatConfigError::Partial);
-        }
-        let root_ca_pem = fs::read(&root_ca_file).map_err(|source| HeartbeatConfigError::RootCertificate {
-            path: root_ca_file,
-            source,
-        })?;
-        Ok(Some(Self::new(
-            endpoint,
-            root_ca_pem,
-            IdentityStore::new(state_dir.join("identity")),
-            CredentialStore::new(state_dir.join("credential")),
-            state_dir.join("heartbeat/state.json"),
-        )))
-    }
-}
-
-#[derive(Debug, thiserror::Error)]
-pub enum HeartbeatConfigError {
-    #[error(
-        "Connect heartbeat configuration requires RUSTFS_CONNECT_ENDPOINT, RUSTFS_CONNECT_ROOT_CA_FILE, and RUSTFS_CONNECT_STATE_DIR"
-    )]
-    Partial,
-    #[error("RUSTFS_CONNECT_ENDPOINT is not valid UTF-8")]
-    EndpointEncoding,
-    #[error("failed to read the Connect root CA at {path}: {source}")]
-    RootCertificate {
-        path: PathBuf,
-        #[source]
-        source: std::io::Error,
-    },
-}
-
-#[cfg(test)]
-mod tests {
-    use super::{HeartbeatConfig, HeartbeatConfigError};
-    use std::ffi::OsString;
-
-    #[test]
-    fn absent_environment_is_disabled_without_side_effects() {
-        assert!(
-            HeartbeatConfig::from_env_values(None, None, None)
-                .expect("absent config")
-                .is_none()
-        );
-    }
-
-    #[test]
-    fn partial_environment_is_rejected() {
-        assert!(matches!(
-            HeartbeatConfig::from_env_values(Some(OsString::from("https://connect.example/agent/")), None, None),
-            Err(HeartbeatConfigError::Partial)
-        ));
-    }
-
-    #[test]
-    fn complete_environment_builds_the_durable_paths() {
-        let temp = tempfile::tempdir().expect("tempdir");
-        let root = temp.path().join("root.pem");
-        std::fs::write(&root, b"root certificate").expect("root CA");
-        let state = temp.path().join("state");
-        let config = HeartbeatConfig::from_env_values(
-            Some(OsString::from("https://connect.example/agent/")),
-            Some(root.into_os_string()),
-            Some(state.clone().into_os_string()),
-        )
-        .expect("complete config")
-        .expect("enabled config");
-
-        assert_eq!(config.endpoint, "https://connect.example/agent/");
-        assert_eq!(config.root_ca_pem, b"root certificate");
-        assert_eq!(config.state_path, state.join("heartbeat/state.json"));
-        assert!(!state.exists(), "parsing configuration must not create state");
-    }
-}
@@ -1,585 +0,0 @@
-// 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.
-
-use std::fs;
-use std::io::{self, Write as _};
-use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicU64, Ordering};
-use std::time::Duration;
-
-use chrono::{DateTime, SecondsFormat, Utc};
-use reqwest::{Client, StatusCode, Url, header};
-use rustls::RootCertStore;
-use rustls::pki_types::{CertificateDer, pem::PemObject as _};
-use serde::{Deserialize, Serialize};
-use uuid::Uuid;
-use zeroize::Zeroizing;
-
-use super::config::HeartbeatConfig;
-use super::credential_store::{CredentialStoreError, DeviceCredential};
-use super::identity::IdentityError;
-use super::identity_store::StoreError;
-use super::registration::{CredentialValidationError, validate_stored_credential};
-
-const PROTOCOL_VERSION: &str = "v1";
-const AGENT_VERSION: &str = concat!("rustfs-agent/", env!("CARGO_PKG_VERSION"));
-const MAX_SEQUENCE: u64 = 9_007_199_254_740_991;
-const MAX_RESPONSE_BYTES: usize = 64 * 1024;
-#[cfg(unix)]
-const FILE_MODE: u32 = 0o600;
-static STAGING_SEQUENCE: AtomicU64 = AtomicU64::new(0);
-
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(deny_unknown_fields, rename_all = "camelCase")]
-pub struct CoarseNodeSummary {
-    total: u16,
-    healthy: u16,
-    degraded: u16,
-}
-
-impl CoarseNodeSummary {
-    pub fn new(total: u16, healthy: u16, degraded: u16) -> Result<Self, HeartbeatError> {
-        let summary = Self {
-            total,
-            healthy,
-            degraded,
-        };
-        if !summary.is_valid() {
-            return Err(HeartbeatError::NodeSummary);
-        }
-        Ok(summary)
-    }
-
-    fn is_valid(&self) -> bool {
-        self.total != 0
-            && self.total <= 4096
-            && self.healthy <= 4096
-            && self.degraded <= 4096
-            && self.healthy.saturating_add(self.degraded) <= self.total
-    }
-}
-
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub enum HeartbeatStatus {
-    Starting,
-    Online { server_time: String },
-    BackingOff { delay: Duration },
-    AuthenticationStopped { status: u16, reason: Option<String> },
-    Failed { reason: String },
-    Stopped,
-}
-
-#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(deny_unknown_fields, rename_all = "camelCase")]
-pub(crate) struct PendingHeartbeat {
-    protocol_version: String,
-    request_id: String,
-    agent_version: String,
-    capabilities: [String; 1],
-    sequence: u64,
-    client_time: String,
-    coarse_node_summary: CoarseNodeSummary,
-}
-
-impl PendingHeartbeat {
-    fn is_valid(&self) -> bool {
-        self.protocol_version == PROTOCOL_VERSION
-            && self.agent_version == AGENT_VERSION
-            && self.capabilities[0] == "heartbeat"
-            && self.sequence <= MAX_SEQUENCE
-            && self.coarse_node_summary.is_valid()
-            && is_exact_utc_seconds(&self.client_time)
-            && Uuid::parse_str(&self.request_id)
-                .is_ok_and(|request_id| request_id.get_version_num() == 4 && request_id.to_string() == self.request_id)
-    }
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct HeartbeatResponse {
-    server_time: String,
-    accepted_version: String,
-    #[serde(default)]
-    capability_hints: Vec<String>,
-}
-
-pub(crate) enum Delivery {
-    Accepted { server_time: String },
-    Retry { retry_after: Option<Duration> },
-    AuthenticationStopped { status: u16, reason: Option<String> },
-    Rejected { status: u16, reason: Option<String> },
-}
-
-pub(crate) struct HeartbeatSender {
-    endpoint: Url,
-    root_store: RootCertStore,
-    roots: Vec<CertificateDer<'static>>,
-    config: HeartbeatConfig,
-}
-
-impl HeartbeatSender {
-    pub(crate) fn new(config: HeartbeatConfig) -> Result<Self, HeartbeatError> {
-        let mut endpoint = Url::parse(&config.endpoint).map_err(|_| HeartbeatError::Endpoint)?;
-        if endpoint.scheme() != "https"
-            || endpoint.cannot_be_a_base()
-            || !endpoint.username().is_empty()
-            || endpoint.password().is_some()
-            || endpoint.query().is_some()
-            || endpoint.fragment().is_some()
-        {
-            return Err(HeartbeatError::Endpoint);
-        }
-        if !endpoint.path().ends_with('/') {
-            endpoint.set_path(&format!("{}/", endpoint.path()));
-        }
-        let roots = CertificateDer::pem_slice_iter(&config.root_ca_pem)
-            .collect::<Result<Vec<_>, _>>()
-            .map_err(|_| HeartbeatError::RootCertificate)?;
-        if roots.is_empty() {
-            return Err(HeartbeatError::RootCertificate);
-        }
-        let mut root_store = RootCertStore::empty();
-        let (accepted, rejected) = root_store.add_parsable_certificates(roots.clone());
-        if accepted != roots.len() || rejected != 0 {
-            return Err(HeartbeatError::RootCertificate);
-        }
-        let schedule = config.schedule;
-        if schedule.cadence.is_zero()
-            || schedule.timeout.is_zero()
-            || schedule.timeout > Duration::from_secs(5)
-            || schedule.initial_backoff.is_zero()
-            || schedule.max_backoff < schedule.initial_backoff
-            || schedule.max_backoff > Duration::from_secs(5 * 60)
-            || schedule.jitter > schedule.cadence
-        {
-            return Err(HeartbeatError::Schedule);
-        }
-        Ok(Self {
-            endpoint,
-            root_store,
-            roots,
-            config,
-        })
-    }
-
-    pub(crate) async fn send(&self, heartbeat: &PendingHeartbeat) -> Result<Delivery, HeartbeatError> {
-        let (cluster_uid, client) = {
-            let _lock = self.config.credential_store.lock().await?;
-            let credential = self.config.credential_store.load()?.ok_or(HeartbeatError::NotRegistered)?;
-            let identity = self.config.identity_store.load()?.ok_or(HeartbeatError::IdentityMissing)?;
-            validate_stored_credential(&credential, &identity, &self.root_store, &self.roots)?;
-            let now = Utc::now().timestamp();
-            if now < credential.not_before_unix || now >= credential.not_after_unix {
-                return Err(HeartbeatError::CredentialExpired);
-            }
-            let cluster_uid = cluster_uid(&credential)?.to_owned();
-            let client = self.client(&credential, &identity.to_pkcs8_pem()?)?;
-            (cluster_uid, client)
-        };
-        let url = self.endpoint.join(&format!("clusters/{cluster_uid}/heartbeats"))?;
-        let response = match client.post(url).json(heartbeat).send().await {
-            Ok(response) => response,
-            Err(error) if error.is_timeout() || error.is_connect() || error.is_request() => {
-                return Ok(Delivery::Retry { retry_after: None });
-            }
-            Err(error) => return Err(error.into()),
-        };
-        let status = response.status();
-        if status == StatusCode::TOO_MANY_REQUESTS {
-            return Ok(Delivery::Retry {
-                retry_after: retry_after(response.headers(), Utc::now(), self.config.schedule.max_backoff),
-            });
-        }
-        if status == StatusCode::REQUEST_TIMEOUT || status.is_server_error() {
-            return Ok(Delivery::Retry { retry_after: None });
-        }
-        if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) {
-            return Ok(Delivery::AuthenticationStopped {
-                status: status.as_u16(),
-                reason: response_reason(response).await,
-            });
-        }
-        if status != StatusCode::OK {
-            return Ok(Delivery::Rejected {
-                status: status.as_u16(),
-                reason: response_reason(response).await,
-            });
-        }
-        let accepted: HeartbeatResponse =
-            serde_json::from_slice(&bounded_body(response).await?).map_err(|_| HeartbeatError::Response)?;
-        if accepted.accepted_version != PROTOCOL_VERSION
-            || accepted.capability_hints.len() > 32
-            || accepted.capability_hints.iter().any(|hint| hint.len() > 32)
-            || !is_exact_utc_seconds(&accepted.server_time)
-        {
-            return Err(HeartbeatError::Response);
-        }
-        Ok(Delivery::Accepted {
-            server_time: accepted.server_time,
-        })
-    }
-
-    fn client(&self, credential: &DeviceCredential, key: &Zeroizing<String>) -> Result<Client, HeartbeatError> {
-        let mut pem = Zeroizing::new(Vec::with_capacity(credential.certificate_chain.len() + key.len() + 1));
-        pem.extend_from_slice(credential.certificate_chain.as_bytes());
-        pem.push(b'\n');
-        pem.extend_from_slice(key.as_bytes());
-        let identity = reqwest::Identity::from_pem(&pem).map_err(|_| HeartbeatError::IdentityCertificate)?;
-        let roots = self
-            .roots
-            .iter()
-            .map(|root| reqwest::Certificate::from_der(root.as_ref()))
-            .collect::<Result<Vec<_>, _>>()?;
-        Client::builder()
-            .https_only(true)
-            .redirect(reqwest::redirect::Policy::none())
-            .timeout(self.config.schedule.timeout)
-            .tls_certs_only(roots)
-            .identity(identity)
-            .build()
-            .map_err(Into::into)
-    }
-}
-
-#[derive(Clone)]
-pub(crate) struct HeartbeatStateStore {
-    path: PathBuf,
-}
-
-#[derive(Default, Serialize, Deserialize)]
-#[serde(deny_unknown_fields, rename_all = "camelCase")]
-struct HeartbeatState {
-    next_sequence: u64,
-    pending: Option<PendingHeartbeat>,
-}
-
-impl HeartbeatStateStore {
-    pub(crate) fn new(path: PathBuf) -> Self {
-        Self { path }
-    }
-
-    pub(crate) fn try_runtime_lock(&self) -> Result<fs::File, HeartbeatError> {
-        let directory = parent(&self.path)?;
-        fs::create_dir_all(directory).map_err(|source| state_io(directory, source))?;
-        let name = filename(&self.path)?;
-        let path = directory.join(format!(".{name}.lock"));
-        let mut options = fs::OpenOptions::new();
-        options.create(true).truncate(false).read(true).write(true);
-        #[cfg(unix)]
-        {
-            use std::os::unix::fs::OpenOptionsExt as _;
-            options.mode(FILE_MODE);
-        }
-        let lock = options.open(&path).map_err(|source| state_io(&path, source))?;
-        check_mode(&path)?;
-        lock.try_lock().map_err(|_| HeartbeatError::AlreadyRunning)?;
-        Ok(lock)
-    }
-
-    pub(crate) async fn prepare(
-        &self,
-        summary: CoarseNodeSummary,
-        now: DateTime<Utc>,
-    ) -> Result<PendingHeartbeat, HeartbeatError> {
-        let store = self.clone();
-        tokio::task::spawn_blocking(move || store.prepare_sync(summary, now))
-            .await
-            .map_err(|source| state_io(&self.path, io::Error::other(source)))?
-    }
-
-    pub(crate) async fn mark_accepted(&self, accepted: &PendingHeartbeat) -> Result<(), HeartbeatError> {
-        let store = self.clone();
-        let accepted = accepted.clone();
-        tokio::task::spawn_blocking(move || store.mark_accepted_sync(&accepted))
-            .await
-            .map_err(|source| state_io(&self.path, io::Error::other(source)))?
-    }
-
-    fn prepare_sync(&self, summary: CoarseNodeSummary, now: DateTime<Utc>) -> Result<PendingHeartbeat, HeartbeatError> {
-        let mut state = self.read()?;
-        if let Some(pending) = state.pending {
-            return Ok(pending);
-        }
-        if state.next_sequence > MAX_SEQUENCE {
-            return Err(HeartbeatError::SequenceExhausted);
-        }
-        let pending = PendingHeartbeat {
-            protocol_version: PROTOCOL_VERSION.to_owned(),
-            request_id: Uuid::new_v4().to_string(),
-            agent_version: AGENT_VERSION.to_owned(),
-            capabilities: ["heartbeat".to_owned()],
-            sequence: state.next_sequence,
-            client_time: now.to_rfc3339_opts(SecondsFormat::Secs, true),
-            coarse_node_summary: summary,
-        };
-        state.pending = Some(pending.clone());
-        self.write(&state)?;
-        Ok(pending)
-    }
-
-    fn mark_accepted_sync(&self, accepted: &PendingHeartbeat) -> Result<(), HeartbeatError> {
-        let mut state = self.read()?;
-        if state.pending.as_ref() != Some(accepted) {
-            return Err(HeartbeatError::StateConflict);
-        }
-        state.next_sequence = accepted.sequence.checked_add(1).ok_or(HeartbeatError::SequenceExhausted)?;
-        state.pending = None;
-        self.write(&state)
-    }
-
-    fn read(&self) -> Result<HeartbeatState, HeartbeatError> {
-        let bytes = match fs::read(&self.path) {
-            Ok(bytes) => bytes,
-            Err(source) if source.kind() == io::ErrorKind::NotFound => return Ok(HeartbeatState::default()),
-            Err(source) => return Err(state_io(&self.path, source)),
-        };
-        check_mode(&self.path)?;
-        let state: HeartbeatState = serde_json::from_slice(&bytes).map_err(|source| HeartbeatError::StateInvalid {
-            path: self.path.clone(),
-            source,
-        })?;
-        if state.next_sequence > MAX_SEQUENCE + 1
-            || state
-                .pending
-                .as_ref()
-                .is_some_and(|pending| pending.sequence != state.next_sequence || !pending.is_valid())
-        {
-            return Err(HeartbeatError::StateCorrupt { path: self.path.clone() });
-        }
-        Ok(state)
-    }
-
-    fn write(&self, state: &HeartbeatState) -> Result<(), HeartbeatError> {
-        let bytes = serde_json::to_vec(state).map_err(|source| HeartbeatError::StateInvalid {
-            path: self.path.clone(),
-            source,
-        })?;
-        let directory = parent(&self.path)?;
-        fs::create_dir_all(directory).map_err(|source| state_io(directory, source))?;
-        let temp = stage(directory, &self.path, &bytes)?;
-        let result = fs::rename(&temp, &self.path)
-            .map_err(|source| state_io(&self.path, source))
-            .and_then(|()| fsync_dir(directory).map_err(|source| state_io(directory, source)));
-        if result.is_err() {
-            let _ = fs::remove_file(temp);
-        }
-        result
-    }
-}
-
-fn cluster_uid(credential: &DeviceCredential) -> Result<&str, HeartbeatError> {
-    let mut parts = credential.name.split('/');
-    let valid = parts.next() == Some("organizations");
-    let organization_uid = parts.next();
-    let valid = valid && parts.next() == Some("clusters");
-    let cluster_uid = parts.next();
-    let valid = valid && parts.next() == Some("clusterDevices");
-    let device_uid = parts.next();
-    if !valid
-        || organization_uid.is_none_or(str::is_empty)
-        || cluster_uid.is_none_or(str::is_empty)
-        || device_uid != Some(credential.uid.as_str())
-        || parts.next().is_some()
-    {
-        return Err(HeartbeatError::CredentialName);
-    }
-    cluster_uid.ok_or(HeartbeatError::CredentialName)
-}
-
-fn retry_after(headers: &header::HeaderMap, now: DateTime<Utc>, maximum: Duration) -> Option<Duration> {
-    let value = headers.get(header::RETRY_AFTER)?.to_str().ok()?;
-    let delay = value.parse::<u64>().ok().map(Duration::from_secs).or_else(|| {
-        DateTime::parse_from_rfc2822(value)
-            .ok()
-            .and_then(|at| (at.with_timezone(&Utc) - now).to_std().ok())
-    })?;
-    Some(delay.min(maximum))
-}
-
-fn is_exact_utc_seconds(value: &str) -> bool {
-    DateTime::parse_from_rfc3339(value).is_ok_and(|time| {
-        time.offset().local_minus_utc() == 0
-            && value.ends_with('Z')
-            && time.with_timezone(&Utc).to_rfc3339_opts(SecondsFormat::Secs, true) == value
-    })
-}
-
-async fn response_reason(response: reqwest::Response) -> Option<String> {
-    #[derive(Deserialize)]
-    struct Envelope {
-        #[serde(default)]
-        details: Vec<Detail>,
-    }
-    #[derive(Deserialize)]
-    struct Detail {
-        #[serde(default)]
-        reason: String,
-    }
-
-    serde_json::from_slice::<Envelope>(&bounded_body(response).await.ok()?)
-        .ok()?
-        .details
-        .into_iter()
-        .find_map(|detail| (!detail.reason.is_empty()).then_some(detail.reason))
-}
-
-async fn bounded_body(mut response: reqwest::Response) -> Result<Vec<u8>, HeartbeatError> {
-    let mut body = Vec::new();
-    while let Some(chunk) = response.chunk().await? {
-        if body.len().saturating_add(chunk.len()) > MAX_RESPONSE_BYTES {
-            return Err(HeartbeatError::ResponseTooLarge);
-        }
-        body.extend_from_slice(&chunk);
-    }
-    Ok(body)
-}
-
-fn parent(path: &Path) -> Result<&Path, HeartbeatError> {
-    path.parent()
-        .ok_or_else(|| state_io(path, io::Error::new(io::ErrorKind::InvalidInput, "state path has no parent")))
-}
-
-fn filename(path: &Path) -> Result<&str, HeartbeatError> {
-    path.file_name()
-        .and_then(|name| name.to_str())
-        .ok_or_else(|| state_io(path, io::Error::new(io::ErrorKind::InvalidInput, "state filename is invalid")))
-}
-
-fn stage(directory: &Path, destination: &Path, bytes: &[u8]) -> Result<PathBuf, HeartbeatError> {
-    let name = filename(destination)?;
-    loop {
-        let path = directory.join(format!(
-            ".{name}.{}.{}.tmp",
-            std::process::id(),
-            STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed)
-        ));
-        let mut options = fs::OpenOptions::new();
-        options.write(true).create_new(true);
-        #[cfg(unix)]
-        {
-            use std::os::unix::fs::OpenOptionsExt as _;
-            options.mode(FILE_MODE);
-        }
-        let mut file = match options.open(&path) {
-            Ok(file) => file,
-            Err(source) if source.kind() == io::ErrorKind::AlreadyExists => continue,
-            Err(source) => return Err(state_io(&path, source)),
-        };
-        if let Err(source) = file.write_all(bytes).and_then(|()| file.sync_all()) {
-            let _ = fs::remove_file(&path);
-            return Err(state_io(&path, source));
-        }
-        return Ok(path);
-    }
-}
-
-fn state_io(path: &Path, source: io::Error) -> HeartbeatError {
-    HeartbeatError::StateIo {
-        path: path.to_path_buf(),
-        source,
-    }
-}
-
-#[cfg(unix)]
-fn check_mode(path: &Path) -> Result<(), HeartbeatError> {
-    use std::os::unix::fs::PermissionsExt as _;
-
-    let mode = fs::metadata(path)
-        .map_err(|source| state_io(path, source))?
-        .permissions()
-        .mode()
-        & 0o7777;
-    if mode != FILE_MODE {
-        return Err(HeartbeatError::StatePermissions {
-            path: path.to_path_buf(),
-            mode,
-            expected: FILE_MODE,
-        });
-    }
-    Ok(())
-}
-
-#[cfg(not(unix))]
-fn check_mode(_path: &Path) -> Result<(), HeartbeatError> {
-    Ok(())
-}
-
-fn fsync_dir(directory: &Path) -> io::Result<()> {
-    #[cfg(unix)]
-    fs::File::open(directory)?.sync_all()?;
-    #[cfg(not(unix))]
-    let _ = directory;
-    Ok(())
-}
-
-#[derive(Debug, thiserror::Error)]
-pub enum HeartbeatError {
-    #[error("Connect heartbeat endpoint must be an HTTPS base URL without credentials, query, or fragment")]
-    Endpoint,
-    #[error("Connect heartbeat root CA configuration is invalid")]
-    RootCertificate,
-    #[error("Connect heartbeat schedule is invalid")]
-    Schedule,
-    #[error("RustFS is not registered with Connect")]
-    NotRegistered,
-    #[error("the Connect device private key is missing")]
-    IdentityMissing,
-    #[error("the stored Connect certificate and device private key cannot form a TLS identity")]
-    IdentityCertificate,
-    #[error("the stored Connect credential name is invalid")]
-    CredentialName,
-    #[error("the stored Connect device certificate is not currently valid")]
-    CredentialExpired,
-    #[error("the Connect heartbeat node summary is outside protocol bounds")]
-    NodeSummary,
-    #[error("the Connect heartbeat sequence is exhausted")]
-    SequenceExhausted,
-    #[error("a Connect heartbeat runtime already owns this state")]
-    AlreadyRunning,
-    #[error("the persisted Connect heartbeat changed while delivery was in flight")]
-    StateConflict,
-    #[error("Connect heartbeat state I/O failed at {path}: {source}")]
-    StateIo {
-        path: PathBuf,
-        #[source]
-        source: io::Error,
-    },
-    #[error("Connect heartbeat state at {path} is invalid: {source}")]
-    StateInvalid {
-        path: PathBuf,
-        #[source]
-        source: serde_json::Error,
-    },
-    #[error("Connect heartbeat state at {path} violates the protocol invariants")]
-    StateCorrupt { path: PathBuf },
-    #[cfg(unix)]
-    #[error("Connect heartbeat state at {path} has mode {mode:o}, expected {expected:o}")]
-    StatePermissions { path: PathBuf, mode: u32, expected: u32 },
-    #[error("Connect heartbeat response exceeded 64 KiB")]
-    ResponseTooLarge,
-    #[error("Connect returned an invalid heartbeat response")]
-    Response,
-    #[error(transparent)]
-    Url(#[from] url::ParseError),
-    #[error(transparent)]
-    Transport(#[from] reqwest::Error),
-    #[error(transparent)]
-    Identity(#[from] IdentityError),
-    #[error(transparent)]
-    IdentityStore(#[from] StoreError),
-    #[error(transparent)]
-    CredentialStore(#[from] CredentialStoreError),
-    #[error(transparent)]
-    CredentialValidation(#[from] CredentialValidationError),
-}
@@ -21,26 +21,20 @@
 //! canonical transcript frozen by
 //! `protocol/agent/v1/registration-proof.md`.
 //!
-//! Enrolled deployments may start the optional outbound heartbeat runtime.
-//! An unconfigured server starts no Connect task, generates no key, and holds
-//! no Connect identity.
+//! Nothing here contacts the network or starts a task. A deployment that has
+//! not been enrolled into a Connect control plane never calls into it, so an
+//! unconfigured server generates no key and holds no identity.

 pub mod client;
-pub mod config;
 pub mod credential_store;
-pub mod heartbeat;
 pub mod identity;
 pub mod identity_store;
 pub mod offline;
 pub mod registration;
-pub mod runtime;

 pub use client::{ClientError, ConnectClient, ConnectConfig};
-pub use config::{HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule};
 pub use credential_store::{CredentialStore, DeviceCredential};
-pub use heartbeat::{CoarseNodeSummary, HeartbeatError, HeartbeatStatus};
 pub use identity::{DeviceIdentity, IdentityError, RegistrationProof, RegistrationTranscript};
 pub use identity_store::{IdentityStore, StoreError};
 pub use offline::{EnrollmentError, OfflineEnrollment, OfflineKeyStore, VerifiedChallenge};
 pub use registration::{RegistrationToken, TokenError};
-pub use runtime::{HeartbeatRuntime, spawn_heartbeat_runtime};
@@ -1,156 +0,0 @@
-// 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.
-
-use std::future::Future;
-use std::time::Duration;
-
-use chrono::Utc;
-use rand::RngExt as _;
-use tokio::sync::watch;
-use tokio::task::JoinHandle;
-use tokio_util::sync::CancellationToken;
-
-use super::config::HeartbeatConfig;
-use super::heartbeat::{CoarseNodeSummary, Delivery, HeartbeatError, HeartbeatSender, HeartbeatStateStore, HeartbeatStatus};
-
-pub struct HeartbeatRuntime {
-    shutdown: CancellationToken,
-    status: watch::Receiver<HeartbeatStatus>,
-    task: Option<JoinHandle<()>>,
-}
-
-impl HeartbeatRuntime {
-    pub fn status(&self) -> watch::Receiver<HeartbeatStatus> {
-        self.status.clone()
-    }
-
-    pub async fn shutdown(mut self) {
-        self.shutdown.cancel();
-        if let Some(task) = self.task.take() {
-            let _ = task.await;
-        }
-    }
-}
-
-impl Drop for HeartbeatRuntime {
-    fn drop(&mut self) {
-        self.shutdown.cancel();
-    }
-}
-
-pub fn spawn_heartbeat_runtime<F>(
-    config: Option<HeartbeatConfig>,
-    parent_shutdown: &CancellationToken,
-    sample: F,
-) -> Result<Option<HeartbeatRuntime>, HeartbeatError>
-where
-    F: Fn() -> CoarseNodeSummary + Send + Sync + 'static,
-{
-    let Some(config) = config else {
-        return Ok(None);
-    };
-    let sender = HeartbeatSender::new(config.clone())?;
-    let store = HeartbeatStateStore::new(config.state_path.clone());
-    let lock = store.try_runtime_lock()?;
-    let schedule = config.schedule;
-    let shutdown = parent_shutdown.child_token();
-    let task_shutdown = shutdown.clone();
-    let (status_tx, status_rx) = watch::channel(HeartbeatStatus::Starting);
-    let task = tokio::spawn(async move {
-        let _lock = lock;
-        let mut backoff = schedule.initial_backoff;
-        loop {
-            if task_shutdown.is_cancelled() {
-                break;
-            }
-            let pending = match store.prepare(sample(), Utc::now()).await {
-                Ok(pending) => pending,
-                Err(error) => return failed(&status_tx, error),
-            };
-            let delivery = match cancellable(&task_shutdown, sender.send(&pending)).await {
-                Some(Ok(delivery)) => delivery,
-                Some(Err(error)) => return failed(&status_tx, error),
-                None => break,
-            };
-            let delay = match delivery {
-                Delivery::Accepted { server_time } => {
-                    if let Err(error) = store.mark_accepted(&pending).await {
-                        return failed(&status_tx, error);
-                    }
-                    backoff = schedule.initial_backoff;
-                    let _ = status_tx.send(HeartbeatStatus::Online { server_time });
-                    schedule.cadence.saturating_add(jitter(schedule.jitter))
-                }
-                Delivery::Retry { retry_after } => {
-                    let delay = retry_after
-                        .unwrap_or(backoff)
-                        .clamp(schedule.initial_backoff, schedule.max_backoff);
-                    backoff = backoff.saturating_mul(2).min(schedule.max_backoff);
-                    let _ = status_tx.send(HeartbeatStatus::BackingOff { delay });
-                    delay
-                }
-                Delivery::AuthenticationStopped { status, reason } => {
-                    let _ = status_tx.send(HeartbeatStatus::AuthenticationStopped { status, reason });
-                    return;
-                }
-                Delivery::Rejected { status, reason } => {
-                    let suffix = reason.map_or_else(String::new, |reason| format!("; reason={reason}"));
-                    let _ = status_tx.send(HeartbeatStatus::Failed {
-                        reason: format!("Connect rejected heartbeat with HTTP {status}{suffix}"),
-                    });
-                    return;
-                }
-            };
-            if sleep_or_cancel(&task_shutdown, delay).await {
-                break;
-            }
-        }
-        let _ = status_tx.send(HeartbeatStatus::Stopped);
-    });
-    Ok(Some(HeartbeatRuntime {
-        shutdown,
-        status: status_rx,
-        task: Some(task),
-    }))
-}
-
-fn failed(status: &watch::Sender<HeartbeatStatus>, error: HeartbeatError) {
-    let _ = status.send(HeartbeatStatus::Failed {
-        reason: error.to_string(),
-    });
-}
-
-fn jitter(maximum: Duration) -> Duration {
-    if maximum.is_zero() {
-        Duration::ZERO
-    } else {
-        maximum.mul_f64(rand::rng().random_range(0.0..=1.0))
-    }
-}
-
-async fn cancellable<T>(shutdown: &CancellationToken, future: impl Future<Output = T>) -> Option<T> {
-    tokio::select! {
-        biased;
-        () = shutdown.cancelled() => None,
-        value = future => Some(value),
-    }
-}
-
-async fn sleep_or_cancel(shutdown: &CancellationToken, delay: Duration) -> bool {
-    tokio::select! {
-        biased;
-        () = shutdown.cancelled() => true,
-        () = tokio::time::sleep(delay) => false,
-    }
-}
@@ -128,7 +128,6 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
    } = lifecycle;
    let StartupServiceRuntime {
        optional_runtimes,
-        heartbeat,
        iam_bootstrap,
        enable_scanner,
    } = service_runtime;
@@ -163,9 +162,6 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
        shutdown_token,
    )
    .await;
-    if let Some(heartbeat) = heartbeat {
-        heartbeat.shutdown().await;
-    }
    if let Err(err) = event_notifier_reconciler.await {
        tracing::warn!(
            target: "rustfs::main::run",
@@ -16,7 +16,6 @@ use crate::site_replication_reconcile::spawn_site_replication_reconcile_task;
 use crate::storage_api::startup::services::{ECStore, EndpointServerPools, ServerContextSlot};
 use crate::{
    config::Config,
-    connect::{CoarseNodeSummary, HeartbeatConfig, HeartbeatRuntime, spawn_heartbeat_runtime},
    init::{init_buffer_profile_system, init_kms_system},
    server::ServiceStateManager,
    startup_audit::init_audit_runtime,
@@ -36,7 +35,6 @@ use tokio_util::sync::CancellationToken;

 pub(crate) struct StartupServiceRuntime {
    pub(crate) optional_runtimes: OptionalRuntimeServices,
-    pub(crate) heartbeat: Option<HeartbeatRuntime>,
    pub(crate) iam_bootstrap: IamBootstrapDisposition,
    pub(crate) enable_scanner: bool,
 }
@@ -75,8 +73,6 @@ pub(crate) async fn init_startup_runtime_services(
    init_kms_system(config).await?;

    let optional_runtimes = init_optional_runtime_services().await?;
-    let heartbeat_config = HeartbeatConfig::from_env().map_err(std::io::Error::other)?;
-    let heartbeat_nodes = heartbeat_config.as_ref().map(|_| endpoint_pools.get_nodes().len());

    init_buffer_profile_system(config);
    init_deadlock_detector_runtime();
@@ -96,27 +92,10 @@ pub(crate) async fn init_startup_runtime_services(
    init_notification_runtime(endpoint_pools, buckets).await?;
    let enable_scanner = init_background_service_runtime(store.clone()).await?;
    init_observability_runtime(store.clone(), ctx.clone()).await;
-    let heartbeat = start_heartbeat_runtime(heartbeat_config, heartbeat_nodes, &ctx)?;

    Ok(StartupServiceRuntime {
        optional_runtimes,
-        heartbeat,
        iam_bootstrap,
        enable_scanner,
    })
 }
-
-fn start_heartbeat_runtime(
-    config: Option<HeartbeatConfig>,
-    node_count: Option<usize>,
-    shutdown: &CancellationToken,
-) -> Result<Option<HeartbeatRuntime>> {
-    let Some(config) = config else {
-        return Ok(None);
-    };
-    let summary = u16::try_from(node_count.unwrap_or_default())
-        .ok()
-        .and_then(|total| CoarseNodeSummary::new(total, 0, 0).ok())
-        .ok_or_else(|| std::io::Error::other("Connect heartbeat node count is outside protocol bounds"))?;
-    spawn_heartbeat_runtime(Some(config), shutdown, move || summary).map_err(std::io::Error::other)
-}
@@ -1,687 +0,0 @@
-// 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.
-
-use std::collections::VecDeque;
-use std::fs;
-use std::path::Path;
-use std::sync::{Arc, Mutex};
-use std::time::Duration;
-
-use bytes::Bytes;
-use http_body_util::{BodyExt as _, Full};
-use hyper::service::service_fn;
-use hyper::{Request, Response, StatusCode};
-use hyper_util::rt::TokioIo;
-use rcgen::{
-    BasicConstraints, CertificateParams, DistinguishedName, DnType, ExtendedKeyUsagePurpose, IsCa, Issuer, KeyPair,
-    KeyUsagePurpose, SanType,
-};
-use rustfs::connect::{
-    CoarseNodeSummary, CredentialStore, DeviceCredential, HeartbeatConfig, HeartbeatSchedule, HeartbeatStatus, IdentityStore,
-    spawn_heartbeat_runtime,
-};
-use rustls::RootCertStore;
-use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
-use rustls::server::WebPkiClientVerifier;
-use serde_json::{Value, json};
-use time::OffsetDateTime;
-use tokio::net::TcpListener;
-use tokio::sync::watch;
-use tokio_rustls::TlsAcceptor;
-use tokio_util::sync::CancellationToken;
-
-const ORGANIZATION_UID: &str = "0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70";
-const CLUSTER_UID: &str = "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81";
-const DEVICE_UID: &str = "0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92";
-
-struct TestPki {
-    root_params: CertificateParams,
-    root_key: KeyPair,
-    root_der: CertificateDer<'static>,
-    root_pem: String,
-    server_der: CertificateDer<'static>,
-    server_key: PrivatePkcs8KeyDer<'static>,
-}
-
-impl TestPki {
-    fn new() -> Self {
-        let now = OffsetDateTime::now_utc();
-        let root_key = KeyPair::generate().expect("generate root key");
-        let mut root_params = CertificateParams::default();
-        root_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
-        root_params.not_before = now - time::Duration::days(30);
-        root_params.not_after = now + time::Duration::days(30);
-        root_params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::DigitalSignature];
-        let root = root_params.self_signed(&root_key).expect("sign root");
-
-        let server_key = KeyPair::generate().expect("generate server key");
-        let mut server_params = CertificateParams::default();
-        server_params.not_before = now - time::Duration::hours(1);
-        server_params.not_after = now + time::Duration::days(2);
-        server_params
-            .subject_alt_names
-            .push(SanType::DnsName("localhost".try_into().expect("valid DNS name")));
-        server_params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth];
-        let server = server_params
-            .signed_by(&server_key, &Issuer::from_params(&root_params, &root_key))
-            .expect("sign server certificate");
-        Self {
-            root_params,
-            root_key,
-            root_der: root.der().clone(),
-            root_pem: root.pem(),
-            server_der: server.der().clone(),
-            server_key: PrivatePkcs8KeyDer::from(server_key.serialize_der()),
-        }
-    }
-
-    fn server_config(&self) -> rustls::ServerConfig {
-        let mut roots = RootCertStore::empty();
-        roots.add(self.root_der.clone()).expect("add client root");
-        let verifier = WebPkiClientVerifier::builder(Arc::new(roots))
-            .build()
-            .expect("client verifier");
-        rustls::ServerConfig::builder()
-            .with_client_cert_verifier(verifier)
-            .with_single_cert(vec![self.server_der.clone()], PrivateKeyDer::Pkcs8(self.server_key.clone_key()))
-            .expect("server TLS")
-    }
-
-    fn stores(&self, temp: &tempfile::TempDir) -> (IdentityStore, CredentialStore) {
-        let now = OffsetDateTime::now_utc();
-        self.stores_with_certificate(temp, now - time::Duration::hours(1), now + time::Duration::hours(23), true)
-    }
-
-    fn stores_with_certificate(
-        &self,
-        temp: &tempfile::TempDir,
-        not_before: OffsetDateTime,
-        not_after: OffsetDateTime,
-        bind_identity: bool,
-    ) -> (IdentityStore, CredentialStore) {
-        let identity_store = IdentityStore::new(temp.path().join("identity"));
-        let identity = identity_store.load_or_create().expect("create identity");
-        let private_key = PrivatePkcs8KeyDer::from(identity.to_pkcs8_der().expect("serialize key").to_vec());
-        let device_key = if bind_identity {
-            KeyPair::from_pkcs8_der_and_sign_algo(&private_key, &rcgen::PKCS_ECDSA_P256_SHA256).expect("device key")
-        } else {
-            KeyPair::generate().expect("mismatched device key")
-        };
-        let mut params = CertificateParams::default();
-        params.not_before = not_before;
-        params.not_after = not_after;
-        params.serial_number = Some(vec![1; 16].into());
-        params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
-        params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ClientAuth];
-        params.distinguished_name = DistinguishedName::new();
-        params.distinguished_name.push(DnType::CommonName, DEVICE_UID);
-        params.subject_alt_names.push(SanType::URI(
-            format!("urn:rustfs:connect:device:{DEVICE_UID}")
-                .try_into()
-                .expect("device URI"),
-        ));
-        let certificate = params
-            .signed_by(&device_key, &Issuer::from_params(&self.root_params, &self.root_key))
-            .expect("device certificate");
-        let cluster = format!("organizations/{ORGANIZATION_UID}/clusters/{CLUSTER_UID}");
-        let credential = DeviceCredential {
-            name: format!("{cluster}/clusterDevices/{DEVICE_UID}"),
-            uid: DEVICE_UID.to_owned(),
-            protocol_version: "v1".to_owned(),
-            key_id: format!("x509-{}", "01".repeat(16)),
-            certificate_serial: "01".repeat(16),
-            certificate: certificate.pem(),
-            certificate_chain: certificate.pem(),
-            not_before_unix: not_before.unix_timestamp(),
-            not_after_unix: not_after.unix_timestamp(),
-        };
-        let directory = temp.path().join("credential");
-        fs::create_dir_all(&directory).expect("credential directory");
-        let path = directory.join("device.crt.json");
-        fs::write(&path, serde_json::to_vec(&credential).expect("credential JSON")).expect("write credential");
-        private_mode(&path);
-        (identity_store, CredentialStore::new(directory))
-    }
-}
-
-#[derive(Clone)]
-struct Reply {
-    status: StatusCode,
-    body: Value,
-    retry_after: Option<&'static str>,
-    delay: Duration,
-}
-
-impl Reply {
-    fn ok(time: &str) -> Self {
-        Self {
-            status: StatusCode::OK,
-            body: json!({
-                "serverTime": time,
-                "acceptedVersion": "v1",
-                "capabilityHints": [],
-                "futureField": true
-            }),
-            retry_after: None,
-            delay: Duration::ZERO,
-        }
-    }
-
-    fn error(status: StatusCode) -> Self {
-        Self {
-            status,
-            body: json!({"details": []}),
-            retry_after: None,
-            delay: Duration::ZERO,
-        }
-    }
-}
-
-struct TestServer {
-    endpoint: String,
-    seen: Arc<Mutex<Vec<Value>>>,
-    task: tokio::task::JoinHandle<()>,
-}
-
-impl Drop for TestServer {
-    fn drop(&mut self) {
-        self.task.abort();
-    }
-}
-
-async fn server(pki: &TestPki, replies: Vec<Reply>) -> TestServer {
-    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind server");
-    let address = listener.local_addr().expect("server address");
-    let acceptor = TlsAcceptor::from(Arc::new(pki.server_config()));
-    let replies = Arc::new(Mutex::new(VecDeque::from(replies)));
-    let seen = Arc::new(Mutex::new(Vec::new()));
-    let captured = seen.clone();
-    let task = tokio::spawn(async move {
-        while let Ok((stream, _)) = listener.accept().await {
-            let acceptor = acceptor.clone();
-            let replies = replies.clone();
-            let seen = captured.clone();
-            tokio::spawn(async move {
-                let Ok(stream) = acceptor.accept(stream).await else { return };
-                let service = service_fn(move |request: Request<hyper::body::Incoming>| {
-                    let replies = replies.clone();
-                    let seen = seen.clone();
-                    async move {
-                        assert_eq!(request.uri().path(), format!("/agent/clusters/{CLUSTER_UID}/heartbeats"));
-                        let body = request.into_body().collect().await.expect("request body").to_bytes();
-                        seen.lock()
-                            .expect("seen lock")
-                            .push(serde_json::from_slice(&body).expect("request JSON"));
-                        let reply = replies
-                            .lock()
-                            .expect("reply lock")
-                            .pop_front()
-                            .unwrap_or_else(|| Reply::error(StatusCode::SERVICE_UNAVAILABLE));
-                        if !reply.delay.is_zero() {
-                            tokio::time::sleep(reply.delay).await;
-                        }
-                        let mut builder = Response::builder()
-                            .status(reply.status)
-                            .header("content-type", "application/json");
-                        if let Some(value) = reply.retry_after {
-                            builder = builder.header("retry-after", value);
-                        }
-                        Ok::<_, hyper::Error>(
-                            builder
-                                .body(Full::new(Bytes::from(serde_json::to_vec(&reply.body).expect("reply JSON"))))
-                                .expect("reply"),
-                        )
-                    }
-                });
-                let _ = hyper::server::conn::http1::Builder::new()
-                    .serve_connection(TokioIo::new(stream), service)
-                    .await;
-            });
-        }
-    });
-    TestServer {
-        endpoint: format!("https://localhost:{}/agent/", address.port()),
-        seen,
-        task,
-    }
-}
-
-fn config(temp: &tempfile::TempDir, pki: &TestPki, server: &TestServer) -> HeartbeatConfig {
-    let (identity_store, credential_store) = pki.stores(temp);
-    config_with_stores(temp, pki, server, identity_store, credential_store)
-}
-
-fn config_with_stores(
-    temp: &tempfile::TempDir,
-    pki: &TestPki,
-    server: &TestServer,
-    identity_store: IdentityStore,
-    credential_store: CredentialStore,
-) -> HeartbeatConfig {
-    HeartbeatConfig {
-        endpoint: server.endpoint.clone(),
-        root_ca_pem: pki.root_pem.as_bytes().to_vec(),
-        identity_store,
-        credential_store,
-        state_path: temp.path().join("heartbeat/state.json"),
-        schedule: HeartbeatSchedule {
-            cadence: Duration::from_millis(40),
-            jitter: Duration::ZERO,
-            timeout: Duration::from_millis(200),
-            initial_backoff: Duration::from_millis(20),
-            max_backoff: Duration::from_millis(80),
-        },
-    }
-}
-
-fn rewrite_credential(temp: &tempfile::TempDir, update: impl FnOnce(&mut DeviceCredential)) {
-    let path = temp.path().join("credential/device.crt.json");
-    let mut credential: DeviceCredential =
-        serde_json::from_slice(&fs::read(&path).expect("read credential")).expect("parse credential");
-    update(&mut credential);
-    fs::write(&path, serde_json::to_vec(&credential).expect("credential JSON")).expect("rewrite credential");
-    private_mode(&path);
-}
-
-fn summary() -> CoarseNodeSummary {
-    CoarseNodeSummary::new(8, 7, 1).expect("node summary")
-}
-
-async fn wait_for(
-    status: &mut watch::Receiver<HeartbeatStatus>,
-    predicate: impl Fn(&HeartbeatStatus) -> bool,
-) -> HeartbeatStatus {
-    tokio::time::timeout(Duration::from_secs(3), async {
-        loop {
-            let current = status.borrow_and_update().clone();
-            if predicate(&current) {
-                return current;
-            }
-            status.changed().await.expect("status channel");
-        }
-    })
-    .await
-    .expect("heartbeat status timeout")
-}
-
-async fn assert_credential_failure(config: HeartbeatConfig, server: &TestServer, expected: &str) {
-    let shutdown = CancellationToken::new();
-    let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, summary)
-        .expect("start runtime")
-        .expect("configured runtime");
-    let mut status = runtime.status();
-    assert!(matches!(
-        wait_for(&mut status, |status| matches!(status, HeartbeatStatus::Failed { .. })).await,
-        HeartbeatStatus::Failed { reason } if reason.contains(expected)
-    ));
-    assert!(server.seen.lock().expect("seen lock").is_empty());
-    runtime.shutdown().await;
-}
-
-#[tokio::test]
-async fn connect_config_absent_starts_no_task() {
-    let shutdown = CancellationToken::new();
-    let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
-    let sampled = calls.clone();
-    let runtime = spawn_heartbeat_runtime(None, &shutdown, move || {
-        sampled.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        summary()
-    })
-    .expect("absent config");
-
-    assert!(runtime.is_none());
-    tokio::task::yield_now().await;
-    assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 0);
-}
-
-#[tokio::test]
-async fn duplicate_runtime_is_rejected_without_a_second_task() {
-    let pki = TestPki::new();
-    let mut reply = Reply::ok("2026-08-22T01:02:03Z");
-    reply.delay = Duration::from_secs(5);
-    let server = server(&pki, vec![reply]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let shutdown = CancellationToken::new();
-    let config = config(&temp, &pki, &server);
-    let runtime = spawn_heartbeat_runtime(Some(config.clone()), &shutdown, summary)
-        .expect("first runtime")
-        .expect("configured runtime");
-
-    assert!(matches!(
-        spawn_heartbeat_runtime(Some(config), &shutdown, summary),
-        Err(rustfs::connect::HeartbeatError::AlreadyRunning)
-    ));
-    runtime.shutdown().await;
-}
-
-#[tokio::test(flavor = "current_thread")]
-async fn dropped_runtime_keeps_the_lock_until_its_task_stops() {
-    let pki = TestPki::new();
-    let server = server(&pki, vec![Reply::ok("2026-08-22T01:02:03Z")]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let shutdown = CancellationToken::new();
-    let config = config(&temp, &pki, &server);
-    let runtime = spawn_heartbeat_runtime(Some(config.clone()), &shutdown, summary)
-        .expect("first runtime")
-        .expect("configured runtime");
-
-    drop(runtime);
-    assert!(matches!(
-        spawn_heartbeat_runtime(Some(config.clone()), &shutdown, summary),
-        Err(rustfs::connect::HeartbeatError::AlreadyRunning)
-    ));
-
-    let replacement = tokio::time::timeout(Duration::from_secs(3), async {
-        loop {
-            match spawn_heartbeat_runtime(Some(config.clone()), &shutdown, summary) {
-                Ok(Some(runtime)) => break runtime,
-                Err(rustfs::connect::HeartbeatError::AlreadyRunning) => tokio::task::yield_now().await,
-                Ok(None) => panic!("configured replacement returned no runtime"),
-                Err(error) => panic!("unexpected replacement error: {error}"),
-            }
-        }
-    })
-    .await
-    .expect("dropped runtime releases its lock after stopping");
-    replacement.shutdown().await;
-}
-
-#[tokio::test]
-async fn corrupt_persisted_state_is_rejected_before_network_delivery() {
-    let pki = TestPki::new();
-    let server = server(&pki, vec![Reply::ok("2026-08-22T01:02:03Z")]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let shutdown = CancellationToken::new();
-    let config = config(&temp, &pki, &server);
-    let directory = config.state_path.parent().expect("state directory");
-    fs::create_dir_all(directory).expect("create state directory");
-    fs::write(
-        &config.state_path,
-        br#"{"nextSequence":0,"pending":{"protocolVersion":"v1","requestId":"550e8400-e29b-41d4-a716-446655440000","agentVersion":"rustfs-agent/1.0.0-rc.3","capabilities":["heartbeat"],"sequence":0,"clientTime":"2026-08-22T01:02:03Z","coarseNodeSummary":{"total":0,"healthy":0,"degraded":0}}}"#,
-    )
-    .expect("write corrupt state");
-    private_mode(&config.state_path);
-    let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, summary)
-        .expect("start runtime")
-        .expect("configured runtime");
-    let mut status = runtime.status();
-
-    assert!(matches!(
-        wait_for(&mut status, |status| matches!(status, HeartbeatStatus::Failed { .. })).await,
-        HeartbeatStatus::Failed { reason } if reason.contains("violates the protocol invariants")
-    ));
-    assert!(server.seen.lock().expect("seen lock").is_empty());
-    runtime.shutdown().await;
-}
-
-#[tokio::test]
-async fn invalid_stored_resource_name_is_rejected_before_network_delivery() {
-    let pki = TestPki::new();
-    let server = server(&pki, vec![]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let config = config(&temp, &pki, &server);
-    rewrite_credential(&temp, |credential| {
-        credential.name = format!("organizations/{ORGANIZATION_UID}/clusters/not-a-uuid/clusterDevices/{DEVICE_UID}");
-    });
-
-    assert_credential_failure(config, &server, "wrong device identity").await;
-}
-
-#[tokio::test]
-async fn invalid_stored_protocol_is_rejected_before_network_delivery() {
-    let pki = TestPki::new();
-    let server = server(&pki, vec![]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let config = config(&temp, &pki, &server);
-    rewrite_credential(&temp, |credential| credential.protocol_version = "v2".to_owned());
-
-    assert_credential_failure(config, &server, "wrong device identity").await;
-}
-
-#[tokio::test]
-async fn stored_certificate_key_mismatch_is_rejected_before_network_delivery() {
-    let pki = TestPki::new();
-    let server = server(&pki, vec![]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let now = OffsetDateTime::now_utc();
-    let (identity_store, credential_store) =
-        pki.stores_with_certificate(&temp, now - time::Duration::hours(1), now + time::Duration::hours(23), false);
-    let config = config_with_stores(&temp, &pki, &server, identity_store, credential_store);
-
-    assert_credential_failure(config, &server, "different device key").await;
-}
-
-#[tokio::test]
-async fn expired_stored_certificate_is_rejected_before_network_delivery() {
-    let pki = TestPki::new();
-    let server = server(&pki, vec![]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let now = OffsetDateTime::now_utc();
-    let (identity_store, credential_store) =
-        pki.stores_with_certificate(&temp, now - time::Duration::days(2), now - time::Duration::days(1), true);
-    let config = config_with_stores(&temp, &pki, &server, identity_store, credential_store);
-
-    assert_credential_failure(config, &server, "not currently valid").await;
-}
-
-#[tokio::test]
-async fn sends_only_l0_fields_and_accepts_additive_response_fields() {
-    let pki = TestPki::new();
-    let server = server(&pki, vec![Reply::ok("2038-01-19T03:14:07Z")]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let shutdown = CancellationToken::new();
-    let runtime = spawn_heartbeat_runtime(Some(config(&temp, &pki, &server)), &shutdown, summary)
-        .expect("start runtime")
-        .expect("configured runtime");
-    let mut status = runtime.status();
-
-    assert_eq!(
-        wait_for(&mut status, |status| matches!(status, HeartbeatStatus::Online { .. })).await,
-        HeartbeatStatus::Online {
-            server_time: "2038-01-19T03:14:07Z".to_owned()
-        }
-    );
-    runtime.shutdown().await;
-    let seen = server.seen.lock().expect("seen lock");
-    let request = &seen[0];
-    let mut keys = request
-        .as_object()
-        .expect("heartbeat object")
-        .keys()
-        .map(String::as_str)
-        .collect::<Vec<_>>();
-    keys.sort_unstable();
-    assert_eq!(
-        keys,
-        [
-            "agentVersion",
-            "capabilities",
-            "clientTime",
-            "coarseNodeSummary",
-            "protocolVersion",
-            "requestId",
-            "sequence"
-        ]
-    );
-    assert_eq!(request["capabilities"], json!(["heartbeat"]));
-    assert_eq!(request["coarseNodeSummary"], json!({"total": 8, "healthy": 7, "degraded": 1}));
-    assert_ne!(request["clientTime"], "2038-01-19T03:14:07Z");
-    assert!(request.get("authorization").is_none());
-}
-
-#[tokio::test]
-async fn restart_replays_pending_request_then_advances_sequence() {
-    let pki = TestPki::new();
-    let first_server = server(&pki, vec![Reply::error(StatusCode::SERVICE_UNAVAILABLE)]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let shutdown = CancellationToken::new();
-    let first_config = config(&temp, &pki, &first_server);
-    let runtime = spawn_heartbeat_runtime(Some(first_config.clone()), &shutdown, summary)
-        .expect("start runtime")
-        .expect("configured runtime");
-    let mut status = runtime.status();
-    wait_for(&mut status, |status| matches!(status, HeartbeatStatus::BackingOff { .. })).await;
-    runtime.shutdown().await;
-    let first = first_server.seen.lock().expect("seen lock")[0].clone();
-    drop(first_server);
-
-    let second_server = server(&pki, vec![Reply::ok("2026-08-22T01:02:03Z"), Reply::ok("2026-08-22T01:02:04Z")]).await;
-    let mut second_config = first_config;
-    second_config.endpoint = second_server.endpoint.clone();
-    let runtime = spawn_heartbeat_runtime(Some(second_config), &shutdown, summary)
-        .expect("restart runtime")
-        .expect("configured runtime");
-    tokio::time::timeout(Duration::from_secs(3), async {
-        while second_server.seen.lock().expect("seen lock").len() < 2 {
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .expect("two heartbeats");
-    runtime.shutdown().await;
-
-    let seen = second_server.seen.lock().expect("seen lock");
-    assert_eq!(seen[0]["requestId"], first["requestId"]);
-    assert_eq!(seen[0]["sequence"], first["sequence"]);
-    assert_ne!(seen[1]["requestId"], seen[0]["requestId"]);
-    assert_eq!(seen[1]["sequence"].as_u64(), seen[0]["sequence"].as_u64().map(|value| value + 1));
-}
-
-#[tokio::test]
-async fn retry_after_is_respected_with_the_local_upper_bound() {
-    let pki = TestPki::new();
-    let mut reply = Reply::error(StatusCode::TOO_MANY_REQUESTS);
-    reply.retry_after = Some("300");
-    let server = server(&pki, vec![reply]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let shutdown = CancellationToken::new();
-    let runtime = spawn_heartbeat_runtime(Some(config(&temp, &pki, &server)), &shutdown, summary)
-        .expect("start runtime")
-        .expect("configured runtime");
-    let mut status = runtime.status();
-    assert_eq!(
-        wait_for(&mut status, |status| matches!(status, HeartbeatStatus::BackingOff { .. })).await,
-        HeartbeatStatus::BackingOff {
-            delay: Duration::from_millis(80)
-        }
-    );
-    runtime.shutdown().await;
-}
-
-#[tokio::test]
-async fn disconnects_use_exponential_backoff_with_a_cap() {
-    let pki = TestPki::new();
-    let server = server(
-        &pki,
-        vec![
-            Reply::error(StatusCode::SERVICE_UNAVAILABLE),
-            Reply::error(StatusCode::SERVICE_UNAVAILABLE),
-            Reply::error(StatusCode::SERVICE_UNAVAILABLE),
-        ],
-    )
-    .await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let shutdown = CancellationToken::new();
-    let runtime = spawn_heartbeat_runtime(Some(config(&temp, &pki, &server)), &shutdown, summary)
-        .expect("start runtime")
-        .expect("configured runtime");
-    let mut status = runtime.status();
-    for delay in [20, 40, 80] {
-        assert_eq!(
-            wait_for(&mut status, |status| {
-                matches!(status, HeartbeatStatus::BackingOff { delay: observed } if *observed == Duration::from_millis(delay))
-            })
-            .await,
-            HeartbeatStatus::BackingOff {
-                delay: Duration::from_millis(delay)
-            }
-        );
-    }
-    runtime.shutdown().await;
-}
-
-#[tokio::test]
-async fn revoked_credential_stops_and_exposes_local_status() {
-    let pki = TestPki::new();
-    let mut reply = Reply::error(StatusCode::UNAUTHORIZED);
-    reply.body = json!({"details": [{"reason": "CREDENTIAL_REVOKED"}]});
-    let server = server(&pki, vec![reply]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let shutdown = CancellationToken::new();
-    let runtime = spawn_heartbeat_runtime(Some(config(&temp, &pki, &server)), &shutdown, summary)
-        .expect("start runtime")
-        .expect("configured runtime");
-    let mut status = runtime.status();
-    assert_eq!(
-        wait_for(&mut status, |status| matches!(status, HeartbeatStatus::AuthenticationStopped { .. })).await,
-        HeartbeatStatus::AuthenticationStopped {
-            status: 401,
-            reason: Some("CREDENTIAL_REVOKED".to_owned())
-        }
-    );
-    tokio::time::sleep(Duration::from_millis(100)).await;
-    assert_eq!(server.seen.lock().expect("seen lock").len(), 1);
-    runtime.shutdown().await;
-}
-
-#[tokio::test]
-async fn shutdown_cancels_an_in_flight_request() {
-    let pki = TestPki::new();
-    let mut reply = Reply::ok("2026-08-22T01:02:03Z");
-    reply.delay = Duration::from_secs(5);
-    let server = server(&pki, vec![reply]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
-    let shutdown = CancellationToken::new();
-    let runtime = spawn_heartbeat_runtime(Some(config(&temp, &pki, &server)), &shutdown, summary)
-        .expect("start runtime")
-        .expect("configured runtime");
-    tokio::time::timeout(Duration::from_secs(3), async {
-        while server.seen.lock().expect("seen lock").is_empty() {
-            tokio::task::yield_now().await;
-        }
-    })
-    .await
-    .expect("request reached server");
-    tokio::time::timeout(Duration::from_millis(250), runtime.shutdown())
-        .await
-        .expect("cancellable shutdown");
-}
-
-#[test]
-fn consumes_the_frozen_heartbeat_fixtures() {
-    let registry: Value =
-        serde_json::from_str(include_str!("../../protocol/agent/v1/fixtures/fixture-sets.json")).expect("fixture registry");
-    let heartbeat = registry["sets"]
-        .as_array()
-        .expect("fixture sets")
-        .iter()
-        .find(|set| set["name"] == "heartbeat")
-        .expect("heartbeat fixture set");
-    assert_eq!(heartbeat["status"], "populated");
-    let valid: Value =
-        serde_json::from_str(include_str!("../../protocol/agent/v1/fixtures/heartbeat/valid.json")).expect("valid fixture");
-    assert_eq!(valid["request"]["protocolVersion"], "v1");
-    let overflow: Value =
-        serde_json::from_str(include_str!("../../protocol/agent/v1/fixtures/heartbeat/overflow.json")).expect("overflow fixture");
-    assert_eq!(overflow["expected"]["httpStatus"], 422);
-}
-
-#[cfg(unix)]
-fn private_mode(path: &Path) {
-    use std::os::unix::fs::PermissionsExt as _;
-    fs::set_permissions(path, fs::Permissions::from_mode(0o600)).expect("private mode");
-}
-
-#[cfg(not(unix))]
-fn private_mode(_path: &Path) {}