feat(connect): persist sanitized inventory snapshot (#6537)

* feat(connect): persist sanitized inventory snapshot

* fix(connect): harden inventory persistence boundary

* fix(connect): harden inventory path anchors

* fix(connect): fail closed outside Linux

* fix(connect): gate inventory persistence to Linux

* fix(connect): keep runtime failure codes stable

* fix(connect): satisfy cross-platform lint

* fix(connect): preserve inventory persistence invariants

* fix(connect): preserve newer local inventory

* fix(connect): retain legacy inventory capture age

* chore(connect): document unsafe boundaries

* test(connect): secure inventory state fixtures

* fix(connect): preserve heartbeat runtime status

rustfs/src/connect/config.rs

@@ -14,6 +14,7 @@
 
 use std::env;
 use std::ffi::OsString;
+#[cfg(target_os = "linux")]
 use std::fs;
 use std::path::PathBuf;
 use std::time::Duration;
@@ -63,16 +64,37 @@ impl HeartbeatConfig {
         credential_store: CredentialStore,
         state_path: impl Into<PathBuf>,
     ) -> Self {
+        let state_path = state_path.into();
         Self {
             endpoint: endpoint.into(),
             root_ca_pem: root_ca_pem.into(),
             identity_store,
             credential_store,
-            state_path: state_path.into(),
+            state_path,
             schedule: HeartbeatSchedule::default(),
         }
     }
 
+    #[cfg(any(target_os = "linux", test))]
+    pub(crate) fn state_only(state_root: PathBuf) -> Self {
+        Self {
+            endpoint: String::new(),
+            root_ca_pem: Vec::new(),
+            identity_store: IdentityStore::new(state_root.join("identity")),
+            credential_store: CredentialStore::new(state_root.join("credential")),
+            state_path: state_root.join("heartbeat/state.json"),
+            schedule: HeartbeatSchedule::default(),
+        }
+    }
+
+    pub(crate) fn transport_enabled(&self) -> bool {
+        !self.endpoint.is_empty()
+    }
+
+    pub(crate) fn state_root(&self) -> Option<&std::path::Path> {
+        self.state_path.parent().and_then(std::path::Path::parent)
+    }
+
     pub fn from_env() -> Result<Option<Self>, HeartbeatConfigError> {
         Self::from_env_values(
             env::var_os(ENV_CONNECT_ENDPOINT),
@@ -90,19 +112,33 @@ impl HeartbeatConfig {
         if !configured {
             return Ok(None);
         }
-        let (Some(endpoint), Some(root_ca_file), Some(state_dir)) = (endpoint, root_ca_file, state_dir) else {
+        let Some(state_dir) = 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() {
+        if state_dir.as_os_str().is_empty() || endpoint.is_some() != root_ca_file.is_some() {
             return Err(HeartbeatConfigError::Partial);
         }
+        #[cfg(not(target_os = "linux"))]
+        return Err(HeartbeatConfigError::PlatformSecurity);
+        #[cfg(target_os = "linux")]
+        let (Some(endpoint), Some(root_ca_file)) = (endpoint, root_ca_file) else {
+            return Ok(Some(Self::state_only(state_dir)));
+        };
+        #[cfg(target_os = "linux")]
+        let endpoint = endpoint.into_string().map_err(|_| HeartbeatConfigError::EndpointEncoding)?;
+        #[cfg(target_os = "linux")]
+        let root_ca_file = PathBuf::from(root_ca_file);
+        #[cfg(target_os = "linux")]
+        if endpoint.is_empty() || root_ca_file.as_os_str().is_empty() {
+            return Err(HeartbeatConfigError::Partial);
+        }
+        #[cfg(target_os = "linux")]
         let root_ca_pem = fs::read(&root_ca_file).map_err(|source| HeartbeatConfigError::RootCertificate {
             path: root_ca_file,
             source,
         })?;
+        #[cfg(target_os = "linux")]
         Ok(Some(Self::new(
             endpoint,
             root_ca_pem,
@@ -116,17 +152,19 @@ impl HeartbeatConfig {
 #[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"
+        "Connect requires RUSTFS_CONNECT_STATE_DIR and either both or neither of RUSTFS_CONNECT_ENDPOINT and RUSTFS_CONNECT_ROOT_CA_FILE"
     )]
     Partial,
     #[error("RUSTFS_CONNECT_ENDPOINT is not valid UTF-8")]
     EndpointEncoding,
-    #[error("failed to read the Connect root CA at {path}: {source}")]
+    #[error("Connect root CA could not be read")]
     RootCertificate {
         path: PathBuf,
         #[source]
         source: std::io::Error,
     },
+    #[error("Connect inventory persistence requires Linux filesystem security guarantees")]
+    PlatformSecurity,
 }
 
 #[cfg(test)]
@@ -149,9 +187,34 @@ mod tests {
             HeartbeatConfig::from_env_values(Some(OsString::from("https://connect.example/agent/")), None, None),
             Err(HeartbeatConfigError::Partial)
         ));
+        assert!(matches!(
+            HeartbeatConfig::from_env_values(
+                Some(OsString::from("https://connect.example/agent/")),
+                Some(OsString::from("root.pem")),
+                None,
+            ),
+            Err(HeartbeatConfigError::Partial)
+        ));
+        assert!(matches!(
+            HeartbeatConfig::from_env_values(None, Some(OsString::from("root.pem")), Some(OsString::from("state"))),
+            Err(HeartbeatConfigError::Partial)
+        ));
     }
 
     #[test]
+    #[cfg(target_os = "linux")]
+    fn state_directory_alone_enables_local_inventory_without_transport() {
+        let state = tempfile::tempdir().expect("tempdir").keep();
+        let config = HeartbeatConfig::from_env_values(None, None, Some(state.clone().into_os_string()))
+            .expect("state-only config")
+            .expect("enabled config");
+
+        assert_eq!(config.state_root(), Some(state.as_path()));
+        assert!(!config.transport_enabled());
+    }
+
+    #[test]
+    #[cfg(target_os = "linux")]
     fn complete_environment_builds_the_durable_paths() {
         let temp = tempfile::tempdir().expect("tempdir");
         let root = temp.path().join("root.pem");
@@ -168,6 +231,24 @@ mod tests {
         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_eq!(config.state_root(), Some(state.as_path()));
         assert!(!state.exists(), "parsing configuration must not create state");
     }
+
+    #[test]
+    #[cfg(not(target_os = "linux"))]
+    fn configured_inventory_fails_without_linux_filesystem_guarantees() {
+        assert!(matches!(
+            HeartbeatConfig::from_env_values(None, None, Some(OsString::from("state"))),
+            Err(HeartbeatConfigError::PlatformSecurity)
+        ));
+        assert!(matches!(
+            HeartbeatConfig::from_env_values(
+                Some(OsString::from("https://connect.example/agent/")),
+                Some(OsString::from("missing-root.pem")),
+                Some(OsString::from("state")),
+            ),
+            Err(HeartbeatConfigError::PlatformSecurity)
+        ));
+    }
 }

rustfs/src/connect/runtime.rs

@@ -32,7 +32,6 @@ pub struct HeartbeatRuntime {
     shutdown: CancellationToken,
     status: watch::Receiver<HeartbeatStatus>,
     task: Option<JoinHandle<()>>,
-    inventory: Option<InventoryRuntime>,
 }
 
 impl HeartbeatRuntime {
@@ -40,22 +39,11 @@ impl HeartbeatRuntime {
         self.status.clone()
     }
 
-    pub(crate) fn with_inventory(mut self, inventory: Option<InventoryRuntime>) -> Self {
-        self.inventory = inventory;
-        self
-    }
-
     pub async fn shutdown(mut self) {
         self.shutdown.cancel();
-        if let Some(inventory) = self.inventory.as_ref() {
-            inventory.shutdown.cancel();
-        }
         if let Some(task) = self.task.take() {
             let _ = task.await;
         }
-        if let Some(inventory) = self.inventory.take() {
-            inventory.shutdown().await;
-        }
     }
 }
 
@@ -90,6 +78,20 @@ impl Drop for InventoryRuntime {
     }
 }
 
+pub(crate) async fn shutdown_connect_runtimes(heartbeat: Option<HeartbeatRuntime>, inventory: Option<InventoryRuntime>) {
+    let heartbeat = async move {
+        if let Some(runtime) = heartbeat {
+            runtime.shutdown().await;
+        }
+    };
+    let inventory = async move {
+        if let Some(runtime) = inventory {
+            runtime.shutdown().await;
+        }
+    };
+    tokio::join!(heartbeat, inventory);
+}
+
 pub fn spawn_heartbeat_runtime<F>(
     config: Option<HeartbeatConfig>,
     parent_shutdown: &CancellationToken,
@@ -101,6 +103,9 @@ where
     let Some(config) = config else {
         return Ok(None);
     };
+    if !config.transport_enabled() {
+        return Ok(None);
+    }
     let sender = HeartbeatSender::new(config.clone())?;
     let store = HeartbeatStateStore::new(config.state_path.clone());
     let lock = store.try_runtime_lock()?;
@@ -163,7 +168,6 @@ where
         shutdown,
         status: status_rx,
         task: Some(task),
-        inventory: None,
     }))
 }
 
@@ -184,9 +188,14 @@ where
         return Err(InventoryError::Schedule);
     }
     let retry_schedule = config.schedule;
