From d8ef9e803ee65fdb02f410ada6f36b41e96c6f05 Mon Sep 17 00:00:00 2001 From: Chris Date: Sun, 13 Sep 2026 04:01:17 +0800 Subject: [PATCH] feat(connect): advertise environment diagnostics (#7714) --- rustfs/src/connect/heartbeat.rs | 10 ++++++- rustfs/tests/connect_heartbeat.rs | 49 +++++++++++++++++++++++++++++-- 2 files changed, 55 insertions(+), 4 deletions(-) diff --git a/rustfs/src/connect/heartbeat.rs b/rustfs/src/connect/heartbeat.rs index 8b842d082..51e77ea64 100644 --- a/rustfs/src/connect/heartbeat.rs +++ b/rustfs/src/connect/heartbeat.rs @@ -25,6 +25,7 @@ use uuid::Uuid; use super::config::HeartbeatConfig; use super::credential_store::CredentialStoreError; use super::diagnostics::DiagnosticCollectionPolicy; +use super::environment::ENVIRONMENT_CAPABILITY; use super::identity::IdentityError; use super::identity_store::StoreError; use super::registration::CredentialValidationError; @@ -94,7 +95,13 @@ impl PendingHeartbeat { self.protocol_version == PROTOCOL_VERSION && self.agent_version == AGENT_VERSION && (self.capabilities == ["heartbeat"] - || self.capabilities == ["heartbeat", DiagnosticCollectionPolicy::policy_sync_capability()]) + || self.capabilities == ["heartbeat", DiagnosticCollectionPolicy::policy_sync_capability()] + || self.capabilities + == [ + "heartbeat", + DiagnosticCollectionPolicy::policy_sync_capability(), + ENVIRONMENT_CAPABILITY, + ]) && self.sequence <= MAX_SEQUENCE && self.coarse_node_summary.is_valid() && is_exact_utc_seconds(&self.client_time) @@ -243,6 +250,7 @@ impl HeartbeatStateStore { capabilities: vec![ "heartbeat".to_owned(), DiagnosticCollectionPolicy::policy_sync_capability().to_owned(), + ENVIRONMENT_CAPABILITY.to_owned(), ], sequence: state.next_sequence, client_time: now.to_rfc3339_opts(SecondsFormat::Secs, true), diff --git a/rustfs/tests/connect_heartbeat.rs b/rustfs/tests/connect_heartbeat.rs index 7a83200e0..4a72efbb2 100644 --- a/rustfs/tests/connect_heartbeat.rs +++ b/rustfs/tests/connect_heartbeat.rs @@ -28,8 +28,8 @@ use rcgen::{ KeyUsagePurpose, SanType, }; use rustfs::connect::{ - CoarseNodeSummary, CredentialStore, DeviceCredential, HeartbeatConfig, HeartbeatSchedule, HeartbeatStatus, IdentityStore, - spawn_heartbeat_runtime, + CoarseNodeSummary, CredentialStore, DeviceCredential, ENVIRONMENT_CAPABILITY, HeartbeatConfig, HeartbeatSchedule, + HeartbeatStatus, IdentityStore, spawn_heartbeat_runtime, }; use rustls::RootCertStore; use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; @@ -514,12 +514,55 @@ async fn sends_only_l0_fields_and_accepts_additive_response_fields() { "sequence" ] ); - assert_eq!(request["capabilities"], json!(["heartbeat", "diagnostics.policy.v1"])); + assert_eq!(ENVIRONMENT_CAPABILITY, "inventory.environment@1"); + assert_eq!( + request["capabilities"], + json!(["heartbeat", "diagnostics.policy.v1", "inventory.environment@1"]) + ); 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_a_pending_heartbeat_from_before_environment_collection() { + 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"); + let state = json!({ + "nextSequence": 0, + "pending": { + "protocolVersion": "v1", + "requestId": "550e8400-e29b-41d4-a716-446655440000", + "agentVersion": format!("rustfs-agent/{}", env!("CARGO_PKG_VERSION")), + "capabilities": ["heartbeat", "diagnostics.policy.v1"], + "sequence": 0, + "clientTime": "2026-08-22T01:02:03Z", + "coarseNodeSummary": {"total": 1, "healthy": 1, "degraded": 0} + } + }); + fs::write(&config.state_path, serde_json::to_vec(&state).expect("heartbeat state JSON")).expect("write heartbeat 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::Online { .. })).await, + HeartbeatStatus::Online { .. } + )); + runtime.shutdown().await; + + let seen = server.seen.lock().expect("seen lock"); + assert_eq!(seen.len(), 1); + assert_eq!(seen[0]["capabilities"], json!(["heartbeat", "diagnostics.policy.v1"])); +} + #[tokio::test] async fn restart_replays_pending_request_then_advances_sequence() { let pki = TestPki::new();