From e13049a846d53dae06fbbe0cd3b7dc0bd6e79fc9 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 22 Aug 2026 15:54:42 +0800 Subject: [PATCH] feat(connect): emit durable heartbeats --- protocol/agent/v1/fixtures/fixture-sets.json | 2 +- .../v1/fixtures/heartbeat/MANIFEST.sha256 | 5 + .../v1/fixtures/heartbeat/duplicate.json | 9 + .../agent/v1/fixtures/heartbeat/overflow.json | 11 + .../agent/v1/fixtures/heartbeat/stale.json | 9 + .../agent/v1/fixtures/heartbeat/unknown.json | 18 + .../agent/v1/fixtures/heartbeat/valid.json | 21 + rustfs/src/connect/config.rs | 173 +++++ rustfs/src/connect/heartbeat.rs | 585 +++++++++++++++ rustfs/src/connect/mod.rs | 12 +- rustfs/src/connect/runtime.rs | 156 ++++ rustfs/src/startup_lifecycle.rs | 4 + rustfs/src/startup_services.rs | 21 + rustfs/tests/connect_heartbeat.rs | 687 ++++++++++++++++++ 14 files changed, 1709 insertions(+), 4 deletions(-) create mode 100644 protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 create mode 100644 protocol/agent/v1/fixtures/heartbeat/duplicate.json create mode 100644 protocol/agent/v1/fixtures/heartbeat/overflow.json create mode 100644 protocol/agent/v1/fixtures/heartbeat/stale.json create mode 100644 protocol/agent/v1/fixtures/heartbeat/unknown.json create mode 100644 protocol/agent/v1/fixtures/heartbeat/valid.json create mode 100644 rustfs/src/connect/config.rs create mode 100644 rustfs/src/connect/heartbeat.rs create mode 100644 rustfs/src/connect/runtime.rs create mode 100644 rustfs/tests/connect_heartbeat.rs diff --git a/protocol/agent/v1/fixtures/fixture-sets.json b/protocol/agent/v1/fixtures/fixture-sets.json index d3beb8f23..72c4c1919 100644 --- a/protocol/agent/v1/fixtures/fixture-sets.json +++ b/protocol/agent/v1/fixtures/fixture-sets.json @@ -25,7 +25,7 @@ }, { "name": "heartbeat", - "status": "reserved", + "status": "populated", "purpose": "Heartbeat payloads, Connect receive time, and freshness window behavior." }, { diff --git a/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 b/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 new file mode 100644 index 000000000..7aa8e3191 --- /dev/null +++ b/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 @@ -0,0 +1,5 @@ +975c1ca53eefeef6766a6fc0b3d3281f7408255342b0686e5e2aee5ad055414c duplicate.json +963529a38a02849c6c2acc6d72668dca9f63218b49c89fae41a451b584850411 overflow.json +e3adeee1c8a19aa17e70894896fb79c072e3785bea3611b93c11e79f039ed5af stale.json +35b9cebd8525389a701e8fe69fbe96407bcb31aa28392fe95babf4a4886985ad unknown.json +37941735dbd6ad3d238258a7b2cae6f0b3aa0ecaae1d8817817c3d718d11d633 valid.json diff --git a/protocol/agent/v1/fixtures/heartbeat/duplicate.json b/protocol/agent/v1/fixtures/heartbeat/duplicate.json new file mode 100644 index 000000000..11cad8ac1 --- /dev/null +++ b/protocol/agent/v1/fixtures/heartbeat/duplicate.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} +} diff --git a/protocol/agent/v1/fixtures/heartbeat/overflow.json b/protocol/agent/v1/fixtures/heartbeat/overflow.json new file mode 100644 index 000000000..7b62b5b45 --- /dev/null +++ b/protocol/agent/v1/fixtures/heartbeat/overflow.json @@ -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"} +} diff --git a/protocol/agent/v1/fixtures/heartbeat/stale.json b/protocol/agent/v1/fixtures/heartbeat/stale.json new file mode 100644 index 000000000..aabe98a6f --- /dev/null +++ b/protocol/agent/v1/fixtures/heartbeat/stale.json @@ -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} +} diff --git a/protocol/agent/v1/fixtures/heartbeat/unknown.json b/protocol/agent/v1/fixtures/heartbeat/unknown.json new file mode 100644 index 000000000..6639cc0ff --- /dev/null +++ b/protocol/agent/v1/fixtures/heartbeat/unknown.json @@ -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": [] + } +} diff --git a/protocol/agent/v1/fixtures/heartbeat/valid.json b/protocol/agent/v1/fixtures/heartbeat/valid.json new file mode 100644 index 000000000..ed3561b80 --- /dev/null +++ b/protocol/agent/v1/fixtures/heartbeat/valid.json @@ -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" + } +} diff --git a/rustfs/src/connect/config.rs b/rustfs/src/connect/config.rs new file mode 100644 index 000000000..391d4ef45 --- /dev/null +++ b/rustfs/src/connect/config.rs @@ -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, + pub identity_store: IdentityStore, + pub credential_store: CredentialStore, + pub state_path: PathBuf, + pub schedule: HeartbeatSchedule, +} + +impl HeartbeatConfig { + pub fn new( + endpoint: impl Into, + root_ca_pem: impl Into>, + identity_store: IdentityStore, + credential_store: CredentialStore, + state_path: impl Into, + ) -> 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, 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, + root_ca_file: Option, + state_dir: Option, + ) -> Result, 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"); + } +} diff --git a/rustfs/src/connect/heartbeat.rs b/rustfs/src/connect/heartbeat.rs new file mode 100644 index 000000000..2eeb8c135 --- /dev/null +++ b/rustfs/src/connect/heartbeat.rs @@ -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 { + 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 }, + 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, +} + +pub(crate) enum Delivery { + Accepted { server_time: String }, + Retry { retry_after: Option }, + AuthenticationStopped { status: u16, reason: Option }, + Rejected { status: u16, reason: Option }, +} + +pub(crate) struct HeartbeatSender { + endpoint: Url, + root_store: RootCertStore, + roots: Vec>, + config: HeartbeatConfig, +} + +impl HeartbeatSender { + pub(crate) fn new(config: HeartbeatConfig) -> Result { + 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::, _>>() + .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 { + 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) -> Result { + 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::, _>>()?; + 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, +} + +impl HeartbeatStateStore { + pub(crate) fn new(path: PathBuf) -> Self { + Self { path } + } + + pub(crate) fn try_runtime_lock(&self) -> Result { + 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, + ) -> Result { + 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) -> Result { + 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 { + 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, maximum: Duration) -> Option { + let value = headers.get(header::RETRY_AFTER)?.to_str().ok()?; + let delay = value.parse::().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 { + #[derive(Deserialize)] + struct Envelope { + #[serde(default)] + details: Vec, + } + #[derive(Deserialize)] + struct Detail { + #[serde(default)] + reason: String, + } + + serde_json::from_slice::(&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, 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 { + 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), +} diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 350cceb38..3ce2ad975 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -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}; diff --git a/rustfs/src/connect/runtime.rs b/rustfs/src/connect/runtime.rs new file mode 100644 index 000000000..f0d919a07 --- /dev/null +++ b/rustfs/src/connect/runtime.rs @@ -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, + task: Option>, +} + +impl HeartbeatRuntime { + pub fn status(&self) -> watch::Receiver { + 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( + config: Option, + parent_shutdown: &CancellationToken, + sample: F, +) -> Result, 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, 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(shutdown: &CancellationToken, future: impl Future) -> Option { + 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, + } +} diff --git a/rustfs/src/startup_lifecycle.rs b/rustfs/src/startup_lifecycle.rs index c4d09872b..ad5681366 100644 --- a/rustfs/src/startup_lifecycle.rs +++ b/rustfs/src/startup_lifecycle.rs @@ -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", diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 585e9677a..c07db0f67 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -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, 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, + node_count: Option, + shutdown: &CancellationToken, +) -> Result> { + 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) +} diff --git a/rustfs/tests/connect_heartbeat.rs b/rustfs/tests/connect_heartbeat.rs new file mode 100644 index 000000000..9414cce46 --- /dev/null +++ b/rustfs/tests/connect_heartbeat.rs @@ -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>>, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for TestServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn server(pki: &TestPki, replies: Vec) -> 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| { + 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, + predicate: impl Fn(&HeartbeatStatus) -> bool, +) -> HeartbeatStatus { + tokio::time::timeout(Duration::from_secs(3), async { + loop { + let current = status.borrow_and_update().clone(); + if predicate(¤t) { + 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::>(); + 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) {}