-    let store = InventoryStateStore::from_heartbeat_path(&config.state_path)?;
+    let state_root = config.state_root().ok_or(InventoryError::StatePath)?;
+    let store = InventoryStateStore::from_state_root(state_root)?;
     let lock = store.try_runtime_lock()?;
-    let sender = InventorySender::new(config)?;
+    let sender = if config.transport_enabled() {
+        Some(InventorySender::new(config)?)
+    } else {
+        None
+    };
     let shutdown = parent_shutdown.child_token();
     let task_shutdown = shutdown.clone();
     let (status_tx, status_rx) = watch::channel(InventoryStatus::Starting);
@@ -197,8 +206,19 @@ where
             if task_shutdown.is_cancelled() {
                 break;
             }
-            let pending = match store.pending().await {
+            let pending = match if sender.is_some() { store.pending().await } else { Ok(None) } {
-                Ok(Some(pending)) => pending,
+                Ok(Some((pending, captured_at))) => {
+                    if let Err(error) = store
+                        .ensure_latest(pending.snapshot().clone(), captured_at, task_shutdown.clone())
+                        .await
+                    {
+                        if matches!(error, InventoryError::Cancelled) && task_shutdown.is_cancelled() {
+                            break;
+                        }
+                        return failed_inventory(&status_tx, error);
+                    }
+                    pending
+                }
                 Ok(None) => {
                     let snapshot = match cancellable(&task_shutdown, sample()).await {
                         Some(Ok(snapshot)) => snapshot,
@@ -218,6 +238,27 @@ where
                         Ok(content_hash) => content_hash,
                         Err(error) => return failed_inventory(&status_tx, error),
                     };
+                    let captured_at = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
+                    if let Err(error) = store
+                        .publish_latest(snapshot.clone(), captured_at, task_shutdown.clone())
+                        .await
+                    {
+                        if matches!(error, InventoryError::Cancelled) && task_shutdown.is_cancelled() {
+                            break;
+                        }
+                        return failed_inventory(&status_tx, error);
+                    }
+                    if task_shutdown.is_cancelled() {
+                        break;
+                    }
+                    if sender.is_none() {
+                        backoff = retry_schedule.initial_backoff;
+                        let _ = status_tx.send(InventoryStatus::Unchanged { content_hash });
+                        if sleep_or_cancel(&task_shutdown, schedule.cadence.saturating_add(jitter(schedule.jitter))).await {
+                            break;
+                        }
+                        continue;
+                    }
                     match store.prepare(snapshot).await {
                         Ok(Some(pending)) => pending,
                         Ok(None) => {
@@ -233,7 +274,15 @@ where
                 }
                 Err(error) => return failed_inventory(&status_tx, error),
             };
-            let delivery = match cancellable(&task_shutdown, sender.send(&pending)).await {
+            let delivery = match cancellable(
+                &task_shutdown,
+                sender
+                    .as_ref()
+                    .expect("sender exists when delivery state is prepared")
+                    .send(&pending),
+            )
+            .await
+            {
                 Some(Ok(delivery)) => delivery,
                 Some(Err(error)) => return failed_inventory(&status_tx, error),
                 None => break,
@@ -261,14 +310,13 @@ where
                     let _ = status_tx.send(InventoryStatus::BackingOff { delay });
                     delay
                 }
-                InventoryDelivery::AuthenticationStopped { status, reason } => {
+                InventoryDelivery::AuthenticationStopped { status } => {
-                    let _ = status_tx.send(InventoryStatus::AuthenticationStopped { status, reason });
+                    let _ = status_tx.send(InventoryStatus::AuthenticationStopped { status, reason: None });
                     return;
                 }
-                InventoryDelivery::Rejected { status, reason } => {
+                InventoryDelivery::Rejected { status } => {
-                    let suffix = reason.map_or_else(String::new, |reason| format!("; reason={reason}"));
                     let _ = status_tx.send(InventoryStatus::Failed {
-                        reason: format!("Connect rejected inventory with HTTP {status}{suffix}"),
+                        reason: format!("connect_inventory_rejected_http_{status}"),
                     });
                     return;
                 }
@@ -292,6 +340,46 @@ fn failed(status: &watch::Sender<HeartbeatStatus>, error: HeartbeatError) {
     });
 }
 
+pub(crate) fn heartbeat_failure_reason(error: &HeartbeatError) -> &'static str {
+    use super::registration::CredentialValidationError;
+
+    match error {
+        HeartbeatError::Endpoint => "connect_heartbeat_endpoint",
+        HeartbeatError::RootCertificate => "connect_heartbeat_root_certificate",
+        HeartbeatError::Schedule => "connect_heartbeat_schedule",
+        HeartbeatError::NotRegistered => "connect_heartbeat_not_registered",
+        HeartbeatError::IdentityMissing => "connect_heartbeat_identity_missing",
+        HeartbeatError::IdentityCertificate => "connect_heartbeat_identity_certificate",
+        HeartbeatError::CredentialName => "connect_heartbeat_credential_name",
+        HeartbeatError::CredentialExpired => "connect_heartbeat_credential_expired",
+        HeartbeatError::NodeSummary => "connect_heartbeat_node_summary",
+        HeartbeatError::SequenceExhausted => "connect_heartbeat_sequence_exhausted",
+        HeartbeatError::AlreadyRunning => "connect_heartbeat_already_running",
+        HeartbeatError::StateConflict => "connect_heartbeat_state_conflict",
+        HeartbeatError::StateIo { .. } => "connect_heartbeat_state_io",
+        HeartbeatError::StateInvalid { .. } => "connect_heartbeat_state_invalid",
+        HeartbeatError::StateCorrupt { .. } => "connect_heartbeat_state_corrupt",
+        #[cfg(unix)]
+        HeartbeatError::StatePermissions { .. } => "connect_heartbeat_state_permissions",
+        HeartbeatError::ResponseTooLarge => "connect_heartbeat_response_too_large",
+        HeartbeatError::Response => "connect_heartbeat_response",
+        HeartbeatError::Url(_) => "connect_heartbeat_url",
+        HeartbeatError::Transport(_) => "connect_heartbeat_transport",
+        HeartbeatError::Identity(_) => "connect_heartbeat_identity",
+        HeartbeatError::IdentityStore(_) => "connect_heartbeat_identity_store",
+        HeartbeatError::CredentialStore(_) => "connect_heartbeat_credential_store",
+        HeartbeatError::CredentialValidation(error) => match error {
+            CredentialValidationError::Certificate => "connect_heartbeat_credential_certificate",
+            CredentialValidationError::Chain => "connect_heartbeat_credential_chain",
+            CredentialValidationError::Identity => "connect_heartbeat_credential_identity",
+            CredentialValidationError::Key => "connect_heartbeat_credential_key",
+            CredentialValidationError::Validity => "connect_heartbeat_credential_validity",
+            CredentialValidationError::CertificateRequest => "connect_heartbeat_credential_request",
+            CredentialValidationError::RotationTranscript => "connect_heartbeat_credential_rotation_transcript",
+        },
+    }
+}
+
 fn failed_inventory(status: &watch::Sender<InventoryStatus>, error: InventoryError) {
     let _ = status.send(InventoryStatus::Failed {
         reason: error.to_string(),
@@ -327,7 +415,51 @@ mod tests {
     use super::*;
 
     #[tokio::test]
-    async fn heartbeat_shutdown_cancels_inventory_before_waiting_for_heartbeat() {
+    async fn state_only_configuration_does_not_start_heartbeat() {
+        let temp = tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory");
+        let config = HeartbeatConfig::state_only(temp.path().to_path_buf());
+        let shutdown = CancellationToken::new();
+
+        assert!(
+            spawn_heartbeat_runtime(Some(config), &shutdown, || { CoarseNodeSummary::new(1, 0, 0).expect("summary") })
+                .expect("disabled transport")
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn runtimes_expose_only_stable_machine_reasons() {
+        let error = HeartbeatError::StateIo {
+            path: std::path::PathBuf::from("/private/connect/state.json"),
+            source: std::io::Error::other("transport.internal"),
+        };
+        assert_eq!(heartbeat_failure_reason(&error), "connect_heartbeat_state_io");
+        assert_eq!(
+            heartbeat_failure_reason(&HeartbeatError::StateCorrupt {
+                path: std::path::PathBuf::from("/private/connect/state.json"),
+            }),
+            "connect_heartbeat_state_corrupt"
+        );
+        assert_eq!(
+            heartbeat_failure_reason(&HeartbeatError::CredentialExpired),
+            "connect_heartbeat_credential_expired"
+        );
+        assert_eq!(
+            heartbeat_failure_reason(&HeartbeatError::CredentialValidation(
+                super::super::registration::CredentialValidationError::Identity
+            )),
+            "connect_heartbeat_credential_identity"
+        );
+        assert_eq!(
+            heartbeat_failure_reason(&HeartbeatError::CredentialValidation(
+                super::super::registration::CredentialValidationError::Key
+            )),
+            "connect_heartbeat_credential_key"
+        );
+    }
+
+    #[tokio::test]
+    async fn unified_shutdown_cancels_inventory_before_waiting_for_heartbeat() {
         let heartbeat_shutdown = CancellationToken::new();
         let inventory_shutdown = CancellationToken::new();
         let task_inventory_shutdown = inventory_shutdown.clone();
@@ -342,18 +474,18 @@ mod tests {
             task_inventory_shutdown.cancelled().await;
             let _ = inventory_stopped.send(());
         });
-        let runtime = HeartbeatRuntime {
+        let heartbeat = HeartbeatRuntime {
             shutdown: heartbeat_shutdown,
             status: heartbeat_status,
             task: Some(heartbeat_task),
-            inventory: Some(InventoryRuntime {
+        };
-                shutdown: inventory_shutdown,
+        let inventory = InventoryRuntime {
-                status: inventory_status,
+            shutdown: inventory_shutdown,
-                task: Some(inventory_task),
+            status: inventory_status,
-            }),
+            task: Some(inventory_task),
         };
 
-        let shutdown = tokio::spawn(runtime.shutdown());
+        let shutdown = tokio::spawn(shutdown_connect_runtimes(Some(heartbeat), Some(inventory)));
         tokio::time::timeout(Duration::from_millis(250), stopped)
             .await
             .expect("inventory cancellation must not wait for heartbeat")

rustfs/src/startup_lifecycle.rs

@@ -14,6 +14,7 @@
 
 use crate::storage_api::startup::lifecycle::ECStore;
 use crate::{
+    connect::runtime::shutdown_connect_runtimes,
     server::{ServiceStateManager, ShutdownHandle, start_persisted_event_notifier_reconciler, wait_for_shutdown},
     startup_iam::{IamBootstrapDisposition, publish_ready_for_iam_bootstrap},
     startup_runtime_sources,
@@ -129,6 +130,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
     let StartupServiceRuntime {
         optional_runtimes,
         heartbeat,
+        inventory,
         iam_bootstrap,
         enable_scanner,
     } = service_runtime;
@@ -163,9 +165,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
         shutdown_token,
     )
     .await;
-    if let Some(heartbeat) = heartbeat {
+    shutdown_connect_runtimes(heartbeat, inventory).await;
-        heartbeat.shutdown().await;
-    }
     if let Err(err) = event_notifier_reconciler.await {
         tracing::warn!(
             target: "rustfs::main::run",

rustfs/src/startup_services.rs

@@ -17,8 +17,9 @@ use crate::storage_api::startup::services::{ECStore, EndpointServerPools, Server
 use crate::{
     config::Config,
     connect::{
-        CoarseNodeSummary, HeartbeatConfig, HeartbeatRuntime, InventoryError, InventoryFlag, InventoryRuntime, InventorySchedule,
+        CoarseNodeSummary, HeartbeatConfig, HeartbeatError, HeartbeatRuntime, InventoryError, InventoryFlag, InventoryRuntime,
-        InventorySnapshot, spawn_heartbeat_runtime, spawn_inventory_runtime,
+        InventorySchedule, InventorySnapshot, runtime::heartbeat_failure_reason, spawn_heartbeat_runtime,
+        spawn_inventory_runtime,
     },
     init::{init_buffer_profile_system, init_kms_system},
     server::ServiceStateManager,
@@ -40,6 +41,7 @@ use tokio_util::sync::CancellationToken;
 pub(crate) struct StartupServiceRuntime {
     pub(crate) optional_runtimes: OptionalRuntimeServices,
     pub(crate) heartbeat: Option<HeartbeatRuntime>,
+    pub(crate) inventory: Option<InventoryRuntime>,
     pub(crate) iam_bootstrap: IamBootstrapDisposition,
     pub(crate) enable_scanner: bool,
 }
@@ -104,11 +106,11 @@ pub(crate) async fn init_startup_runtime_services(
     init_observability_runtime(store.clone(), ctx.clone()).await;
     let heartbeat = start_heartbeat_runtime(heartbeat_config.clone(), heartbeat_nodes, &ctx)?;
     let inventory = start_inventory_runtime(heartbeat_config, heartbeat_nodes, inventory_drives, store, &ctx)?;
-    let heartbeat = heartbeat.map(|heartbeat| heartbeat.with_inventory(inventory));
 
     Ok(StartupServiceRuntime {
         optional_runtimes,
         heartbeat,
+        inventory,
         iam_bootstrap,
         enable_scanner,
     })
@@ -122,11 +124,18 @@ fn start_heartbeat_runtime(
     let Some(config) = config else {
         return Ok(None);
     };
+    if !config.transport_enabled() {
+        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)
+    spawn_heartbeat_runtime(Some(config), shutdown, move || summary).map_err(startup_heartbeat_error)
+}
+
+fn startup_heartbeat_error(error: HeartbeatError) -> std::io::Error {
+    std::io::Error::other(heartbeat_failure_reason(&error))
 }
 
 fn start_inventory_runtime(
@@ -318,6 +327,16 @@ fn aggregate_inventory_capacity(
 mod tests {
     use super::*;
 
+    #[test]
+    fn heartbeat_startup_errors_expose_only_stable_codes() {
+        let error = startup_heartbeat_error(HeartbeatError::StateIo {
+            path: std::path::PathBuf::from("/private/connect/canary/state.json"),
+            source: std::io::Error::other("private-source-canary"),
+        });
+
+        assert_eq!(error.to_string(), "connect_heartbeat_state_io");
+    }
+
     fn disk(state: &str, runtime_state: Option<&str>, disk_index: i32) -> rustfs_madmin::Disk {
         rustfs_madmin::Disk {
             state: state.to_owned(),

rustfs/tests/connect_inventory.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![cfg(target_os = "linux")]
+
 use std::collections::VecDeque;
 use std::fs;
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -46,6 +48,10 @@ const CLUSTER_UID: &str = "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81";
 const DEVICE_UID: &str = "0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92";
 const SNAPSHOT_UID: &str = "0198f4b0-4d00-7f40-9051-5b6c7d8e9fa3";
 
+fn safe_tempdir() -> tempfile::TempDir {
+    tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory")
+}
+
 struct TestPki {
     root_params: CertificateParams,
     root_key: KeyPair,
@@ -244,6 +250,7 @@ fn config(temp: &tempfile::TempDir, pki: &TestPki, server: &TestServer) -> Heart
     if let Err(error) = fs::create_dir(temp.path().join("private-config-secret")) {
         assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists, "Connect state root");
     }
+    private_directory_mode(&temp.path().join("private-config-secret"));
     HeartbeatConfig {
         endpoint: server.endpoint.clone(),
         root_ca_pem: pki.root_pem.as_bytes().to_vec(),
@@ -394,6 +401,38 @@ fn connect_inventory_bounds_fail_instead_of_truncating_or_inventing_values() {
     ));
 }
 
+#[cfg(target_os = "linux")]
+#[tokio::test]
+async fn connect_inventory_state_only_persists_without_constructing_transport() {
+    let temp = safe_tempdir();
+    let state = temp.path().join("state");
+    fs::create_dir(&state).expect("state root");
+    private_directory_mode(&state);
+    let config = HeartbeatConfig::new(
+        "",
+        Vec::new(),
+        IdentityStore::new(state.join("identity")),
+        CredentialStore::new(state.join("credential")),
+        state.join("heartbeat/state.json"),
+    );
+    let shutdown = CancellationToken::new();
+    let runtime = spawn_inventory_runtime(Some(config), schedule(), &shutdown, || std::future::ready(Ok(snapshot())))
+        .expect("state-only inventory")
+        .expect("configured inventory");
+    let mut status = runtime.status();
+
+    assert!(matches!(
+        wait_for(&mut status, |status| matches!(status, InventoryStatus::Unchanged { .. })).await,
+        InventoryStatus::Unchanged { .. }
+    ));
+    let envelope: Value = serde_json::from_slice(&fs::read(state.join("inventory/latest.json")).expect("latest inventory"))
+        .expect("latest envelope");
+    assert_eq!(envelope["formatVersion"], "v1");
+    assert_eq!(envelope["snapshot"], serde_json::to_value(snapshot()).expect("snapshot JSON"));
+    assert_eq!(envelope.as_object().expect("envelope object").len(), 4);
+    runtime.shutdown().await;
+}
+
 #[tokio::test]
 async fn connect_inventory_rejects_deserialized_invalid_snapshots_before_state_or_network() {
     let pki = TestPki::new();
@@ -409,7 +448,7 @@ async fn connect_inventory_rejects_deserialized_invalid_snapshots_before_state_o
             "coarseFlags": []
         }))
         .expect("serde should not bypass the runtime validation boundary");
-        let temp = tempfile::tempdir().expect("tempdir");
+        let temp = safe_tempdir();
         let shutdown = CancellationToken::new();
         let runtime = spawn_inventory_runtime(Some(config(&temp, &pki, &server)), schedule(), &shutdown, move || {
             std::future::ready(Ok(invalid.clone()))
@@ -420,7 +459,7 @@ async fn connect_inventory_rejects_deserialized_invalid_snapshots_before_state_o
 
         assert!(matches!(
             wait_for(&mut status, |status| matches!(status, InventoryStatus::Failed { .. })).await,
-            InventoryStatus::Failed { reason } if reason.contains("version is outside protocol bounds")
+            InventoryStatus::Failed { reason } if reason == "connect_inventory_snapshot_version"
         ));
         assert!(!temp.path().join("private-config-secret/inventory/state.json").exists());
         runtime.shutdown().await;
@@ -433,7 +472,7 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
     let pki = TestPki::new();
     let content_hash = snapshot().content_hash().expect("content hash");
     let first_server = server(&pki, vec![Reply::error(StatusCode::SERVICE_UNAVAILABLE, "UNAVAILABLE")]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
+    let temp = safe_tempdir();
     let shutdown = CancellationToken::new();
     let samples = Arc::new(AtomicUsize::new(0));
     let sampled = samples.clone();
@@ -451,7 +490,40 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
     ));
     assert_eq!(samples.load(Ordering::Relaxed), 1);
     let original = first_server.seen.lock().expect("seen lock")[0].clone();
+    let latest: Value = serde_json::from_slice(
+        &fs::read(temp.path().join("private-config-secret/inventory/latest.json")).expect("latest inventory"),
+    )
+    .expect("latest envelope");
+    assert_eq!(latest["snapshot"], serde_json::to_value(snapshot()).expect("snapshot JSON"));
+    for field in [
+        "rustfsVersion",
+        "osVersion",
+        "nodeCount",
+        "driveCount",
+        "capacityTotalBytes",
+        "capacityUsedBytes",
+        "coarseFlags",
+    ] {
+        assert_eq!(latest["snapshot"][field], original[field]);
+    }
     runtime.shutdown().await;
+    let state = temp.path().join("private-config-secret/inventory/state.json");
+    let legacy_persisted_at = std::time::SystemTime::now() - Duration::from_secs(60 * 60);
+    fs::File::options()
+        .write(true)
+        .open(&state)
+        .expect("pending state")
+        .set_times(std::fs::FileTimes::new().set_modified(legacy_persisted_at))
+        .expect("legacy pending timestamp");
+    let legacy_persisted_at = chrono::DateTime::<chrono::Utc>::from(
+        fs::metadata(&state)
+            .and_then(|metadata| metadata.modified())
+            .expect("persisted pending timestamp"),
+    )
+    .format("%Y-%m-%dT%H:%M:%SZ")
+    .to_string();
+    fs::remove_file(temp.path().join("private-config-secret/inventory/latest.json"))
+        .expect("simulate a pending snapshot created before local persistence");
 
     let mut limited = Reply::error(StatusCode::TOO_MANY_REQUESTS, "RATE_LIMITED");
     limited.retry_after = Some("0");
@@ -473,10 +545,17 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
             if accepted == content_hash && received_at == "2026-08-22T01:02:03Z"
     ));
     assert_eq!(restart_samples.load(Ordering::Relaxed), 0);
+    let restored_latest: Value = serde_json::from_slice(
+        &fs::read(temp.path().join("private-config-secret/inventory/latest.json")).expect("restored latest inventory"),
+    )
+    .expect("restored latest envelope");
+    assert_eq!(restored_latest["snapshot"], serde_json::to_value(snapshot()).expect("snapshot JSON"));
+    assert_eq!(restored_latest["capturedAt"], legacy_persisted_at);
     let delivered = restart_server.seen.lock().expect("seen lock").clone();
     assert_eq!(delivered, vec![original.clone(), original.clone()]);
     assert_eq!(original["sequence"], 0);
     let encoded = serde_json::to_string(&original).expect("request JSON");
+    let persisted = serde_json::to_string(&latest).expect("persisted JSON");
     for forbidden in [
         "private-config-secret",
         "BEGIN CERTIFICATE",
@@ -486,9 +565,17 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
         "path",
     ] {
         assert!(!encoded.contains(forbidden), "request exposed {forbidden}");
+        assert!(!persisted.contains(forbidden), "persisted inventory exposed {forbidden}");
     }
     assert_eq!(original.as_object().expect("request object").len(), 10);
     restart.shutdown().await;
+    #[cfg(target_os = "linux")]
+    let latest_before_unchanged = {
+        use std::os::unix::fs::MetadataExt as _;
+        fs::metadata(temp.path().join("private-config-secret/inventory/latest.json"))
+            .expect("latest metadata")
+            .ino()
+    };
 
     let unchanged_samples = Arc::new(AtomicUsize::new(0));
     let sampled = unchanged_samples.clone();
@@ -505,6 +592,17 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
     ));
     assert_eq!(unchanged_samples.load(Ordering::Relaxed), 1);
     assert_eq!(restart_server.seen.lock().expect("seen lock").len(), 2);
+    #[cfg(target_os = "linux")]
+    {
+        use std::os::unix::fs::MetadataExt as _;
+        assert_ne!(
+            fs::metadata(temp.path().join("private-config-secret/inventory/latest.json"))
+                .expect("refreshed latest metadata")
+                .ino(),
+            latest_before_unchanged,
+            "a complete unchanged sample must refresh the local envelope"
+        );
+    }
     unchanged.shutdown().await;
 }
 
@@ -512,7 +610,7 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
 async fn connect_inventory_disconnect_retries_without_resampling() {
     let pki = TestPki::new();
     let unavailable = server(&pki, Vec::new()).await;
-    let temp = tempfile::tempdir().expect("tempdir");
+    let temp = safe_tempdir();
     let config = config(&temp, &pki, &unavailable);
     drop(unavailable);
     let shutdown = CancellationToken::new();
@@ -542,7 +640,7 @@ async fn connect_inventory_retries_an_incomplete_sample_before_delivery() {
     let pki = TestPki::new();
     let content_hash = snapshot().content_hash().expect("content hash");
     let server = server(&pki, vec![Reply::ok(&content_hash)]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
+    let temp = safe_tempdir();
     let shutdown = CancellationToken::new();
     let samples = Arc::new(AtomicUsize::new(0));
     let sampled = samples.clone();
@@ -561,6 +659,11 @@ async fn connect_inventory_retries_an_incomplete_sample_before_delivery() {
     .expect("configured inventory");
     let mut status = runtime.status();
 
+    assert!(matches!(
+        wait_for(&mut status, |status| matches!(status, InventoryStatus::BackingOff { .. })).await,
+        InventoryStatus::BackingOff { .. }
+    ));
+    assert!(!temp.path().join("private-config-secret/inventory/latest.json").exists());
     assert!(matches!(
         wait_for(&mut status, |status| matches!(status, InventoryStatus::Online { .. })).await,
         InventoryStatus::Online { content_hash: accepted, .. } if accepted == content_hash
@@ -575,7 +678,7 @@ async fn connect_inventory_unchanged_sample_resets_incomplete_backoff() {
     let pki = TestPki::new();
     let content_hash = snapshot().content_hash().expect("content hash");
     let server = server(&pki, vec![Reply::ok(&content_hash)]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
+    let temp = safe_tempdir();
     let shutdown = CancellationToken::new();
     let config = config(&temp, &pki, &server);
     let seed = spawn_inventory_runtime(Some(config.clone()), schedule(), &shutdown, || std::future::ready(Ok(snapshot())))
@@ -645,7 +748,7 @@ async fn connect_inventory_unchanged_sample_resets_incomplete_backoff() {
 async fn connect_inventory_revoked_device_stops_without_retrying() {
     let pki = TestPki::new();
     let server = server(&pki, vec![Reply::error(StatusCode::UNAUTHORIZED, "DEVICE_REVOKED")]).await;
-    let temp = tempfile::tempdir().expect("tempdir");
+    let temp = safe_tempdir();
     let shutdown = CancellationToken::new();
     let runtime = spawn_inventory_runtime(Some(config(&temp, &pki, &server)), schedule(), &shutdown, || {
         std::future::ready(Ok(snapshot()))
@@ -656,7 +759,10 @@ async fn connect_inventory_revoked_device_stops_without_retrying() {
 
     assert!(matches!(
         wait_for(&mut status, |status| matches!(status, InventoryStatus::AuthenticationStopped { .. })).await,
-        InventoryStatus::AuthenticationStopped { status: 401, reason: Some(reason) } if reason == "DEVICE_REVOKED"
+        InventoryStatus::AuthenticationStopped {
+            status: 401,
+            reason: None
+        }
     ));
     assert_eq!(server.seen.lock().expect("seen lock").len(), 1);
     runtime.shutdown().await;
@@ -666,10 +772,11 @@ async fn connect_inventory_revoked_device_stops_without_retrying() {
 async fn connect_inventory_sequence_overflow_fails_before_sampling_or_network_delivery() {
     let pki = TestPki::new();
     let server = server(&pki, Vec::new()).await;
-    let temp = tempfile::tempdir().expect("tempdir");
+    let temp = safe_tempdir();
     let config = config(&temp, &pki, &server);
     let state = temp.path().join("private-config-secret/inventory/state.json");
     fs::create_dir_all(state.parent().expect("state directory")).expect("create state directory");
+    private_directory_mode(state.parent().expect("state directory"));
     fs::write(
         &state,
         br#"{"nextSequence":9007199254740992,"pending":null,"lastAcceptedContentHash":null}"#,
@@ -689,7 +796,7 @@ async fn connect_inventory_sequence_overflow_fails_before_sampling_or_network_de
 
     assert!(matches!(
         wait_for(&mut status, |status| matches!(status, InventoryStatus::Failed { .. })).await,
-        InventoryStatus::Failed { reason } if reason.contains("sequence is exhausted")
+        InventoryStatus::Failed { reason } if reason == "connect_inventory_sequence_exhausted"
     ));
     assert_eq!(samples.load(Ordering::Relaxed), 0);
     assert!(server.seen.lock().expect("seen lock").is_empty());
@@ -702,10 +809,11 @@ async fn connect_inventory_rejects_noncanonical_persisted_snapshots_before_deliv
     let server = server(&pki, Vec::new()).await;
 
     for invalid_case in ["flags", "os-version"] {
-        let temp = tempfile::tempdir().expect("tempdir");
+        let temp = safe_tempdir();
         let config = config(&temp, &pki, &server);
         let state = temp.path().join("private-config-secret/inventory/state.json");
         fs::create_dir_all(state.parent().expect("state directory")).expect("create state directory");
+        private_directory_mode(state.parent().expect("state directory"));
         let mut pending = json!({
             "protocolVersion": "v1",
             "requestId": "00000000-0000-4000-8000-000000000001",
@@ -753,7 +861,7 @@ async fn connect_inventory_rejects_noncanonical_persisted_snapshots_before_deliv
                 matches!(status, InventoryStatus::Failed { .. } | InventoryStatus::BackingOff { .. })
             })
             .await,
-            InventoryStatus::Failed { reason } if reason.contains("violates the protocol invariants")
+            InventoryStatus::Failed { reason } if reason == "connect_inventory_state_corrupt"
         ));
         assert_eq!(samples.load(Ordering::Relaxed), 0);
         runtime.shutdown().await;
@@ -762,11 +870,20 @@ async fn connect_inventory_rejects_noncanonical_persisted_snapshots_before_deliv
     assert!(server.seen.lock().expect("seen lock").is_empty());
 }
 
-#[cfg(unix)]
+#[cfg(target_os = "linux")]
 fn private_mode(path: &std::path::Path) {
     use std::os::unix::fs::PermissionsExt as _;
     fs::set_permissions(path, fs::Permissions::from_mode(0o600)).expect("private permissions");
 }
 
-#[cfg(not(unix))]
+#[cfg(target_os = "linux")]
+fn private_directory_mode(path: &std::path::Path) {
+    use std::os::unix::fs::PermissionsExt as _;
+    fs::set_permissions(path, fs::Permissions::from_mode(0o700)).expect("private directory permissions");
+}
+
+#[cfg(not(target_os = "linux"))]
 fn private_mode(_path: &std::path::Path) {}
+
+#[cfg(not(target_os = "linux"))]
+fn private_directory_mode(_path: &std::path::Path) {}