feat(connect): emit durable heartbeats

fix(ecstore): preserve remote delete error types (#6371 )
fix(heal): retain displaced task status (#6370 )
2026-08-22 12:26:37 +00:00 · 2026-08-22 18:22:33 +08:00 · 2026-08-22 17:07:02 +08:00 · 2026-08-22 08:23:40 +00:00
24 changed files with 2306 additions and 324 deletions
@@ -50,10 +50,10 @@ use rustfs_protos::evict_failed_connection;
 use rustfs_protos::proto_gen::node_service::RenamePartRequest;
 use rustfs_protos::proto_gen::node_service::{
    BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, DeletePathsRequest, DeleteRequest,
-    DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest,
-    MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest,
-    ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest,
-    RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
+    DeleteVersionRequest, DeleteVersionsRequest, DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest,
+    ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest,
+    ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest,
+    RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
    SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
    WriteMetadataRequest, node_service_client::NodeServiceClient,
 };
@@ -112,6 +112,28 @@ const EVENT_REMOTE_DISK_RPC: &str = "remote_disk_rpc";
 const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
 pub const REMOTE_SNAPSHOT_LEASE_TTL: Duration = Duration::from_secs(60);

+fn decode_delete_versions_errors(response: DeleteVersionsResponse, expected_len: usize) -> Vec<Option<Error>> {
+    if !response.item_errors.is_empty() {
+        if response.item_errors.len() != expected_len {
+            return vec![Some(Error::other("malformed delete_versions item errors")); expected_len];
+        }
+        return response
+            .item_errors
+            .into_iter()
+            .map(|error| (error.code != 0).then(|| error.into()))
+            .collect();
+    }
+
+    if response.errors.len() != expected_len {
+        return vec![Some(Error::other("malformed delete_versions errors")); expected_len];
+    }
+    response
+        .errors
+        .into_iter()
+        .map(|error| (!error.is_empty()).then(|| Error::other(error)))
+        .collect()
+}
+
 fn snapshot_lease_token_from_response(response: SnapshotLeaseResponse) -> Result<SnapshotLeaseToken> {
    if !response.success {
        return Err(response.error.unwrap_or_default().into());
@@ -2406,8 +2428,6 @@ impl DiskAPI for RemoteDisk {
            return errors;
        }

-        // TODO(backlog): replace string errors with typed `StorageError` variants
-
        let result = self
            .execute_with_timeout(
                || async {
@@ -2439,17 +2459,7 @@ impl DiskAPI for RemoteDisk {
            }
            return errors;
        }
-        response
-            .errors
-            .iter()
-            .map(|error| {
-                if error.is_empty() {
-                    None
-                } else {
-                    Some(Error::other(error.to_string()))
-                }
-            })
-            .collect()
+        decode_delete_versions_errors(response, versions.len())
    }

    #[tracing::instrument(level = "trace", skip_all)]
@@ -3760,6 +3770,63 @@ mod tests {

    static INIT: Once = Once::new();

+    #[test]
+    fn delete_versions_response_preserves_typed_item_errors() {
+        let errors = decode_delete_versions_errors(
+            DeleteVersionsResponse {
+                success: true,
+                errors: vec!["file not found".to_string(), String::new()],
+                error: None,
+                item_errors: vec![
+                    rustfs_protos::proto_gen::node_service::Error {
+                        code: DiskError::FileNotFound.to_u32(),
+                        error_info: "file not found".to_string(),
+                    },
+                    rustfs_protos::proto_gen::node_service::Error::default(),
+                ],
+            },
+            2,
+        );
+
+        assert!(matches!(errors.as_slice(), [Some(DiskError::FileNotFound), None]));
+    }
+
+    #[test]
+    fn delete_versions_response_accepts_legacy_string_errors() {
+        let errors = decode_delete_versions_errors(
+            DeleteVersionsResponse {
+                success: true,
+                errors: vec!["legacy error".to_string(), String::new()],
+                error: None,
+                item_errors: Vec::new(),
+            },
+            2,
+        );
+
+        assert_eq!(errors.len(), 2);
+        assert_eq!(errors[0].as_ref().map(ToString::to_string).as_deref(), Some("io error legacy error"));
+        assert!(errors[1].is_none());
+    }
+
+    #[test]
+    fn delete_versions_response_rejects_misaligned_item_errors() {
+        let errors = decode_delete_versions_errors(
+            DeleteVersionsResponse {
+                success: true,
+                errors: vec!["file not found".to_string()],
+                error: None,
+                item_errors: vec![rustfs_protos::proto_gen::node_service::Error {
+                    code: DiskError::FileNotFound.to_u32(),
+                    error_info: "file not found".to_string(),
+                }],
+            },
+            2,
+        );
+
+        assert_eq!(errors.len(), 2);
+        assert!(errors.iter().all(Option::is_some));
+    }
+
    #[test]
    fn disk_mutation_digest_marks_rolling_compatibility() {
        let mut request = Request::new(());
@@ -297,16 +297,10 @@ impl ECStore {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use crate::bucket::metadata_sys;
    use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
-    use crate::disk::{DeleteOptions, DiskOption, format::FormatV3, new_disk};
-    use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
-    use crate::runtime::instance::InstanceContext;
-    use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
-    use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations};
+    use crate::disk::{DiskOption, format::FormatV3, new_disk};
+    use crate::layout::endpoints::{Endpoints, PoolEndpoints};
    use crate::store::init_format::{load_format_erasure, save_format_file};
-    use crate::store::init_local_disks_with_instance_ctx;
-    use tokio_util::sync::CancellationToken;

    async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
        let format = FormatV3::new(1, 1);
@@ -353,51 +347,6 @@ mod tests {
        }
    }

-    async fn multi_pool_heal_store() -> (tempfile::TempDir, Arc<ECStore>, CancellationToken) {
-        let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created");
-        let mut pool_endpoints = Vec::new();
-        for pool_index in 0..2 {
-            let mut endpoints = Vec::new();
-            for disk_index in 0..4 {
-                let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
-                tokio::fs::create_dir_all(&disk_path)
-                    .await
-                    .expect("multi-pool heal test disk should be created");
-                let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8"))
-                    .expect("test endpoint should parse");
-                endpoint.set_pool_index(pool_index);
-                endpoint.set_set_index(0);
-                endpoint.set_disk_index(disk_index);
-                endpoints.push(endpoint);
-            }
-            pool_endpoints.push(PoolEndpoints {
-                legacy: false,
-                set_count: 1,
-                drives_per_set: 4,
-                endpoints: Endpoints::from(endpoints),
-                cmd_line: format!("heal-owner-pool-{pool_index}"),
-                platform: "test".to_string(),
-            });
-        }
-
-        let endpoint_pools = EndpointServerPools::from(pool_endpoints);
-        let instance_ctx = Arc::new(InstanceContext::new());
-        init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
-            .await
-            .expect("multi-pool local disks should initialize");
-        let shutdown = CancellationToken::new();
-        let store = ECStore::new_with_instance_ctx(
-            "127.0.0.1:0".parse().expect("test address should parse"),
-            endpoint_pools,
-            shutdown.clone(),
-            instance_ctx,
-        )
-        .await
-        .expect("multi-pool test store should initialize");
-        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
-        (temp_dir, store, shutdown)
-    }
-
    #[tokio::test]
    async fn heal_object_pool_scope_selects_only_requested_pool() {
        let store = minimal_heal_store().await;
@@ -557,229 +506,6 @@ mod tests {
        }
    }

-    #[tokio::test]
-    #[serial_test::serial]
-    async fn unscoped_heal_object_suspended_owner_semantics() {
-        let (_temp_dir, store, shutdown) = multi_pool_heal_store().await;
-        let bucket = format!("heal-owner-{}", Uuid::new_v4().simple());
-        let active_object = "active-owner";
-        let suspended_only_object = "suspended-only";
-        let duplicate_object = "duplicate-owner";
-        let marker_object = "marker-owner";
-        let quorum_object = "quorum-owner";
-        store
-            .make_bucket(&bucket, &MakeBucketOptions::default())
-            .await
-            .expect("bucket should be created in all pools");
-
-        let mut active_reader = PutObjReader::from_vec(b"active owner".to_vec());
-        store.pools[0]
-            .put_object(&bucket, active_object, &mut active_reader, &ObjectOptions::default())
-            .await
-            .expect("active owner object should be written");
-        let active_disks = store.pools[0].disk_set[0].disks.read().await.clone();
-        let missing_active_disk = active_disks[0].clone().expect("active disk should be online");
-        missing_active_disk
-            .delete(
-                &bucket,
-                active_object,
-                DeleteOptions {
-                    recursive: true,
-                    immediate: true,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("active owner shard should be removed for repair");
-        assert!(
-            missing_active_disk.read_xl(&bucket, active_object, false).await.is_err(),
-            "the active owner fixture must start with one missing metadata copy"
-        );
-
-        let mut suspended_reader = PutObjReader::from_vec(b"suspended owner".to_vec());
-        store.pools[1]
-            .put_object(&bucket, suspended_only_object, &mut suspended_reader, &ObjectOptions::default())
-            .await
-            .expect("suspended owner object should be written");
-        for (pool_index, mod_time) in [1_i64, 2_i64].into_iter().enumerate() {
-            let mut duplicate_reader = PutObjReader::from_vec(format!("duplicate-pool-{pool_index}").into_bytes());
-            store.pools[pool_index]
-                .put_object(
-                    &bucket,
-                    duplicate_object,
-                    &mut duplicate_reader,
-                    &ObjectOptions {
-                        mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(mod_time)),
-                        ..Default::default()
-                    },
-                )
-                .await
-                .expect("duplicate owner object should be written");
-        }
-        let duplicate_missing_disk = store.pools[0].disk_set[0].disks.read().await[0]
-            .clone()
-            .expect("duplicate active owner disk should be online");
-        duplicate_missing_disk
-            .delete(
-                &bucket,
-                duplicate_object,
-                DeleteOptions {
-                    recursive: true,
-                    immediate: true,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("duplicate active owner shard should be removed for repair");
-        let history_version = Uuid::new_v4();
-        let mut history_reader = PutObjReader::from_vec(b"marker history".to_vec());
-        store.pools[0]
-            .put_object(
-                &bucket,
-                marker_object,
-                &mut history_reader,
-                &ObjectOptions {
-                    versioned: true,
-                    version_id: Some(history_version.to_string()),
-                    mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1)),
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("versioned marker history should be written");
-        store.pools[0]
-            .delete_object(
-                &bucket,
-                marker_object,
-                ObjectOptions {
-                    versioned: true,
-                    mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(2)),
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("delete marker should be written");
-        let mut quorum_reader = PutObjReader::from_vec(b"quorum boundary".to_vec());
-        store.pools[0]
-            .put_object(&bucket, quorum_object, &mut quorum_reader, &ObjectOptions::default())
-            .await
-            .expect("quorum boundary object should be written");
-        {
-            let mut pool_meta = store.pool_meta.write().await;
-            let mut next = PoolMeta::new(&store.pools, &pool_meta);
-            next.pools[1].decommission = Some(PoolDecommissionInfo {
-                start_time: Some(OffsetDateTime::UNIX_EPOCH),
-                ..Default::default()
-            });
-            *pool_meta = next;
-        }
-
-        let (_, duplicate_owner) = store
-            .get_latest_object_info_with_idx(&bucket, duplicate_object, &ObjectOptions::default())
-            .await
-            .expect("duplicate owner should resolve");
-        assert_eq!(duplicate_owner, 1, "latest duplicate must win when all pools are eligible");
-        let (_, active_duplicate_owner) = store
-            .get_latest_object_info_with_idx(
-                &bucket,
-                duplicate_object,
-                &ObjectOptions {
-                    skip_decommissioned: true,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("active duplicate owner should resolve");
-        assert_eq!(
-            active_duplicate_owner, 0,
-            "suspended duplicate must be excluded from active owner selection"
-        );
-        let (duplicate_result, duplicate_err) = store
-            .handle_heal_object(&bucket, duplicate_object, "", &HealOpts::default())
-            .await
-            .expect("duplicate owner heal should complete through the production path");
-        assert_eq!(duplicate_result.object, duplicate_object);
-        assert!(duplicate_err.is_none(), "active duplicate should be repaired: {duplicate_err:?}");
-        assert!(
-            duplicate_missing_disk.read_xl(&bucket, duplicate_object, false).await.is_ok(),
-            "production heal must repair the active duplicate owner rather than the suspended owner"
-        );
-        let (marker_info, marker_owner) = store
-            .get_latest_object_info_with_idx(
-                &bucket,
-                marker_object,
-                &ObjectOptions {
-                    skip_decommissioned: true,
-                    versioned: true,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("latest delete marker should resolve");
-        assert_eq!(marker_owner, 0);
-        assert!(marker_info.delete_marker, "latest version must preserve delete-marker semantics");
-
-        let (active_result, active_err) = store
-            .handle_heal_object(&bucket, active_object, "", &HealOpts::default())
-            .await
-            .expect("unscoped active-owner heal should complete");
-        assert_eq!(active_result.object, active_object);
-        assert!(active_err.is_none(), "active owner must be selected even with a suspended pool");
-        assert!(
-            missing_active_disk.read_xl(&bucket, active_object, false).await.is_ok(),
-            "active owner heal must write the missing disk metadata: result={active_result:?}, err={active_err:?}"
-        );
-        assert!(
-            store.pools[1]
-                .get_object_info(&bucket, active_object, &ObjectOptions::default())
-                .await
-                .is_err(),
-            "the suspended pool must not be written for an active-owner object"
-        );
-
-        let (suspended_result, suspended_err) = store
-            .handle_heal_object(&bucket, suspended_only_object, "", &HealOpts::default())
-            .await
-            .expect("unscoped suspended-only heal should return a terminal result");
-        assert!(suspended_result.object.is_empty());
-        assert!(matches!(suspended_err, Some(Error::FileNotFound)));
-        assert!(
-            store.pools[1]
-                .get_object_info(&bucket, suspended_only_object, &ObjectOptions::default())
-                .await
-                .is_ok(),
-            "suspended-only data must remain untouched when unscoped heal reports absent"
-        );
-
-        let (_, explicit_err) = store
-            .handle_heal_object(
-                &bucket,
-                suspended_only_object,
-                "",
-                &HealOpts {
-                    pool: Some(1),
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("explicit suspended-owner heal should return a mapped error");
-        assert!(matches!(explicit_err, Some(Error::SlowDown)));
-
-        let original_quorum_disks = store.pools[0].disk_set[0].disks.read().await.clone();
-        let surviving_quorum_disk = original_quorum_disks[3].clone();
-        *store.pools[0].disk_set[0].disks.write().await = vec![None, None, None, surviving_quorum_disk];
-        let (_, quorum_err) = store
-            .handle_heal_object(&bucket, quorum_object, "", &HealOpts::default())
-            .await
-            .expect("quorum boundary heal should return a mapped result");
-        *store.pools[0].disk_set[0].disks.write().await = original_quorum_disks;
-        assert!(
-            matches!(quorum_err, Some(Error::ErasureReadQuorum)),
-            "quorum-boundary heal must preserve quorum error, got {quorum_err:?}"
-        );
-        shutdown.cancel();
-    }
-
    #[tokio::test]
    async fn handle_heal_format_continues_after_a_pool_error() {
        let canonical_format = FormatV3::new(1, 3);
@@ -1640,6 +1640,60 @@ mod tests {
        assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0);
    }

+    #[tokio::test]
+    async fn test_process_query_request_reports_displaced_terminal_detail() {
+        let heal_manager = Arc::new(HealManager::new(
+            Arc::new(MockStorage),
+            Some(HealConfig {
+                queue_size: 1,
+                ..HealConfig::default()
+            }),
+        ));
+        let mut displaced = HealRequest::new(
+            HealType::Bucket {
+                bucket: "displaced-channel".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+        displaced.id = "displaced-channel-task".to_string();
+        let displaced_id = displaced.id.clone();
+        heal_manager
+            .submit_heal_request(displaced)
+            .await
+            .expect("initial channel task should queue");
+        heal_manager
+            .submit_heal_request(HealRequest::new(
+                HealType::Bucket {
+                    bucket: "successor-channel".to_string(),
+                },
+                HealOptions::default(),
+                HealPriority::High,
+            ))
+            .await
+            .expect("successor channel task should displace the initial task");
+
+        let processor = HealChannelProcessor::new(heal_manager);
+        let (tx, rx) = oneshot::channel();
+        processor
+            .process_query_request("displaced-channel".to_string(), displaced_id, None, tx)
+            .await
+            .expect("displaced query should process");
+        let response = rx
+            .await
+            .expect("query response should be returned")
+            .expect("displaced query should remain successful");
+        let payload: serde_json::Value = serde_json::from_slice(response.data.as_deref().expect("status payload should exist"))
+            .expect("status payload should be json");
+        assert_eq!(payload["summary"], "stopped");
+        assert!(
+            response
+                .error
+                .as_deref()
+                .is_some_and(|detail| detail.contains("reason=displaced"))
+        );
+    }
+
    #[tokio::test]
    async fn test_process_query_request_reports_running_for_queued_task() {
        let heal_manager = create_test_heal_manager();
@@ -40,6 +40,7 @@ use tracing::{debug, error, info, warn};
 use super::{DiskError, Endpoint, HealDiskExt as _, local_disk_map_read};

 const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
+const DISPLACED_HEAL_REASON: &str = "reason=displaced; retry_hint=submit_again";
 const LOG_COMPONENT_HEAL: &str = "heal";
 const LOG_SUBSYSTEM_DISK_SCANNER: &str = "disk_scanner";
 const LOG_SUBSYSTEM_MANAGER: &str = "manager";
@@ -120,26 +121,30 @@ struct MrfRepairNoticeTarget {
    version_id: Option<[u8; 16]>,
 }

-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone)]
 struct HealAdmissionDecision {
    result: HealAdmissionResult,
-    displaced_task_id: Option<String>,
+    displaced_request: Option<HealRequest>,
 }

 impl HealAdmissionDecision {
    const fn new(result: HealAdmissionResult) -> Self {
        Self {
            result,
-            displaced_task_id: None,
+            displaced_request: None,
        }
    }

-    fn accepted_with_displacement(displaced_task_id: String) -> Self {
+    fn accepted_with_displacement(displaced_request: HealRequest) -> Self {
        Self {
            result: HealAdmissionResult::Accepted,
-            displaced_task_id: Some(displaced_task_id),
+            displaced_request: Some(displaced_request),
        }
    }
+
+    fn displaced_task_id(&self) -> Option<&str> {
+        self.displaced_request.as_ref().map(|request| request.id.as_str())
+    }
 }

 fn lock_mrf_repair_notice_targets(
@@ -151,6 +156,55 @@ fn lock_mrf_repair_notice_targets(
    }
 }

+fn lock_displaced_terminals(
+    registry: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
+) -> StdMutexGuard<'_, HashMap<String, Arc<CompletedHealStatus>>> {
+    match registry.lock() {
+        Ok(guard) => guard,
+        Err(poisoned) => poisoned.into_inner(),
+    }
+}
+
+fn record_displaced_terminal(
+    registry: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
+    request: &HealRequest,
+) -> Arc<CompletedHealStatus> {
+    let terminal = Arc::new(CompletedHealStatus {
+        heal_type: request.heal_type.clone(),
+        status: HealTaskStatus::Failed {
+            error: format!("heal task displaced by a higher-priority request ({DISPLACED_HEAL_REASON})"),
+        },
+        result_items_truncated: false,
+        completed_at: SystemTime::now(),
+        seqed_items: Vec::new(),
+        next_seq: 0,
+        min_seq: 0,
+    });
+    let mut terminals = lock_displaced_terminals(registry);
+    prune_completed_heal_statuses(&mut terminals);
+    terminals.insert(request.id.clone(), Arc::clone(&terminal));
+    terminal
+}
+
+async fn remove_displaced_task_aliases(
+    aliases: &Arc<Mutex<HashMap<String, HealTaskAlias>>>,
+    terminals: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
+    task_id: &str,
+    terminal: &Arc<CompletedHealStatus>,
+) {
+    let mut aliases = aliases.lock().await;
+    let alias_ids = aliases
+        .iter()
+        .filter_map(|(alias_id, alias)| (alias.task_id == task_id).then_some(alias_id.clone()))
+        .collect::<Vec<_>>();
+    let mut displaced_terminals = lock_displaced_terminals(terminals);
+    prune_completed_heal_statuses(&mut displaced_terminals);
+    for alias_id in alias_ids {
+        displaced_terminals.insert(alias_id, Arc::clone(terminal));
+    }
+    aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
+}
+
 async fn remove_task_aliases_for_task(registry: &Arc<Mutex<HashMap<String, HealTaskAlias>>>, task_id: &str) {
    registry
        .lock()
@@ -618,6 +672,14 @@ pub struct HealManager {
    /// are shared so the lookup helper can hand a completed entry to a
    /// caller without cloning the retained result window.
    completed_heals: Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
+    /// Terminals for requests removed by priority displacement. An Accepted
+    /// task ID remains queryable for the same process lifetime and the normal
+    /// ten-minute status TTL; clients should treat `reason=displaced` as a
+    /// terminal result and submit a fresh request. This sidecar is synchronous
+    /// so admission can publish the terminal while the queue transition is
+    /// still under its lock, without awaiting another tokio lock. Queue state
+    /// is process-local, so this guarantee does not extend across restart.
+    displaced_terminals: Arc<StdMutex<HashMap<String, Arc<CompletedHealStatus>>>>,
    /// Client tokens merged into an existing task id.
    task_aliases: Arc<Mutex<HashMap<String, HealTaskAlias>>>,
    /// Heal tasks waiting for a retry backoff to expire.
@@ -659,6 +721,7 @@ struct HealQueueContext<'a> {
    heal_queue: &'a Arc<Mutex<PriorityHealQueue>>,
    active_heals: &'a Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
    completed_heals: &'a Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
+    displaced_terminals: &'a Arc<StdMutex<HashMap<String, Arc<CompletedHealStatus>>>>,
    task_aliases: &'a Arc<Mutex<HashMap<String, HealTaskAlias>>>,
    retrying_heals: &'a Arc<Mutex<HashMap<String, RetryingHeal>>>,
    mrf_repair_notice_targets: &'a Arc<StdMutex<HashMap<String, Vec<MrfRepairNoticeTarget>>>>,
@@ -874,7 +937,7 @@ impl HealManager {
                        result = "accepted_by_displacement",
                        "Heal queue request accepted by displacement"
                    });
-                    return HealAdmissionDecision::accepted_with_displacement(displaced.id);
+                    return HealAdmissionDecision::accepted_with_displacement(displaced);
                }

                demote_to_debug_when!(per_object_request, warn, target: "rustfs::heal::manager", {
@@ -1105,6 +1168,7 @@ impl HealManager {
            active_heals: Arc::new(Mutex::new(HashMap::new())),
            heal_queue: Arc::new(Mutex::new(PriorityHealQueue::new())),
            completed_heals: Arc::new(Mutex::new(HashMap::new())),
+            displaced_terminals: Arc::new(StdMutex::new(HashMap::new())),
            task_aliases: Arc::new(Mutex::new(HashMap::new())),
            retrying_heals: Arc::new(Mutex::new(HashMap::new())),
            mrf_repair_notice_targets: Arc::new(StdMutex::new(HashMap::new())),
@@ -1209,6 +1273,10 @@ impl HealManager {
        active_heals.clear();
        publish_active_heal_count(&active_heals);
        self.completed_heals.lock().await.clear();
+        // Do not let the synchronous guard live across the following async lock.
+        {
+            lock_displaced_terminals(&self.displaced_terminals).clear();
+        }
        self.task_aliases.lock().await.clear();
        self.retrying_heals.lock().await.clear();
        lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).clear();
@@ -1459,7 +1527,11 @@ impl HealManager {
            task_id = queued_id.to_owned();
        }
        let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
-        let displaced_task_id = admission_decision.displaced_task_id;
+        let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
+        let displaced_terminal = admission_decision
+            .displaced_request
+            .as_ref()
+            .map(|request| record_displaced_terminal(&self.displaced_terminals, request));
        if matches!(admission, HealAdmissionResult::Accepted | HealAdmissionResult::Merged)
            && let Some(target) = mrf_notice_target
        {
@@ -1473,8 +1545,12 @@ impl HealManager {
        drop(queue);
        drop(active_heals);

-        if let Some(displaced_task_id) = displaced_task_id {
-            self.remove_aliases_for_task(&displaced_task_id).await;
+        if let (Some(displaced_task_id), Some(displaced_terminal)) = (displaced_task_id, displaced_terminal) {
+            // The queue has already removed the displaced request, so the
+            // synchronous terminal sidecar was published before aliases and
+            // MRF ownership are cleaned up.
+            remove_displaced_task_aliases(&self.task_aliases, &self.displaced_terminals, &displaced_task_id, &displaced_terminal)
+                .await;
        }

        if should_notify {
@@ -1549,6 +1625,15 @@ impl HealManager {
            }
        }

+        if terminal_completed.is_none() {
+            let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals);
+            prune_completed_heal_statuses(&mut displaced_terminals);
+            terminal_completed = displaced_terminals
+                .get(canonical_task_id)
+                .filter(|terminal| matches_path(&terminal.heal_type))
+                .cloned();
+        }
+
        match terminal_completed {
            Some(completed) => TaskStateLookup::Completed(completed),
            None => TaskStateLookup::NotFound,
@@ -1669,9 +1754,19 @@ impl HealManager {

        let mut completed_heals = self.completed_heals.lock().await;
        prune_completed_heal_statuses(&mut completed_heals);
-        completed_heals
+        if completed_heals
            .values()
            .any(|completed| heal_type_matches_path(&completed.heal_type, heal_path))
+        {
+            return true;
+        }
+        drop(completed_heals);
+
+        let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals);
+        prune_completed_heal_statuses(&mut displaced_terminals);
+        displaced_terminals
+            .values()
+            .any(|terminal| heal_type_matches_path(&terminal.heal_type, heal_path))
    }

    /// Get task progress
@@ -21,6 +21,7 @@ impl HealManager {
        let heal_queue = self.heal_queue.clone();
        let active_heals = self.active_heals.clone();
        let task_aliases = self.task_aliases.clone();
+        let displaced_terminals = self.displaced_terminals.clone();
        let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone();
        let storage = self.storage.clone();
        let replacement_recovery_anchors = self.replacement_recovery_anchors.clone();
@@ -481,6 +482,10 @@ impl HealManager {
                            let admission = admission_decision.result;
                            let should_notify =
                                matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
+                            let displaced_terminal = admission_decision
+                                .displaced_request
+                                .as_ref()
+                                .map(|request| record_displaced_terminal(&displaced_terminals, request));
                            if matches!(admission, HealAdmissionResult::Accepted)
                                && let Some(anchor) = recovery_anchor
                            {
@@ -491,8 +496,16 @@ impl HealManager {
                            }
                            drop(queue);
                            drop(config);
-                            if let Some(displaced_task_id) = admission_decision.displaced_task_id {
-                                remove_task_aliases_for_task(&task_aliases, &displaced_task_id).await;
+                            if let (Some(displaced_task_id), Some(displaced_terminal)) =
+                                (admission_decision.displaced_task_id().map(ToOwned::to_owned), displaced_terminal)
+                            {
+                                remove_displaced_task_aliases(
+                                    &task_aliases,
+                                    &displaced_terminals,
+                                    &displaced_task_id,
+                                    &displaced_terminal,
+                                )
+                                .await;
                                lock_mrf_repair_notice_targets(&mrf_repair_notice_targets).remove(&displaced_task_id);
                            }
                            if matches!(admission, HealAdmissionResult::Accepted) {
@@ -21,6 +21,7 @@ impl HealManager {
        let heal_queue = self.heal_queue.clone();
        let active_heals = self.active_heals.clone();
        let completed_heals = self.completed_heals.clone();
+        let displaced_terminals = self.displaced_terminals.clone();
        let task_aliases = self.task_aliases.clone();
        let retrying_heals = self.retrying_heals.clone();
        let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone();
@@ -53,6 +54,7 @@ impl HealManager {
                            heal_queue: &heal_queue,
                            active_heals: &active_heals,
                            completed_heals: &completed_heals,
+                            displaced_terminals: &displaced_terminals,
                            task_aliases: &task_aliases,
                            retrying_heals: &retrying_heals,
                            mrf_repair_notice_targets: &mrf_repair_notice_targets,
@@ -71,6 +73,7 @@ impl HealManager {
                            heal_queue: &heal_queue,
                            active_heals: &active_heals,
                            completed_heals: &completed_heals,
+                            displaced_terminals: &displaced_terminals,
                            task_aliases: &task_aliases,
                            retrying_heals: &retrying_heals,
                            mrf_repair_notice_targets: &mrf_repair_notice_targets,
@@ -98,6 +101,7 @@ impl HealManager {
            heal_queue,
            active_heals,
            completed_heals,
+            displaced_terminals,
            task_aliases,
            retrying_heals,
            mrf_repair_notice_targets,
@@ -183,6 +187,7 @@ impl HealManager {
                let active_heals_clone = active_heals.clone();
                let heal_queue_clone = heal_queue.clone();
                let completed_heals_clone = completed_heals.clone();
+                let displaced_terminals_clone = displaced_terminals.clone();
                let task_aliases_clone = task_aliases.clone();
                let retrying_heals_clone = retrying_heals.clone();
                let mrf_repair_notice_targets_clone = mrf_repair_notice_targets.clone();
@@ -363,6 +368,7 @@ impl HealManager {
                        let retry_heal_queue = heal_queue_clone.clone();
                        let retrying_heals_for_spawn = retrying_heals_clone.clone();
                        let retry_task_aliases = task_aliases_clone.clone();
+                        let retry_displaced_terminals = displaced_terminals_clone.clone();
                        let retry_mrf_repair_notice_targets = mrf_repair_notice_targets_clone.clone();
                        let retry_completed_heals = completed_heals_clone.clone();
                        let retry_notify = notify_clone.clone();
@@ -430,6 +436,14 @@ impl HealManager {
                                let admission = admission_decision.result;
                                let should_notify = matches!(admission, HealAdmissionResult::Accepted)
                                    && retry_config.event_driven_scheduler_enable;
+                                // Publish the terminal synchronously while the
+                                // queue transition is protected. The subsequent
+                                // queue -> retrying handoff retains the lock order
+                                // used by operations_snapshot.
+                                let displaced_terminal = admission_decision
+                                    .displaced_request
+                                    .as_ref()
+                                    .map(|request| record_displaced_terminal(&retry_displaced_terminals, request));
                                match admission {
                                    HealAdmissionResult::Accepted => {
                                        // Transfer ownership while holding queue -> retrying,
@@ -437,10 +451,18 @@ impl HealManager {
                                        #[cfg(test)]
                                        pause_retry_ownership_transition(&retry_request_id, true).await;
                                        retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
-                                        let displaced_task_id = admission_decision.displaced_task_id;
+                                        let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
                                        drop(queue);
-                                        if let Some(displaced_task_id) = displaced_task_id {
-                                            remove_task_aliases_for_task(&retry_task_aliases, &displaced_task_id).await;
+                                        if let (Some(displaced_task_id), Some(displaced_terminal)) =
+                                            (displaced_task_id, displaced_terminal)
+                                        {
+                                            remove_displaced_task_aliases(
+                                                &retry_task_aliases,
+                                                &retry_displaced_terminals,
+                                                &displaced_task_id,
+                                                &displaced_terminal,
+                                            )
+                                            .await;
                                            remove_mrf_repair_notice_targets(
                                                &retry_mrf_repair_notice_targets,
                                                &displaced_task_id,
@@ -84,6 +84,7 @@ async fn process_manager_queue_once(manager: &HealManager) {
        heal_queue: &manager.heal_queue,
        active_heals: &manager.active_heals,
        completed_heals: &manager.completed_heals,
+        displaced_terminals: &manager.displaced_terminals,
        task_aliases: &manager.task_aliases,
        retrying_heals: &manager.retrying_heals,
        mrf_repair_notice_targets: &manager.mrf_repair_notice_targets,
@@ -2778,7 +2779,10 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() {
        HealAdmissionResult::Accepted
    );
    assert_eq!(manager.get_queue_length().await, 1);
-    assert!(matches!(manager.get_task_status(&low_id).await, Err(Error::TaskNotFound { .. })));
+    assert!(matches!(
+        manager.get_task_status(&low_id).await,
+        Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced")
+    ));
    assert_eq!(
        manager
            .get_task_status(&high_id)
@@ -2788,6 +2792,263 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() {
    );
 }

+#[tokio::test]
+async fn displaced_task_remains_queryable() {
+    let manager = HealManager::new(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 1,
+            ..HealConfig::default()
+        }),
+    );
+    let mut displaced = HealRequest::new(
+        HealType::Bucket {
+            bucket: "displaced-bucket".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    displaced.id = "displaced-task".to_string();
+    let displaced_id = displaced.id.clone();
+    manager
+        .submit_heal_request(displaced)
+        .await
+        .expect("displaced request should queue");
+
+    let successor = HealRequest::new(
+        HealType::Bucket {
+            bucket: "successor-bucket".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::High,
+    );
+    manager
+        .submit_heal_request(successor)
+        .await
+        .expect("successor should displace low work");
+
+    let report = manager
+        .get_task_report(&displaced_id)
+        .await
+        .expect("displaced report should remain queryable");
+    assert!(matches!(report.status, HealTaskStatus::Failed { ref error } if error.contains("reason=displaced")));
+}
+
+#[tokio::test]
+async fn displaced_archive_failure_keeps_queryable_terminal() {
+    let manager = HealManager::new(Arc::new(MockStorage), None);
+    let mut request = HealRequest::new(
+        HealType::Bucket {
+            bucket: "archive-failure".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    request.id = "archive-failure-task".to_string();
+    let request_id = request.id.clone();
+    // The synchronous sidecar is the authoritative fallback when the normal
+    // completed-task archive has no entry (the failure window that must not
+    // turn an Accepted ID into NotFound).
+    record_displaced_terminal(&manager.displaced_terminals, &request);
+    assert!(manager.completed_heals.lock().await.is_empty());
+    assert!(matches!(
+        manager.get_task_status(&request_id).await,
+        Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced")
+    ));
+}
+
+#[tokio::test]
+async fn scheduler_retry_displacement_keeps_evicted_task_queryable() {
+    let manager = Arc::new(HealManager::new(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 1,
+            event_driven_scheduler_enable: false,
+            ..HealConfig::default()
+        }),
+    ));
+    let mut retry_request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
+    retry_request.priority = HealPriority::High;
+    let retry_id = retry_request.id.clone();
+    manager
+        .submit_heal_request(retry_request)
+        .await
+        .expect("retry request should queue");
+
+    // Process exactly one queue cycle so the retry task is spawned without a
+    // background scheduler consuming the filler request before the retry wakes.
+    process_manager_queue_once(&manager).await;
+    tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            if manager.retrying_heals.lock().await.contains_key(&retry_id) {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("retry request should enter backoff");
+
+    let filler = HealRequest::new(
+        HealType::Bucket {
+            bucket: "retry-displaced-filler".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    let filler_id = filler.id.clone();
+    manager
+        .submit_heal_request(filler)
+        .await
+        .expect("filler request should occupy the queue");
+
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            if matches!(
+                manager.get_task_status(&filler_id).await,
+                Ok(HealTaskStatus::Failed { ref error }) if error.contains("reason=displaced")
+            ) {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("retry admission should displace the filler request");
+    assert_eq!(manager.get_queue_length().await, 1);
+    assert_eq!(
+        manager.get_task_status(&retry_id).await.expect("retry should be queued"),
+        HealTaskStatus::Pending
+    );
+}
+
+#[tokio::test]
+async fn concurrent_displacers_produce_one_terminal_generation() {
+    let manager = Arc::new(HealManager::new(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 1,
+            ..HealConfig::default()
+        }),
+    ));
+    let mut displaced = HealRequest::new(
+        HealType::Bucket {
+            bucket: "concurrent-displaced".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    displaced.id = "concurrent-displaced-task".to_string();
+    let displaced_id = displaced.id.clone();
+    manager
+        .submit_heal_request(displaced)
+        .await
+        .expect("initial request should queue");
+
+    let first = HealRequest::new(
+        HealType::Bucket {
+            bucket: "concurrent-successor-a".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::High,
+    );
+    let second = HealRequest::new(
+        HealType::Bucket {
+            bucket: "concurrent-successor-b".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::High,
+    );
+    let (first_result, second_result) = tokio::join!(manager.submit_heal_request(first), manager.submit_heal_request(second));
+    let accepted = [&first_result, &second_result]
+        .into_iter()
+        .filter(|result| matches!(result, Ok(HealAdmissionResult::Accepted)))
+        .count();
+    assert_eq!(accepted, 1, "exactly one concurrent displacer should win the full queue");
+    assert!(
+        first_result.is_ok() && second_result.is_ok(),
+        "the losing request should receive a typed Full result"
+    );
+    let terminals = lock_displaced_terminals(&manager.displaced_terminals);
+    assert_eq!(terminals.len(), 1);
+    assert!(terminals.contains_key(&displaced_id));
+}
+
+#[tokio::test]
+async fn successor_chain_is_bounded_and_authorized() {
+    let manager = HealManager::new(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 1,
+            ..HealConfig::default()
+        }),
+    );
+    let mut original = HealRequest::new(
+        HealType::Bucket {
+            bucket: "authorized-original".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    original.id = "authorized-original-task".to_string();
+    let original_id = original.id.clone();
+    manager.submit_heal_request(original).await.expect("original should queue");
+    let mut duplicate = HealRequest::new(
+        HealType::Bucket {
+            bucket: "authorized-original".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    duplicate.id = "authorized-duplicate-task".to_string();
+    let duplicate_id = duplicate.id.clone();
+    manager
+        .submit_heal_request(duplicate)
+        .await
+        .expect("same-target duplicate should merge");
+    let successor = HealRequest::new(
+        HealType::Bucket {
+            bucket: "authorized-successor".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::High,
+    );
+    let successor_id = successor.id.clone();
+    manager.submit_heal_request(successor).await.expect("successor should queue");
+    assert!(manager.task_aliases.lock().await.is_empty());
+    assert!(matches!(manager.get_task_status(&original_id).await, Ok(HealTaskStatus::Failed { .. })));
+    assert!(matches!(manager.get_task_status(&duplicate_id).await, Ok(HealTaskStatus::Failed { .. })));
+    assert_eq!(
+        manager
+            .get_task_status(&successor_id)
+            .await
+            .expect("successor should remain queued"),
+        HealTaskStatus::Pending
+    );
+}
+
+#[tokio::test]
+async fn displaced_terminal_expires_after_bounded_ttl() {
+    let manager = HealManager::new(Arc::new(MockStorage), None);
+    let mut request = HealRequest::new(
+        HealType::Bucket {
+            bucket: "expires".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    request.id = "expires-task".to_string();
+    let request_id = request.id.clone();
+    record_displaced_terminal(&manager.displaced_terminals, &request);
+    {
+        let mut terminals = lock_displaced_terminals(&manager.displaced_terminals);
+        let entry =
+            Arc::get_mut(terminals.get_mut(&request_id).expect("terminal should be retained")).expect("test owns terminal entry");
+        entry.completed_at = SystemTime::now() - KEEP_HEAL_TASK_STATUS_DURATION - Duration::from_secs(1);
+    }
+    assert!(matches!(manager.get_task_status(&request_id).await, Err(Error::TaskNotFound { .. })));
+}
+
 #[tokio::test]
 async fn test_displacing_registered_mrf_task_drops_notice_ownership() {
    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
@@ -722,6 +722,10 @@ pub struct DeleteVersionsResponse {
    pub errors: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
    #[prost(message, optional, tag = "3")]
    pub error: ::core::option::Option<Error>,
+    /// Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries
+    /// when present and fall back to strings for peers that predate this field. Code zero means success.
+    #[prost(message, repeated, tag = "4")]
+    pub item_errors: ::prost::alloc::vec::Vec<Error>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadMultipleRequest {
@@ -493,6 +493,9 @@ message DeleteVersionsResponse {
  bool success = 1;
  repeated string errors = 2;
  optional Error error = 3;
+  // Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries
+  // when present and fall back to strings for peers that predate this field. Code zero means success.
+  repeated Error item_errors = 4;
 }

 message ReadMultipleRequest {
@@ -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)
+}
@@ -146,6 +146,29 @@ fn encode_file_info_msgpack(value: &FileInfo) -> std::result::Result<Vec<u8>, Di
    encode_msgpack_with_capacity(value, "FileInfo", FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT)
 }

+fn encode_delete_versions_errors(disk_errors: Vec<Option<DiskError>>) -> (Vec<String>, Vec<Error>) {
+    let mut errors = Vec::with_capacity(disk_errors.len());
+    let mut item_errors = Vec::with_capacity(disk_errors.len());
+    for error in disk_errors {
+        match error {
+            Some(error) => {
+                let code = match &error {
+                    DiskError::Io(source) if source.kind() == std::io::ErrorKind::NotFound => DiskError::FileNotFound.to_u32(),
+                    _ => error.to_u32(),
+                };
+                let error_info = error.to_string();
+                errors.push(error_info.clone());
+                item_errors.push(Error { code, error_info });
+            }
+            None => {
+                errors.push(String::new());
+                item_errors.push(Error::default());
+            }
+        }
+    }
+    (errors, item_errors)
+}
+
 fn encode_msgpack_named<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, DiskError> {
    let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map();
    value
@@ -552,6 +575,7 @@ impl NodeService {
                            success: false,
                            errors: Vec::new(),
                            error: Some(DiskError::other(format!("decode FileInfoVersions failed: {err}")).into()),
+                            item_errors: Vec::new(),
                        }));
                    }
                };
@@ -563,30 +587,26 @@ impl NodeService {
                        success: false,
                        errors: Vec::new(),
                        error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
+                        item_errors: Vec::new(),
                    }));
                }
            };

-            let errors = disk
-                .delete_versions(&request.volume, versions, opts)
-                .await
-                .into_iter()
-                .map(|error| match error {
-                    Some(e) => e.to_string(),
-                    None => "".to_string(),
-                })
-                .collect();
+            let (errors, item_errors) =
+                encode_delete_versions_errors(disk.delete_versions(&request.volume, versions, opts).await);

            Ok(Response::new(DeleteVersionsResponse {
                success: true,
                errors,
                error: None,
+                item_errors,
            }))
        } else {
            Ok(Response::new(DeleteVersionsResponse {
                success: false,
                errors: Vec::new(),
                error: Some(DiskError::other("cannot find disk".to_string()).into()),
+                item_errors: Vec::new(),
            }))
        }
    }
@@ -1612,8 +1632,8 @@ impl NodeService {
 mod tests {
    use super::{
        compat_response_json, decode_msgpack_or_json, decode_rename_data_request_file_info,
-        encode_batch_read_version_response_payloads, encode_file_info_msgpack, encode_msgpack, encode_msgpack_named,
-        encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
+        encode_batch_read_version_response_payloads, encode_delete_versions_errors, encode_file_info_msgpack, encode_msgpack,
+        encode_msgpack_named, encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
    };
    use crate::storage::rpc::node_service::make_server;
    use crate::storage::storage_api::ReadMultipleResp;
@@ -1632,6 +1652,18 @@ mod tests {
        count: u32,
    }

+    #[test]
+    fn delete_versions_response_dual_writes_typed_item_errors() {
+        let raw_not_found = super::DiskError::Io(std::io::Error::from(std::io::ErrorKind::NotFound));
+        let (errors, item_errors) = encode_delete_versions_errors(vec![Some(raw_not_found), None]);
+
+        assert!(errors[0].starts_with("io error "));
+        assert!(errors[1].is_empty());
+        assert_eq!(item_errors[0].code, super::DiskError::FileNotFound.to_u32());
+        assert_eq!(item_errors[0].error_info, errors[0]);
+        assert_eq!(item_errors[1].code, 0);
+    }
+
    #[tokio::test]
    #[serial]
    async fn handle_read_version_records_attribution_for_missing_disk() {
@@ -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) {}
Author	SHA1	Message	Date
overtrue	e13049a846	feat(connect): emit durable heartbeats	2026-08-22 18:22:33 +08:00
Zhengchao An	1a3be70d98	fix(ecstore): preserve remote delete error types (#6371 )	2026-08-22 17:07:02 +08:00
cxymds	8679570c2a	fix(heal): retain displaced task status (#6370 )	2026-08-22 08:23:40 +00:00