feat(connect): emit durable heartbeats

2026-08-22 12:26:37 +00:00 · 2026-08-22 15:54:42 +08:00
parent 1a3be70d98
commit e13049a846
14 changed files with 1709 additions and 4 deletions
@@ -25,7 +25,7 @@
    },
    {
      "name": "heartbeat",
-      "status": "reserved",
+      "status": "populated",
      "purpose": "Heartbeat payloads, Connect receive time, and freshness window behavior."
    },
    {
@@ -0,0 +1,5 @@
+975c1ca53eefeef6766a6fc0b3d3281f7408255342b0686e5e2aee5ad055414c  duplicate.json
+963529a38a02849c6c2acc6d72668dca9f63218b49c89fae41a451b584850411  overflow.json
+e3adeee1c8a19aa17e70894896fb79c072e3785bea3611b93c11e79f039ed5af  stale.json
+35b9cebd8525389a701e8fe69fbe96407bcb31aa28392fe95babf4a4886985ad  unknown.json
+37941735dbd6ad3d238258a7b2cae6f0b3aa0ecaae1d8817817c3d718d11d633  valid.json
@@ -0,0 +1,9 @@
+{
+  "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}
+}
@@ -0,0 +1,11 @@
+{
+  "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"}
+}
@@ -0,0 +1,9 @@
+{
+  "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}
+}
@@ -0,0 +1,18 @@
+{
+  "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": []
+  }
+}
@@ -0,0 +1,21 @@
+{
+  "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"
+  }
+}
@@ -0,0 +1,173 @@
+// 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");
+    }
+}
@@ -0,0 +1,585 @@
+// 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,20 +21,26 @@
 //! canonical transcript frozen by
 //! `protocol/agent/v1/registration-proof.md`.
 //!
-//! 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.
+//! Enrolled deployments may start the optional outbound heartbeat runtime.
+//! An unconfigured server starts no Connect task, generates no key, and holds
+//! no Connect 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};
@@ -0,0 +1,156 @@
+// 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,6 +128,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
    } = lifecycle;
    let StartupServiceRuntime {
        optional_runtimes,
+        heartbeat,
        iam_bootstrap,
        enable_scanner,
    } = service_runtime;
@@ -162,6 +163,9 @@ 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,6 +16,7 @@ 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,
@@ -35,6 +36,7 @@ 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,
 }
@@ -73,6 +75,8 @@ 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();
@@ -92,10 +96,27 @@ 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)
+}
@@ -0,0 +1,687 @@
+// 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) {}