feat(connect): rotate device credentials at runtime (#6586)

* feat(connect): rotate credentials from heartbeat runtime

* fix(connect): make rotation safe with in-flight telemetry

* fix(connect): keep rotation retry state private

* fix(connect): preserve public rotation retries

* fix(connect): preserve heartbeat error API

* fix(connect): keep heartbeat alive during reenrollment

* fix(connect): validate pending reenrollment before skipping

* fix(connect): validate pending reenrollment token

* fix(connect): bind pending reenrollment state

* fix(connect): recover credentials before telemetry

rustfs/src/connect/client.rs

@@ -15,7 +15,8 @@
 use std::time::Duration;
 
 use base64::Engine as _;
-use reqwest::{Client, StatusCode, Url};
+use chrono::{DateTime, Utc};
+use reqwest::{Client, StatusCode, Url, header};
 use rustls::RootCertStore;
 use rustls::pki_types::{CertificateDer, pem::PemObject as _};
 use serde::Deserialize;
@@ -23,19 +24,21 @@ use uuid::Uuid;
 use zeroize::Zeroizing;
 
 use super::credential_store::{
-    CompletedRegistration, CredentialStore, CredentialStoreError, DeviceCredential, PendingRegistration, PendingRotation,
+    CompletedRegistration, CredentialLock, CredentialStore, CredentialStoreError, DeviceCredential, PendingRegistration,
+    PendingRotation,
 };
 use super::identity::{IdentityError, RegistrationTranscript};
 use super::identity_store::{IdentityStore, StoreError};
 use super::registration::{
     CredentialResponse, CredentialValidationError, ExpectedDevice, RegistrationRequest, RegistrationToken, RotationRequest,
-    certificate_fingerprint, certificate_request_matches, public_key_fingerprint, validate_credential,
+    certificate_fingerprint, certificate_request_matches, is_uuid_v7, public_key_fingerprint, validate_credential,
     validate_stored_credential,
 };
 
 const MAX_ATTEMPTS: usize = 3;
 const MAX_RESPONSE_BYTES: usize = 1024 * 1024;
 const ROTATION_THRESHOLD_SECONDS: i64 = 8 * 60 * 60;
+const PENDING_REGISTRATION_STATE_DOMAIN: &[u8] = b"RUSTFS-CONNECT-PENDING-REGISTRATION-V1";
 
 pub struct ConnectConfig<'a> {
     pub endpoint: &'a str,
@@ -51,6 +54,23 @@ pub struct ConnectClient {
     timeout: Duration,
 }
 
+pub(crate) enum RotationAttempt {
+    Completed(Option<DeviceCredential>),
+    ReenrollmentPending,
+    Unavailable {
+        status: Option<StatusCode>,
+        retry_after: Option<Duration>,
+    },
+}
+
+enum SingleRequest {
+    Response(CredentialResponse),
+    Unavailable {
+        status: Option<StatusCode>,
+        retry_after: Option<Duration>,
+    },
+}
+
 impl ConnectClient {
     pub fn from_optional_config(config: Option<ConnectConfig<'_>>) -> Result<Option<Self>, ClientError> {
         config.map(Self::new).transpose()
@@ -100,26 +120,31 @@ impl ConnectClient {
         credential_store: &CredentialStore,
         token: &RegistrationToken,
     ) -> Result<DeviceCredential, ClientError> {
-        let _lock = credential_store.lock().await?;
+        let lock = credential_store.lock().await?;
-        if let Some((credential, _)) = self.load_valid_credential(identity_store, credential_store)? {
+        if let Some((credential, _, _)) =
+            Self::recover_valid_credential_locked(&lock, identity_store, credential_store, &self.roots, &self.root_certificates)?
+        {
             ensure_credential_time(&credential, unix_now())?;
             return Ok(credential);
         }
 
         let identity = identity_store.load_or_create()?;
-        let candidate = PendingRegistration {
+        let mut candidate = PendingRegistration {
             token_uid: token.registration_token_uid.clone(),
             request_id: Uuid::new_v4().to_string(),
             certificate_request: identity.certificate_request_base64()?,
             previous_credential_fingerprint: None,
             next_public_key_sha256: None,
+            state_proof: String::new(),
         };
+        candidate.state_proof = identity.sign_pending_registration_state(&pending_registration_state(&candidate));
         let pending = credential_store.claim_pending_registration(&candidate)?;
         if pending.token_uid != token.registration_token_uid
             || pending.previous_credential_fingerprint.is_some()
             || pending.next_public_key_sha256.is_some()
             || !is_request_id(&pending.request_id)
             || !certificate_request_matches(&pending.certificate_request, &identity)?
+            || !pending_registration_is_bound(&pending, &identity)
         {
             return Err(ClientError::PendingRegistration);
         }
@@ -143,10 +168,10 @@ impl ConnectClient {
         credential_store: &CredentialStore,
         token: &RegistrationToken,
     ) -> Result<DeviceCredential, ClientError> {
-        let _lock = credential_store.lock().await?;
+        let lock = credential_store.lock().await?;
-        let (credential, _) = self
+        let (credential, _, _) =
-            .load_valid_credential(identity_store, credential_store)?
+            Self::recover_valid_credential_locked(&lock, identity_store, credential_store, &self.roots, &self.root_certificates)?
-            .ok_or(ClientError::NotRegistered)?;
+                .ok_or(ClientError::NotRegistered)?;
         let fingerprint = certificate_fingerprint(&credential.certificate)?;
         if credential_store.load_completed_registration()?.is_some_and(|completed| {
             completed.token_uid == token.registration_token_uid && completed.credential_fingerprint == fingerprint
@@ -156,19 +181,22 @@ impl ConnectClient {
         credential_store.clear_pending_rotation()?;
         let next = identity_store.load_or_create_next()?;
         let next_fingerprint = public_key_fingerprint(&next);
-        let candidate = PendingRegistration {
+        let mut candidate = PendingRegistration {
             token_uid: token.registration_token_uid.clone(),
             request_id: Uuid::new_v4().to_string(),
             certificate_request: next.certificate_request_base64()?,
             previous_credential_fingerprint: Some(fingerprint.clone()),
             next_public_key_sha256: Some(next_fingerprint.clone()),
+            state_proof: String::new(),
         };
+        candidate.state_proof = next.sign_pending_registration_state(&pending_registration_state(&candidate));
         let pending = credential_store.claim_pending_registration(&candidate)?;
         if pending.token_uid != token.registration_token_uid
             || pending.previous_credential_fingerprint.as_deref() != Some(&fingerprint)
             || pending.next_public_key_sha256.as_deref() != Some(&next_fingerprint)
             || !is_request_id(&pending.request_id)
             || !certificate_request_matches(&pending.certificate_request, &next)?
+            || !pending_registration_is_bound(&pending, &next)
         {
             return Err(ClientError::PendingRegistration);
         }
@@ -232,16 +260,39 @@ impl ConnectClient {
         credential_store: &CredentialStore,
         now_unix: i64,
     ) -> Result<Option<DeviceCredential>, ClientError> {
-        let _lock = credential_store.lock().await?;
+        let mut last_status = None;
-        let (credential, identity) = self
+        for attempt in 0..MAX_ATTEMPTS {
-            .load_valid_credential(identity_store, credential_store)?
+            match self.rotate_if_due_once(identity_store, credential_store, now_unix).await? {
-            .ok_or(ClientError::NotRegistered)?;
+                RotationAttempt::Completed(credential) => return Ok(credential),
-        if credential_store.load_pending_registration()?.is_some() {
+                RotationAttempt::ReenrollmentPending => return Err(ClientError::PendingRegistration),
-            return Err(ClientError::PendingRegistration);
+                RotationAttempt::Unavailable {
+                    status: Some(status), ..
+                } => last_status = Some(status),
+                RotationAttempt::Unavailable { status: None, .. } => {}
+            }
+            if attempt + 1 < MAX_ATTEMPTS {
+                tokio::time::sleep(Duration::from_millis(50 * (attempt as u64 + 1))).await;
+            }
+        }
+        Err(ClientError::Unavailable { status: last_status })
+    }
+
+    pub(crate) async fn rotate_if_due_once(
+        &self,
+        identity_store: &IdentityStore,
+        credential_store: &CredentialStore,
+        now_unix: i64,
+    ) -> Result<RotationAttempt, ClientError> {
+        let lock = credential_store.lock().await?;
+        let (credential, identity, reenrollment_pending) =
+            Self::recover_valid_credential_locked(&lock, identity_store, credential_store, &self.roots, &self.root_certificates)?
+                .ok_or(ClientError::NotRegistered)?;
+        if reenrollment_pending {
+            return Ok(RotationAttempt::ReenrollmentPending);
         }
         ensure_credential_time(&credential, now_unix)?;
         if credential.not_after_unix - now_unix > ROTATION_THRESHOLD_SECONDS {
-            return Ok(None);
+            return Ok(RotationAttempt::Completed(None));
         }
         let fingerprint = certificate_fingerprint(&credential.certificate)?;
         let next = identity_store.load_or_create_next()?;
@@ -277,7 +328,12 @@ impl ConnectClient {
         let client = build_client(&self.root_certificates, self.timeout, Some(tls_identity))?;
         let path = format!("clusterDevices/{}:rotateCredential", credential.uid);
         let url = self.url(&path)?;
-        let response = self.send(StatusCode::OK, || client.post(url.clone()).json(&body)).await?;
+        let response = match self.send_once(StatusCode::OK, client.post(url).json(&body)).await? {
+            SingleRequest::Response(response) => response,
+            SingleRequest::Unavailable { status, retry_after } => {
+                return Ok(RotationAttempt::Unavailable { status, retry_after });
+            }
+        };
         let rotated = validate_credential(
             response,
             &next,
@@ -291,29 +347,35 @@ impl ConnectClient {
         credential_store.save(&rotated)?;
         identity_store.commit_next(&next)?;
         credential_store.clear_pending_rotation()?;
-        Ok(Some(rotated))
+        Ok(RotationAttempt::Completed(Some(rotated)))
     }
 
-    fn load_valid_credential(
+    pub(crate) fn recover_valid_credential_locked(
-        &self,
+        _lock: &CredentialLock,
         identity_store: &IdentityStore,
         credential_store: &CredentialStore,
-    ) -> Result<Option<(DeviceCredential, super::identity::DeviceIdentity)>, ClientError> {
+        roots: &RootCertStore,
+        root_certificates: &[CertificateDer<'static>],
+    ) -> Result<Option<(DeviceCredential, super::identity::DeviceIdentity, bool)>, ClientError> {
         let Some(credential) = credential_store.load()? else {
             return Ok(None);
         };
         let current = identity_store.load()?.ok_or(ClientError::IdentityMissing)?;
         if let Some(pending) = credential_store.load_pending_registration()? {
+            if !is_uuid_v7(&pending.token_uid) {
+                return Err(ClientError::PendingRegistration);
+            }
             let Some(previous) = pending.previous_credential_fingerprint.as_deref() else {
                 if pending.next_public_key_sha256.is_some()
                     || !is_request_id(&pending.request_id)
                     || !certificate_request_matches(&pending.certificate_request, &current)?
+                    || !pending_registration_is_bound(&pending, &current)
                 {
                     return Err(ClientError::PendingRegistration);
                 }
-                validate_stored_credential(&credential, &current, &self.roots, &self.root_certificates)?;
+                validate_stored_credential(&credential, &current, roots, root_certificates)?;
                 credential_store.clear_pending_registration()?;
-                return Ok(Some((credential, current)));
+                return Ok(Some((credential, current, false)));
             };
             let next_fingerprint = pending
                 .next_public_key_sha256
@@ -324,28 +386,38 @@ impl ConnectClient {
             }
             let fingerprint = certificate_fingerprint(&credential.certificate)?;
             if fingerprint == previous {
-                validate_stored_credential(&credential, &current, &self.roots, &self.root_certificates)?;
+                validate_stored_credential(&credential, &current, roots, root_certificates)?;
                 let next = identity_store.load_next()?.ok_or(ClientError::PendingRegistration)?;
                 if public_key_fingerprint(&next) != next_fingerprint
                     || !certificate_request_matches(&pending.certificate_request, &next)?
+                    || !pending_registration_is_bound(&pending, &next)
                 {
                     return Err(ClientError::PendingRegistration);
                 }
-                return Ok(Some((credential, current)));
+                if credential_store.load_completed_registration()?.is_some_and(|completed| {
-            }
+                    completed.credential_fingerprint == fingerprint
-            if public_key_fingerprint(&current) == next_fingerprint {
+                        && (!is_uuid_v7(&completed.token_uid) || completed.token_uid == pending.token_uid)
-                if !certificate_request_matches(&pending.certificate_request, &current)? {
+                }) {
                     return Err(ClientError::PendingRegistration);
                 }
-                validate_stored_credential(&credential, &current, &self.roots, &self.root_certificates)?;
+                return Ok(Some((credential, current, true)));
+            }
+            if public_key_fingerprint(&current) == next_fingerprint {
+                if !certificate_request_matches(&pending.certificate_request, &current)?
+                    || !pending_registration_is_bound(&pending, &current)
+                {
+                    return Err(ClientError::PendingRegistration);
+                }
+                validate_stored_credential(&credential, &current, roots, root_certificates)?;
             } else {
                 let next = identity_store.load_next()?.ok_or(ClientError::PendingRegistration)?;
                 if public_key_fingerprint(&next) != next_fingerprint
                     || !certificate_request_matches(&pending.certificate_request, &next)?
+                    || !pending_registration_is_bound(&pending, &next)
                 {
                     return Err(ClientError::PendingRegistration);
                 }
-                validate_stored_credential(&credential, &next, &self.roots, &self.root_certificates)?;
+                validate_stored_credential(&credential, &next, roots, root_certificates)?;
                 identity_store.commit_next(&next)?;
             }
             credential_store.save_completed_registration(&CompletedRegistration {
@@ -354,32 +426,32 @@ impl ConnectClient {
             })?;
             credential_store.clear_pending_registration()?;
             let current = identity_store.load()?.ok_or(ClientError::IdentityMissing)?;
-            return Ok(Some((credential, current)));
+            return Ok(Some((credential, current, false)));
         }
         let Some(pending) = credential_store.load_pending_rotation()? else {
-            validate_stored_credential(&credential, &current, &self.roots, &self.root_certificates)?;
+            validate_stored_credential(&credential, &current, roots, root_certificates)?;
-            return Ok(Some((credential, current)));
+            return Ok(Some((credential, current, false)));
         };
         let fingerprint = certificate_fingerprint(&credential.certificate)?;
         if pending.device_name != credential.name || !is_request_id(&pending.request_id) {
             return Err(ClientError::PendingRotation);
         }
         if fingerprint == pending.credential_fingerprint {
-            validate_stored_credential(&credential, &current, &self.roots, &self.root_certificates)?;
+            validate_stored_credential(&credential, &current, roots, root_certificates)?;
             let next = identity_store.load_next()?.ok_or(ClientError::PendingRotation)?;
             if pending.next_public_key_sha256 != public_key_fingerprint(&next)
                 || !certificate_request_matches(&pending.certificate_request, &next)?
             {
                 return Err(ClientError::PendingRotation);
             }
-            return Ok(Some((credential, current)));
+            return Ok(Some((credential, current, false)));
         }
 
         if public_key_fingerprint(&current) == pending.next_public_key_sha256 {
             if !certificate_request_matches(&pending.certificate_request, &current)? {
                 return Err(ClientError::PendingRotation);
             }
-            validate_stored_credential(&credential, &current, &self.roots, &self.root_certificates)?;
+            validate_stored_credential(&credential, &current, roots, root_certificates)?;
         } else {
             let next = identity_store.load_next()?.ok_or(ClientError::PendingRotation)?;
             if public_key_fingerprint(&next) != pending.next_public_key_sha256
@@ -387,12 +459,12 @@ impl ConnectClient {
             {
                 return Err(ClientError::PendingRotation);
             }
-            validate_stored_credential(&credential, &next, &self.roots, &self.root_certificates)?;
+            validate_stored_credential(&credential, &next, roots, root_certificates)?;
             identity_store.commit_next(&next)?;
         }
         credential_store.clear_pending_rotation()?;
         let current = identity_store.load()?.ok_or(ClientError::IdentityMissing)?;
-        Ok(Some((credential, current)))
+        Ok(Some((credential, current, false)))
     }
 
     async fn send<F>(&self, success: StatusCode, mut request: F) -> Result<CredentialResponse, ClientError>
@@ -435,6 +507,35 @@ impl ConnectClient {
         Err(ClientError::Unavailable { status: last_status })
     }
 
+    async fn send_once(&self, success: StatusCode, request: reqwest::RequestBuilder) -> Result<SingleRequest, ClientError> {
+        let response = match request.send().await {
+            Ok(response) => response,
+            Err(error) if error.is_timeout() || error.is_connect() => {
+                return Ok(SingleRequest::Unavailable {
+                    status: None,
+                    retry_after: None,
+                });
+            }
+            Err(error) => return Err(ClientError::Transport(error)),
+        };
+        let status = response.status();
+        if status == success {
+            return decode_response(response).await.map(SingleRequest::Response);
+        }
+        if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) {
+            let reason = decode_reason(response).await;
+            return Err(ClientError::AccessRevoked { status, reason });
+        }
+        if matches!(status, StatusCode::REQUEST_TIMEOUT | StatusCode::TOO_MANY_REQUESTS) || status.is_server_error() {
+            return Ok(SingleRequest::Unavailable {
+                status: Some(status),
+                retry_after: retry_after(response.headers(), Utc::now()),
+            });
+        }
+        let reason = decode_reason(response).await;
+        Err(ClientError::Rejected { status, reason })
+    }
+
     fn url(&self, path: &str) -> Result<Url, ClientError> {
         self.endpoint.join(path).map_err(|_| ClientError::Endpoint)
     }
@@ -444,6 +545,32 @@ fn is_request_id(value: &str) -> bool {
     Uuid::parse_str(value).is_ok_and(|uuid| uuid.get_version() == Some(uuid::Version::Random) && uuid.to_string() == value)
 }
 
+fn pending_registration_is_bound(pending: &PendingRegistration, identity: &super::identity::DeviceIdentity) -> bool {
+    identity.verifies_pending_registration_state(&pending_registration_state(pending), &pending.state_proof)
+}
+
+fn pending_registration_state(pending: &PendingRegistration) -> Vec<u8> {
+    let mut state = Vec::with_capacity(PENDING_REGISTRATION_STATE_DOMAIN.len() + 512);
+    state.extend_from_slice(PENDING_REGISTRATION_STATE_DOMAIN);
+    for field in [
+        Some(pending.token_uid.as_str()),
+        Some(pending.request_id.as_str()),
+        Some(pending.certificate_request.as_str()),
+        pending.previous_credential_fingerprint.as_deref(),
+        pending.next_public_key_sha256.as_deref(),
+    ] {
+        match field {
+            Some(value) => {
+                state.push(1);
+                state.extend_from_slice(&(value.len() as u64).to_be_bytes());
+                state.extend_from_slice(value.as_bytes());
+            }
+            None => state.push(0),
+        }
+    }
+    state
+}
+
 fn unix_now() -> i64 {
     std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -460,6 +587,15 @@ fn ensure_credential_time(credential: &DeviceCredential, now_unix: i64) -> Resul
     Ok(())
 }
 
+fn retry_after(headers: &header::HeaderMap, now: DateTime<Utc>) -> Option<Duration> {
+    let value = headers.get(header::RETRY_AFTER)?.to_str().ok()?;
+    value.parse::<u64>().ok().map(Duration::from_secs).or_else(|| {
+        DateTime::parse_from_rfc2822(value)
+            .ok()
+            .and_then(|at| (at.with_timezone(&Utc) - now).to_std().ok())
+    })
+}
+
 fn build_client(
     roots: &[CertificateDer<'static>],
     timeout: Duration,

rustfs/src/connect/credential_store.rs

@@ -66,6 +66,7 @@ pub(crate) struct PendingRegistration {
     pub previous_credential_fingerprint: Option<String>,
     #[serde(default)]
     pub next_public_key_sha256: Option<String>,
+    pub state_proof: String,
 }
 
 #[derive(Serialize, Deserialize)]

rustfs/src/connect/heartbeat.rs

@@ -430,6 +430,7 @@ impl From<TelemetryError> for HeartbeatError {
             TelemetryError::IdentityCertificate => Self::IdentityCertificate,
             TelemetryError::CredentialName => Self::CredentialName,
             TelemetryError::CredentialExpired => Self::CredentialExpired,
+            TelemetryError::StateConflict => Self::StateConflict,
             TelemetryError::ResponseTooLarge => Self::ResponseTooLarge,
             TelemetryError::Url(error) => Self::Url(error),
             TelemetryError::Transport(error) => Self::Transport(error),

rustfs/src/connect/identity.rs

@@ -22,7 +22,7 @@
 
 use base64::Engine as _;
 use base64::engine::general_purpose::{STANDARD as BASE64_STANDARD, URL_SAFE_NO_PAD as BASE64_URL_NO_PAD};
-use p256::ecdsa::signature::Signer as _;
+use p256::ecdsa::signature::{Signer as _, Verifier as _};
 use p256::ecdsa::{Signature, SigningKey};
 use p256::elliptic_curve::Generate as _;
 use p256::pkcs8::{DecodePrivateKey as _, EncodePrivateKey as _, LineEnding};
@@ -256,6 +256,24 @@ impl DeviceIdentity {
         }
     }
 
+    pub(crate) fn sign_pending_registration_state(&self, state: &[u8]) -> String {
+        let signature: Signature = self.signing_key.sign(state);
+        BASE64_URL_NO_PAD.encode(signature.normalize_s().to_bytes())
+    }
+
+    pub(crate) fn verifies_pending_registration_state(&self, state: &[u8], proof: &str) -> bool {
+        let Ok(octets) = BASE64_URL_NO_PAD.decode(proof) else {
+            return false;
+        };
+        if BASE64_URL_NO_PAD.encode(&octets) != proof {
+            return false;
+        }
+        let Ok(signature) = Signature::from_slice(&octets) else {
+            return false;
+        };
+        signature.normalize_s() == signature && self.signing_key.verifying_key().verify(state, &signature).is_ok()
+    }
+
     /// The device public key, DER SubjectPublicKeyInfo.
     pub fn public_key_der(&self) -> Vec<u8> {
         use p256::pkcs8::EncodePublicKey as _;

rustfs/src/connect/registration.rs

@@ -496,7 +496,7 @@ fn canonical_serial(raw: &[u8]) -> Result<String, CredentialValidationError> {
     Ok(hex_lower(&padded))
 }
 
-fn is_uuid_v7(value: &str) -> bool {
+pub(crate) fn is_uuid_v7(value: &str) -> bool {
     Uuid::parse_str(value).is_ok_and(|uuid| uuid.get_version() == Some(Version::SortRand) && uuid.to_string() == value)
 }
 

rustfs/src/connect/runtime.rs

@@ -20,8 +20,10 @@ use chrono::Utc;
 use rand::RngExt as _;
 use tokio::sync::watch;
 use tokio::task::JoinHandle;
+use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 
+use super::client::{ClientError, ConnectClient, ConnectConfig, RotationAttempt};
 use super::config::HeartbeatConfig;
 use super::heartbeat::{CoarseNodeSummary, Delivery, HeartbeatError, HeartbeatSender, HeartbeatStateStore, HeartbeatStatus};
 use super::inventory::{
@@ -109,6 +111,14 @@ where
         return Ok(None);
     }
     let sender = HeartbeatSender::new(config.clone())?;
+    let rotation = ConnectClient::new(ConnectConfig {
+        endpoint: &config.endpoint,
+        root_ca_pem: &config.root_ca_pem,
+        timeout: config.schedule.timeout,
+    })
+    .map_err(rotation_failure)?;
+    let identity_store = config.identity_store.clone();
+    let credential_store = config.credential_store.clone();
     let store = HeartbeatStateStore::new(config.state_path.clone());
     let lock = store.try_runtime_lock()?;
     let schedule = config.schedule;
@@ -118,10 +128,46 @@ where
     let task = tokio::spawn(async move {
         let _lock = lock;
         let mut backoff = schedule.initial_backoff;
+        let mut rotation_backoff = schedule.initial_backoff;
+        let mut rotation_retry_at = None;
         loop {
             if task_shutdown.is_cancelled() {
                 break;
             }
+            if rotation_retry_at.is_none_or(|retry_at| Instant::now() >= retry_at) {
+                match cancellable(
+                    &task_shutdown,
+                    rotation.rotate_if_due_once(&identity_store, &credential_store, Utc::now().timestamp()),
+                )
+                .await
+                {
+                    Some(Ok(RotationAttempt::Completed(_))) => {
+                        rotation_backoff = schedule.initial_backoff;
+                        rotation_retry_at = None;
+                    }
+                    Some(Ok(RotationAttempt::ReenrollmentPending)) => {}
+                    Some(Ok(RotationAttempt::Unavailable { retry_after, .. })) => {
+                        let delay = retry_after
+                            .unwrap_or(rotation_backoff)
+                            .clamp(schedule.initial_backoff, schedule.max_backoff);
+                        rotation_backoff = rotation_backoff.saturating_mul(2).min(schedule.max_backoff);
+                        rotation_retry_at = Some(Instant::now() + delay);
+                    }
+                    Some(Err(ClientError::Transport(_))) => {
+                        rotation_retry_at = Some(Instant::now() + rotation_backoff);
+                        rotation_backoff = rotation_backoff.saturating_mul(2).min(schedule.max_backoff);
+                    }
+                    Some(Err(ClientError::AccessRevoked { status, reason })) => {
+                        let _ = status_tx.send(HeartbeatStatus::AuthenticationStopped {
+                            status: status.as_u16(),
+                            reason,
+                        });
+                        return;
+                    }
+                    Some(Err(error)) => return failed(&status_tx, rotation_failure(error)),
+                    None => break,
+                }
+            }
             let pending = match store.prepare(sample(), Utc::now()).await {
                 Ok(pending) => pending,
                 Err(error) => return failed(&status_tx, error),
@@ -344,6 +390,28 @@ fn failed(status: &watch::Sender<HeartbeatStatus>, error: HeartbeatError) {
     });
 }
 
+fn rotation_failure(error: ClientError) -> HeartbeatError {
+    match error {
+        ClientError::Endpoint => HeartbeatError::Endpoint,
+        ClientError::RootCertificate => HeartbeatError::RootCertificate,
+        ClientError::NotRegistered => HeartbeatError::NotRegistered,
+        ClientError::IdentityMissing => HeartbeatError::IdentityMissing,
+        ClientError::CredentialExpired | ClientError::CredentialNotYetValid => HeartbeatError::CredentialExpired,
+        ClientError::IdentityCertificate => HeartbeatError::IdentityCertificate,
+        ClientError::Identity(error) => HeartbeatError::Identity(error),
+        ClientError::IdentityStore(error) => HeartbeatError::IdentityStore(error),
+        ClientError::CredentialStore(error) => HeartbeatError::CredentialStore(error),
+        ClientError::Credential(error) => HeartbeatError::CredentialValidation(error),
+        ClientError::Transport(error) => HeartbeatError::Transport(error),
+        ClientError::ResponseTooLarge => HeartbeatError::ResponseTooLarge,
+        ClientError::PendingRegistration | ClientError::PendingRotation => HeartbeatError::StateConflict,
+        ClientError::AccessRevoked { .. }
+        | ClientError::Rejected { .. }
+        | ClientError::Unavailable { .. }
+        | ClientError::Response => HeartbeatError::Response,
+    }
+}
+
 pub(crate) fn heartbeat_failure_reason(error: &HeartbeatError) -> &'static str {
     use super::registration::CredentialValidationError;
 
@@ -448,6 +516,15 @@ mod tests {
             heartbeat_failure_reason(&HeartbeatError::CredentialExpired),
             "connect_heartbeat_credential_expired"
         );
+        let pending_rotation = rotation_failure(ClientError::PendingRotation);
+        assert!(matches!(pending_rotation, HeartbeatError::StateConflict));
+        assert_eq!(heartbeat_failure_reason(&pending_rotation), "connect_heartbeat_state_conflict");
+        let oversized_rotation_response = rotation_failure(ClientError::ResponseTooLarge);
+        assert!(matches!(oversized_rotation_response, HeartbeatError::ResponseTooLarge));
+        assert_eq!(
+            heartbeat_failure_reason(&oversized_rotation_response),
+            "connect_heartbeat_response_too_large"
+        );
         assert_eq!(
             heartbeat_failure_reason(&HeartbeatError::CredentialValidation(
                 super::super::registration::CredentialValidationError::Identity

rustfs/src/connect/telemetry.rs

@@ -21,11 +21,12 @@ use rustls::pki_types::{CertificateDer, pem::PemObject as _};
 use serde::{Deserialize, Serialize};
 use zeroize::Zeroizing;
 
+use super::client::{ClientError, ConnectClient};
 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};
+use super::registration::{CredentialValidationError, certificate_fingerprint};
 
 const MAX_RESPONSE_BYTES: usize = 64 * 1024;
 
@@ -43,6 +44,13 @@ pub(crate) struct TelemetryTransport {
     config: HeartbeatConfig,
 }
 
+struct AuthenticatedClient {
+    cluster_name: String,
+    cluster_uid: String,
+    credential_fingerprint: String,
+    client: Client,
+}
+
 impl TelemetryTransport {
     pub(crate) fn new(config: HeartbeatConfig) -> Result<Self, TelemetryError> {
         let mut endpoint = Url::parse(&config.endpoint).map_err(|_| TelemetryError::Endpoint)?;
@@ -87,55 +95,80 @@ impl TelemetryTransport {
     }
 
     pub(crate) async fn post<T: Serialize>(&self, collection: &str, value: &T) -> Result<TelemetryDelivery, TelemetryError> {
-        let (cluster_name, cluster_uid, client) = self.authenticated_client().await?;
+        let mut authenticated = self.authenticated_client().await?;
-        let url = self.endpoint.join(&format!("clusters/{cluster_uid}/{collection}"))?;
+        let mut refreshed = false;
-        let response = match client.post(url).json(value).send().await {
+        loop {
-            Ok(response) => response,
+            let url = self
-            Err(error) if error.is_timeout() || error.is_connect() || error.is_request() => {
+                .endpoint
+                .join(&format!("clusters/{}/{collection}", authenticated.cluster_uid))?;
+            let response = match authenticated.client.post(url).json(value).send().await {
+                Ok(response) => response,
+                Err(error) if error.is_timeout() || error.is_connect() || error.is_request() => {
+                    return Ok(TelemetryDelivery::Retry { retry_after: None });
+                }
+                Err(error) => return Err(error.into()),
+            };
+            let status = response.status();
+            if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) && !refreshed {
+                let current = self.authenticated_client().await?;
+                if current.credential_fingerprint != authenticated.credential_fingerprint {
+                    authenticated = current;
+                    refreshed = true;
+                    continue;
+                }
+            }
+            if status == StatusCode::TOO_MANY_REQUESTS {
+                return Ok(TelemetryDelivery::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(TelemetryDelivery::Retry { retry_after: None });
             }
-            Err(error) => return Err(error.into()),
+            if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) {
-        };
+                return Ok(TelemetryDelivery::AuthenticationStopped {
-        let status = response.status();
+                    status: status.as_u16(),
-        if status == StatusCode::TOO_MANY_REQUESTS {
+                    reason: response_reason(response).await,
-            return Ok(TelemetryDelivery::Retry {
+                });
-                retry_after: retry_after(response.headers(), Utc::now(), self.config.schedule.max_backoff),
+            }
+            if status != StatusCode::OK {
+                return Ok(TelemetryDelivery::Rejected {
+                    status: status.as_u16(),
+                    reason: response_reason(response).await,
+                });
+            }
+            return Ok(TelemetryDelivery::Accepted {
+                cluster_name: authenticated.cluster_name,
+                body: bounded_body(response).await?,
             });
         }
-        if status == StatusCode::REQUEST_TIMEOUT || status.is_server_error() {
-            return Ok(TelemetryDelivery::Retry { retry_after: None });
-        }
-        if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) {
-            return Ok(TelemetryDelivery::AuthenticationStopped {
-                status: status.as_u16(),
-                reason: response_reason(response).await,
-            });
-        }
-        if status != StatusCode::OK {
-            return Ok(TelemetryDelivery::Rejected {
-                status: status.as_u16(),
-                reason: response_reason(response).await,
-            });
-        }
-        Ok(TelemetryDelivery::Accepted {
-            cluster_name,
-            body: bounded_body(response).await?,
-        })
     }
 
-    async fn authenticated_client(&self) -> Result<(String, String, Client), TelemetryError> {
+    async fn authenticated_client(&self) -> Result<AuthenticatedClient, TelemetryError> {
-        let _lock = self.config.credential_store.lock().await?;
+        let lock = self.config.credential_store.lock().await?;
-        let credential = self.config.credential_store.load()?.ok_or(TelemetryError::NotRegistered)?;
+        let (credential, identity, _) = ConnectClient::recover_valid_credential_locked(
-        let identity = self.config.identity_store.load()?.ok_or(TelemetryError::IdentityMissing)?;
+            &lock,
-        validate_stored_credential(&credential, &identity, &self.root_store, &self.roots)?;
+            &self.config.identity_store,
+            &self.config.credential_store,
+            &self.root_store,
+            &self.roots,
+        )
+        .map_err(credential_recovery_error)?
+        .ok_or(TelemetryError::NotRegistered)?;
         let now = Utc::now().timestamp();
         if now < credential.not_before_unix || now >= credential.not_after_unix {
             return Err(TelemetryError::CredentialExpired);
         }
         let (organization_uid, cluster_uid) = credential_parent(&credential)?;
         let cluster_name = format!("organizations/{organization_uid}/clusters/{cluster_uid}");
+        let credential_fingerprint = certificate_fingerprint(&credential.certificate)?;
         let client = self.client(&credential, &identity.to_pkcs8_pem()?)?;
-        Ok((cluster_name, cluster_uid.to_owned(), client))
+        Ok(AuthenticatedClient {
+            cluster_name,
+            cluster_uid: cluster_uid.to_owned(),
+            credential_fingerprint,
+            client,
+        })
     }
 
     fn client(&self, credential: &DeviceCredential, key: &Zeroizing<String>) -> Result<Client, TelemetryError> {
@@ -160,6 +193,28 @@ impl TelemetryTransport {
     }
 }
 
+fn credential_recovery_error(error: ClientError) -> TelemetryError {
+    match error {
+        ClientError::Endpoint => TelemetryError::Endpoint,
+        ClientError::RootCertificate => TelemetryError::RootCertificate,
+        ClientError::PendingRegistration | ClientError::PendingRotation => TelemetryError::StateConflict,
+        ClientError::NotRegistered => TelemetryError::NotRegistered,
+        ClientError::IdentityMissing => TelemetryError::IdentityMissing,
+        ClientError::CredentialExpired | ClientError::CredentialNotYetValid => TelemetryError::CredentialExpired,
+        ClientError::IdentityCertificate => TelemetryError::IdentityCertificate,
+        ClientError::Identity(error) => TelemetryError::Identity(error),
+        ClientError::IdentityStore(error) => TelemetryError::IdentityStore(error),
+        ClientError::CredentialStore(error) => TelemetryError::CredentialStore(error),
+        ClientError::Credential(error) => TelemetryError::CredentialValidation(error),
+        ClientError::AccessRevoked { .. }
+        | ClientError::Rejected { .. }
+        | ClientError::Unavailable { .. }
+        | ClientError::ResponseTooLarge
+        | ClientError::Response
+        | ClientError::Transport(_) => TelemetryError::StateConflict,
+    }
+}
+
 fn credential_parent(credential: &DeviceCredential) -> Result<(&str, &str), TelemetryError> {
     let mut parts = credential.name.split('/');
     let valid = parts.next() == Some("organizations");
@@ -248,6 +303,8 @@ pub(crate) enum TelemetryError {
     CredentialName,
     #[error("the stored Connect device certificate is not currently valid")]
     CredentialExpired,
+    #[error("the persisted Connect credential transition is invalid")]
+    StateConflict,
     #[error("Connect telemetry response exceeded 64 KiB")]
     ResponseTooLarge,
     #[error(transparent)]

rustfs/tests/connect_registration.rs

@@ -31,7 +31,12 @@ use rcgen::{
     BasicConstraints, CertificateParams, DistinguishedName, DnType, ExtendedKeyUsagePurpose, IsCa, Issuer, KeyPair,
     KeyUsagePurpose, SanType, SerialNumber,
 };
-use rustfs::connect::{ClientError, ConnectClient, ConnectConfig, CredentialStore, IdentityStore, RegistrationToken, TokenError};
+use rustfs::connect::{
+    ClientError, CoarseNodeSummary, ConnectClient, ConnectConfig, CredentialStore, HeartbeatConfig, HeartbeatError,
+    HeartbeatSchedule, HeartbeatStatus, IdentityStore, RegistrationToken, TokenError, spawn_heartbeat_runtime,
+};
+#[cfg(target_os = "linux")]
+use rustfs::connect::{InventorySchedule, InventorySnapshot, InventoryStatus, spawn_inventory_runtime};
 use rustls::RootCertStore;
 use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, pem::PemObject as _};
 use rustls::server::WebPkiClientVerifier;
@@ -40,7 +45,9 @@ use sha2::{Digest as _, Sha256};
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
 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";
@@ -48,6 +55,8 @@ const DEVICE_UID: &str = "0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92";
 const TOKEN_UID: &str = "0198f4b0-6f00-7b60-9271-7d8e9fa0b1c5";
 const FRESH_TOKEN_UID: &str = "0198f4b0-7f00-7c70-a381-8e9fa0b1c2d6";
 const SECOND_TOKEN_UID: &str = "0198f4b0-8f00-7d80-b491-9fa0b1c2d3e7";
+const UNRELATED_TOKEN_UID: &str = "0198f4b0-9f00-7e90-85a1-afb1c2d3e4f8";
+const INVENTORY_UID: &str = "0198f4b0-af00-7fa0-96b1-bfc1d2e3f509";
 
 struct TestPki {
     root_params: CertificateParams,
@@ -161,6 +170,11 @@ impl TestPki {
 enum Reply {
     Json(StatusCode, Value),
     DelayedClose(Duration),
+    RateLimited(&'static str),
+    UnauthorizedAfterCredential {
+        path: std::path::PathBuf,
+        certificate_serial: String,
+    },
     VerifiedRotation {
         response: Value,
         current_public_key: Vec<u8>,
@@ -172,6 +186,8 @@ enum Reply {
 struct TestServer {
     endpoint: String,
     seen: Arc<Mutex<Vec<Value>>>,
+    paths: Arc<Mutex<Vec<String>>>,
+    client_certificates: Arc<Mutex<Vec<Option<String>>>>,
     task: tokio::task::JoinHandle<()>,
 }
 
@@ -191,7 +207,11 @@ async fn server_with_client_auth(pki: &TestPki, replies: Vec<Reply>, require_cli
     let acceptor = TlsAcceptor::from(Arc::new(pki.server_config(require_client)));
     let replies = Arc::new(Mutex::new(VecDeque::from(replies)));
     let seen = Arc::new(Mutex::new(Vec::new()));
+    let paths = Arc::new(Mutex::new(Vec::new()));
+    let client_certificates = Arc::new(Mutex::new(Vec::new()));
     let captured = seen.clone();
+    let captured_paths = paths.clone();
+    let captured_certificates = client_certificates.clone();
     let task = tokio::spawn(async move {
         loop {
             let Ok((stream, _)) = listener.accept().await else {
@@ -200,18 +220,35 @@ async fn server_with_client_auth(pki: &TestPki, replies: Vec<Reply>, require_cli
             let acceptor = acceptor.clone();
             let replies = replies.clone();
             let seen = captured.clone();
+            let paths = captured_paths.clone();
+            let client_certificates = captured_certificates.clone();
             tokio::spawn(async move {
                 let Ok(stream) = acceptor.accept(stream).await else {
                     return;
                 };
+                let client_certificate = stream
+                    .get_ref()
+                    .1
+                    .peer_certificates()
+                    .and_then(|certificates| certificates.first())
+                    .map(|certificate| certificate_der_fingerprint(certificate.as_ref()));
                 let service = service_fn(move |request: Request<hyper::body::Incoming>| {
                     let replies = replies.clone();
                     let seen = seen.clone();
+                    let paths = paths.clone();
+                    let client_certificates = client_certificates.clone();
+                    let client_certificate = client_certificate.clone();
                     async move {
+                        let path = request.uri().path().to_owned();
                         let body = request.into_body().collect().await.expect("read request body").to_bytes();
                         let value: Value = serde_json::from_slice(&body).expect("request JSON");
-                        seen.lock().expect("seen lock").push(value.clone());
                         let reply = replies.lock().expect("reply lock").pop_front().expect("planned reply");
+                        client_certificates
+                            .lock()
+                            .expect("client certificates lock")
+                            .push(client_certificate);
+                        paths.lock().expect("paths lock").push(path);
+                        seen.lock().expect("seen lock").push(value.clone());
                         match reply {
                             Reply::Json(status, value) => Ok::<_, hyper::Error>(
                                 Response::builder()
@@ -227,6 +264,38 @@ async fn server_with_client_auth(pki: &TestPki, replies: Vec<Reply>, require_cli
                                     .body(Full::new(Bytes::new()))
                                     .expect("response"))
                             }
+                            Reply::RateLimited(retry_after) => Ok(Response::builder()
+                                .status(StatusCode::TOO_MANY_REQUESTS)
+                                .header("content-type", "application/json")
+                                .header("retry-after", retry_after)
+                                .body(Full::new(Bytes::from_static(br#"{"details":[]}"#)))
+                                .expect("response")),
+                            Reply::UnauthorizedAfterCredential {
+                                path,
+                                certificate_serial,
+                            } => {
+                                tokio::time::timeout(Duration::from_secs(3), async {
+                                    loop {
+                                        let changed = fs::read(&path)
+                                            .ok()
+                                            .and_then(|bytes| serde_json::from_slice::<Value>(&bytes).ok())
+                                            .is_some_and(|credential| {
+                                                credential["certificateSerial"].as_str() == Some(certificate_serial.as_str())
+                                            });
+                                        if changed {
+                                            break;
+                                        }
+                                        tokio::time::sleep(Duration::from_millis(5)).await;
+                                    }
+                                })
+                                .await
+                                .expect("rotated credential commit");
+                                Ok(Response::builder()
+                                    .status(StatusCode::UNAUTHORIZED)
+                                    .header("content-type", "application/json")
+                                    .body(Full::new(Bytes::from_static(br#"{"details":[{"reason":"CREDENTIAL_REVOKED"}]}"#)))
+                                    .expect("response"))
+                            }
                             Reply::VerifiedRotation {
                                 response,
                                 current_public_key,
@@ -257,6 +326,8 @@ async fn server_with_client_auth(pki: &TestPki, replies: Vec<Reply>, require_cli
     TestServer {
         endpoint: format!("https://localhost:{}/agent/", address.port()),
         seen,
+        paths,
+        client_certificates,
         task,
     }
 }
@@ -311,10 +382,11 @@ fn certificate_fingerprint(pem: &str) -> String {
         .next()
         .expect("leaf certificate")
         .expect("certificate PEM");
-    Sha256::digest(certificate.as_ref())
+    certificate_der_fingerprint(certificate.as_ref())
-        .iter()
+}
-        .map(|byte| format!("{byte:02x}"))
+
-        .collect()
+fn certificate_der_fingerprint(certificate: &[u8]) -> String {
+    Sha256::digest(certificate).iter().map(|byte| format!("{byte:02x}")).collect()
 }
 
 fn token_document() -> Value {
@@ -355,6 +427,89 @@ fn client(server: &TestServer, pki: &TestPki, timeout: Duration) -> ConnectClien
     .expect("build Connect client")
 }
 
+fn due_runtime_config(
+    temp: &tempfile::TempDir,
+    pki: &TestPki,
+    endpoint: &str,
+) -> (HeartbeatConfig, Value, rustfs::connect::DeviceIdentity) {
+    runtime_config(temp, pki, endpoint, 17)
+}
+
+fn runtime_config(
+    temp: &tempfile::TempDir,
+    pki: &TestPki,
+    endpoint: &str,
+    elapsed_hours: i64,
+) -> (HeartbeatConfig, Value, rustfs::connect::DeviceIdentity) {
+    let (identity_store, credential_store) = stores(temp);
+    let identity = identity_store.load_or_create().expect("create current identity");
+    let now = OffsetDateTime::now_utc().replace_nanosecond(0).expect("whole second");
+    let credential = pki.credential_window(
+        &identity,
+        &format!("urn:rustfs:connect:device:{DEVICE_UID}"),
+        0x20,
+        now - time::Duration::hours(elapsed_hours),
+        now + time::Duration::hours(24 - elapsed_hours),
+    );
+    let not_before =
+        OffsetDateTime::parse(credential["notBefore"].as_str().expect("notBefore"), &Rfc3339).expect("parse notBefore");
+    let not_after = OffsetDateTime::parse(credential["notAfter"].as_str().expect("notAfter"), &Rfc3339).expect("parse notAfter");
+    assert_eq!(not_after - not_before, time::Duration::hours(24));
+    fs::create_dir_all(temp.path().join("credential")).expect("credential directory");
+    write_stored_credential(&temp.path().join("credential/device.crt.json"), &credential);
+    let next = identity_store.load_or_create_next().expect("create next identity");
+    let mut config = HeartbeatConfig::new(
+        endpoint,
+        pki.root_pem.as_bytes(),
+        identity_store,
+        credential_store,
+        temp.path().join("heartbeat/state.json"),
+    );
+    config.schedule = HeartbeatSchedule {
+        cadence: Duration::from_millis(100),
+        jitter: Duration::ZERO,
+        timeout: Duration::from_millis(250),
+        initial_backoff: Duration::from_millis(20),
+        max_backoff: Duration::from_secs(2),
+    };
+    (config, credential, next)
+}
+
+fn heartbeat_response(server_time: &str) -> Value {
+    json!({
+        "serverTime": server_time,
+        "acceptedVersion": "v1",
+        "capabilityHints": [],
+    })
+}
+
+async fn wait_for_requests(server: &TestServer, count: usize) {
+    tokio::time::timeout(Duration::from_secs(3), async {
+        while server.paths.lock().expect("paths lock").len() < count {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("planned requests");
+}
+
+async fn wait_for_heartbeat_status(
+    status: &mut watch::Receiver<HeartbeatStatus>,
+    predicate: impl Fn(&HeartbeatStatus) -> bool,
+) -> HeartbeatStatus {
+    tokio::time::timeout(Duration::from_secs(3), async {
+        loop {
+            let current = status.borrow_and_update().clone();
+            if predicate(&current) {
+                return current;
+            }
+            status.changed().await.expect("heartbeat status channel");
+        }
+    })
+    .await
+    .expect("heartbeat status")
+}
+
 fn rotation_response(pki: &TestPki, identity: &rustfs::connect::DeviceIdentity, serial: u8) -> (Value, Value) {
     let stored = pki.credential(identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), serial);
     let mut wire = stored.clone();
@@ -577,7 +732,7 @@ async fn concurrent_rotation_retries_converge_and_promote_the_next_key() {
     assert!(matches!(second, Err(ClientError::Unavailable { .. })));
     let (request_id, certificate_request) = {
         let seen = retries.seen.lock().expect("seen lock");
-        assert!(seen.len() >= 3, "bounded retries must reach the server");
+        assert_eq!(seen.len(), 6, "each public rotation call makes three attempts");
         for request in &seen[1..] {
             assert_eq!(request["requestId"], seen[0]["requestId"]);
             assert_eq!(request["certificateRequest"], seen[0]["certificateRequest"]);
@@ -645,6 +800,539 @@ async fn concurrent_rotation_retries_converge_and_promote_the_next_key() {
     assert!(!temp.path().join("credential/rotation.pending.json").exists());
 }
 
+#[tokio::test]
+async fn heartbeat_runtime_retries_rotation_and_reloads_the_rotated_credential() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let pki = TestPki::new();
+    let (mut config, current, next) = due_runtime_config(&temp, &pki, "https://localhost/agent/");
+    let (rotated, rotated_stored) = rotation_response(&pki, &next, 0x21);
+    let server = server_with_client_auth(
+        &pki,
+        vec![
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+            Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:03Z")),
+            Reply::VerifiedRotation {
+                response: rotated,
+                current_public_key: config
+                    .identity_store
+                    .load()
+                    .expect("load current identity")
+                    .expect("current identity")
+                    .public_key_der(),
+                current_certificate_fingerprint: certificate_fingerprint(
+                    current["certificate"].as_str().expect("current certificate"),
+                ),
+                device_name: current["name"].as_str().expect("device name").to_owned(),
+            },
+            Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:04Z")),
+        ],
+        true,
+    )
+    .await;
+    config.endpoint = server.endpoint.clone();
+    let shutdown = CancellationToken::new();
+    let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary"))
+        .expect("start heartbeat runtime")
+        .expect("configured runtime");
+
+    wait_for_requests(&server, 4).await;
+    runtime.shutdown().await;
+
+    let rotation_path = format!("/agent/clusterDevices/{DEVICE_UID}:rotateCredential");
+    let heartbeat_path = format!("/agent/clusters/{CLUSTER_UID}/heartbeats");
+    assert_eq!(
+        &server.paths.lock().expect("paths lock")[..4],
+        [rotation_path.clone(), heartbeat_path.clone(), rotation_path, heartbeat_path,]
+    );
+    let current_fingerprint = certificate_fingerprint(current["certificate"].as_str().expect("current certificate"));
+    let rotated_fingerprint = certificate_fingerprint(rotated_stored["certificate"].as_str().expect("rotated certificate"));
+    let certificates = server.client_certificates.lock().expect("client certificates lock");
+    assert_eq!(certificates[1].as_deref(), Some(current_fingerprint.as_str()));
+    assert_eq!(certificates[3].as_deref(), Some(rotated_fingerprint.as_str()));
+    drop(certificates);
+    let stored: Value =
+        serde_json::from_slice(&fs::read(temp.path().join("credential/device.crt.json")).expect("stored rotated credential"))
+            .expect("stored credential JSON");
+    assert_eq!(stored["name"], current["name"]);
+    assert_eq!(stored["uid"], current["uid"]);
+    assert_eq!(stored["certificateSerial"], rotated_stored["certificateSerial"]);
+}
+
+#[tokio::test]
+async fn heartbeat_runtime_respects_rotation_retry_after_without_blocking_heartbeats() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let pki = TestPki::new();
+    let server = server_with_client_auth(
+        &pki,
+        std::iter::once(Reply::RateLimited("1"))
+            .chain((0..8).map(|_| Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:03Z"))))
+            .collect(),
+        true,
+    )
+    .await;
+    let (config, _, _) = due_runtime_config(&temp, &pki, &server.endpoint);
+    let shutdown = CancellationToken::new();
+    let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary"))
+        .expect("start heartbeat runtime")
+        .expect("configured runtime");
+    let mut status = runtime.status();
+
+    wait_for_heartbeat_status(&mut status, |status| matches!(status, HeartbeatStatus::Online { .. })).await;
+    tokio::time::sleep(Duration::from_millis(350)).await;
+    runtime.shutdown().await;
+
+    let paths = server.paths.lock().expect("paths lock");
+    assert_eq!(paths.iter().filter(|path| path.ends_with(":rotateCredential")).count(), 1);
+    assert!(paths.iter().filter(|path| path.ends_with("/heartbeats")).count() >= 3);
+}
+
+#[tokio::test]
+async fn heartbeat_runtime_skips_only_valid_pending_reenrollment() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let pki = TestPki::new();
+    let (mut config, _, next) = due_runtime_config(&temp, &pki, "https://localhost/agent/");
+    let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 0x24);
+    let server = server_with_client_auth(
+        &pki,
+        vec![
+            Reply::Json(StatusCode::CREATED, enrolled),
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+            Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:03Z")),
+        ],
+        true,
+    )
+    .await;
+    config.endpoint = server.endpoint.clone();
+    let validation_config = config.clone();
+    client(&server, &pki, Duration::from_millis(250))
+        .reenroll(&config.identity_store, &config.credential_store, &token_with_uid(FRESH_TOKEN_UID))
+        .await
+        .expect("complete prior reenrollment");
+    assert!(matches!(
+        client(&server, &pki, Duration::from_millis(250))
+            .reenroll(&config.identity_store, &config.credential_store, &token_with_uid(SECOND_TOKEN_UID))
+            .await,
+        Err(ClientError::Unavailable { .. })
+    ));
+    let pending_path = temp.path().join("credential/registration.pending.json");
+    let next_path = temp.path().join("identity/device.key.next");
+    let credential_path = temp.path().join("credential/device.crt.json");
+    let pending = fs::read(&pending_path).expect("pending reenrollment");
+    let next = fs::read(&next_path).expect("pending reenrollment key");
+    let credential = fs::read(&credential_path).expect("current credential");
+    let shutdown = CancellationToken::new();
+    let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary"))
+        .expect("start heartbeat runtime")
+        .expect("configured runtime");
+    let mut status = runtime.status();
+
+    assert!(matches!(
+        wait_for_heartbeat_status(&mut status, |status| matches!(status, HeartbeatStatus::Online { .. })).await,
+        HeartbeatStatus::Online { .. }
+    ));
+    runtime.shutdown().await;
+
+    assert_eq!(fs::read(&pending_path).expect("preserved pending reenrollment"), pending);
+    assert_eq!(fs::read(&next_path).expect("preserved pending reenrollment key"), next);
+    assert_eq!(fs::read(&credential_path).expect("preserved current credential"), credential);
+    let paths = server.paths.lock().expect("paths lock");
+    assert_eq!(
+        paths
+            .iter()
+            .filter(|path| path.ends_with("registrationTokens:exchange"))
+            .count(),
+        4
+    );
+    assert_eq!(paths.iter().filter(|path| path.ends_with(":rotateCredential")).count(), 0);
+    assert_eq!(paths.iter().filter(|path| path.ends_with("/heartbeats")).count(), 1);
+    drop(paths);
+
+    for (field, value) in [
+        ("requestId", "not-a-request-id"),
+        ("tokenUid", ""),
+        ("tokenUid", "not-a-token-uid"),
+        ("tokenUid", FRESH_TOKEN_UID),
+        ("tokenUid", UNRELATED_TOKEN_UID),
+    ] {
+        let mut corrupted_pending: Value = serde_json::from_slice(&pending).expect("pending reenrollment JSON");
+        corrupted_pending[field] = json!(value);
+        let corrupted_pending = serde_json::to_vec(&corrupted_pending).expect("corrupted pending reenrollment JSON");
+        fs::write(&pending_path, &corrupted_pending).expect("corrupt pending reenrollment");
+        set_owner_only(&pending_path);
+        let corrupted_runtime = spawn_heartbeat_runtime(Some(validation_config.clone()), &shutdown, || {
+            CoarseNodeSummary::new(1, 1, 0).expect("node summary")
+        })
+        .expect("start heartbeat runtime with corrupt pending state")
+        .expect("configured runtime");
+        let mut corrupted_status = corrupted_runtime.status();
+        assert_eq!(
+            wait_for_heartbeat_status(&mut corrupted_status, |status| matches!(status, HeartbeatStatus::Failed { .. })).await,
+            HeartbeatStatus::Failed {
+                reason: HeartbeatError::StateConflict.to_string(),
+            }
+        );
+        corrupted_runtime.shutdown().await;
+        assert_eq!(fs::read(&pending_path).expect("preserved corrupt pending state"), corrupted_pending);
+    }
+    assert_eq!(fs::read(&next_path).expect("preserved pending reenrollment key"), next);
+    assert_eq!(fs::read(&credential_path).expect("preserved current credential"), credential);
+    let paths = server.paths.lock().expect("paths lock");
+    assert_eq!(paths.iter().filter(|path| path.ends_with(":rotateCredential")).count(), 0);
+    assert_eq!(paths.iter().filter(|path| path.ends_with("/heartbeats")).count(), 1);
+}
+
+#[tokio::test]
+async fn heartbeat_retries_with_the_new_credential_after_concurrent_rotation() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let pki = TestPki::new();
+    let (mut config, current, next) = runtime_config(&temp, &pki, "https://localhost/agent/", 1);
+    let (rotated, rotated_stored) = rotation_response(&pki, &next, 0x22);
+    let server = server_with_client_auth(
+        &pki,
+        vec![
+            Reply::UnauthorizedAfterCredential {
+                path: temp.path().join("credential/device.crt.json"),
+                certificate_serial: rotated_stored["certificateSerial"]
+                    .as_str()
+                    .expect("rotated serial")
+                    .to_owned(),
+            },
+            Reply::VerifiedRotation {
+                response: rotated,
+                current_public_key: config
+                    .identity_store
+                    .load()
+                    .expect("load current identity")
+                    .expect("current identity")
+                    .public_key_der(),
+                current_certificate_fingerprint: certificate_fingerprint(
+                    current["certificate"].as_str().expect("current certificate"),
+                ),
+                device_name: current["name"].as_str().expect("device name").to_owned(),
+            },
+            Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:03Z")),
+        ],
+        true,
+    )
+    .await;
+    config.endpoint = server.endpoint.clone();
+    let identity_store = config.identity_store.clone();
+    let credential_store = config.credential_store.clone();
+    let shutdown = CancellationToken::new();
+    let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary"))
+        .expect("start heartbeat runtime")
+        .expect("configured runtime");
+    let mut status = runtime.status();
+
+    wait_for_requests(&server, 1).await;
+    let due = OffsetDateTime::parse(current["notAfter"].as_str().expect("notAfter"), &Rfc3339)
+        .expect("parse notAfter")
+        .unix_timestamp()
+        - 8 * 60 * 60;
+    client(&server, &pki, Duration::from_millis(250))
+        .rotate_if_due(&identity_store, &credential_store, due)
+        .await
+        .expect("concurrent rotation")
+        .expect("rotation due");
+    assert!(matches!(
+        wait_for_heartbeat_status(&mut status, |status| matches!(status, HeartbeatStatus::Online { .. })).await,
+        HeartbeatStatus::Online { .. }
+    ));
+    runtime.shutdown().await;
+
+    let heartbeat_path = format!("/agent/clusters/{CLUSTER_UID}/heartbeats");
+    assert_eq!(
+        server.paths.lock().expect("paths lock").as_slice(),
+        [
+            heartbeat_path.clone(),
+            format!("/agent/clusterDevices/{DEVICE_UID}:rotateCredential"),
+            heartbeat_path,
+        ]
+    );
+    let current_fingerprint = certificate_fingerprint(current["certificate"].as_str().expect("current certificate"));
+    let rotated_fingerprint = certificate_fingerprint(rotated_stored["certificate"].as_str().expect("rotated certificate"));
+    let certificates = server.client_certificates.lock().expect("client certificates lock");
+    assert_eq!(certificates[0].as_deref(), Some(current_fingerprint.as_str()));
+    assert_eq!(certificates[2].as_deref(), Some(rotated_fingerprint.as_str()));
+}
+
+#[cfg(target_os = "linux")]
+#[tokio::test]
+async fn inventory_retries_with_the_new_credential_after_concurrent_rotation() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let pki = TestPki::new();
+    let (mut config, current, next) = runtime_config(&temp, &pki, "https://localhost/agent/", 1);
+    let (rotated, rotated_stored) = rotation_response(&pki, &next, 0x23);
+    let server = server_with_client_auth(
+        &pki,
+        vec![
+            Reply::UnauthorizedAfterCredential {
+                path: temp.path().join("credential/device.crt.json"),
+                certificate_serial: rotated_stored["certificateSerial"]
+                    .as_str()
+                    .expect("rotated serial")
+                    .to_owned(),
+            },
+            Reply::VerifiedRotation {
+                response: rotated,
+                current_public_key: config
+                    .identity_store
+                    .load()
+                    .expect("load current identity")
+                    .expect("current identity")
+                    .public_key_der(),
+                current_certificate_fingerprint: certificate_fingerprint(
+                    current["certificate"].as_str().expect("current certificate"),
+                ),
+                device_name: current["name"].as_str().expect("device name").to_owned(),
+            },
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+        ],
+        true,
+    )
+    .await;
+    config.endpoint = server.endpoint.clone();
+    let identity_store = config.identity_store.clone();
+    let credential_store = config.credential_store.clone();
+    let snapshot = InventorySnapshot::current(1, 1, 100, 50, []).expect("inventory snapshot");
+    let shutdown = CancellationToken::new();
+    let runtime = spawn_inventory_runtime(
+        Some(config),
+        InventorySchedule {
+            cadence: Duration::from_millis(100),
+            jitter: Duration::ZERO,
+        },
+        &shutdown,
+        move || std::future::ready(Ok(snapshot.clone())),
+    )
+    .expect("start inventory runtime")
+    .expect("configured runtime");
+    let mut status = runtime.status();
+
+    wait_for_requests(&server, 1).await;
+    let due = OffsetDateTime::parse(current["notAfter"].as_str().expect("notAfter"), &Rfc3339)
+        .expect("parse notAfter")
+        .unix_timestamp()
+        - 8 * 60 * 60;
+    client(&server, &pki, Duration::from_millis(250))
+        .rotate_if_due(&identity_store, &credential_store, due)
+        .await
+        .expect("concurrent rotation")
+        .expect("rotation due");
+    assert!(matches!(
+        tokio::time::timeout(Duration::from_secs(3), async {
+            loop {
+                let current = status.borrow_and_update().clone();
+                if matches!(current, InventoryStatus::BackingOff { .. }) {
+                    break current;
+                }
+                status.changed().await.expect("inventory status channel");
+            }
+        })
+        .await
+        .expect("inventory retry status"),
+        InventoryStatus::BackingOff { .. }
+    ));
+    runtime.shutdown().await;
+
+    let inventory_path = format!("/agent/clusters/{CLUSTER_UID}/inventorySnapshots");
+    assert_eq!(
+        server.paths.lock().expect("paths lock").as_slice(),
+        [
+            inventory_path.clone(),
+            format!("/agent/clusterDevices/{DEVICE_UID}:rotateCredential"),
+            inventory_path,
+        ]
+    );
+    let current_fingerprint = certificate_fingerprint(current["certificate"].as_str().expect("current certificate"));
+    let rotated_fingerprint = certificate_fingerprint(rotated_stored["certificate"].as_str().expect("rotated certificate"));
+    let certificates = server.client_certificates.lock().expect("client certificates lock");
+    assert_eq!(certificates[0].as_deref(), Some(current_fingerprint.as_str()));
+    assert_eq!(certificates[2].as_deref(), Some(rotated_fingerprint.as_str()));
+}
+
+#[cfg(target_os = "linux")]
+#[tokio::test]
+async fn inventory_first_recovers_a_saved_reenrollment_before_telemetry() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let pki = TestPki::new();
+    let failed = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await;
+    let (mut config, _, next) = runtime_config(&temp, &pki, &failed.endpoint, 1);
+    assert!(matches!(
+        client(&failed, &pki, Duration::from_millis(250))
+            .reenroll(&config.identity_store, &config.credential_store, &token_with_uid(SECOND_TOKEN_UID))
+            .await,
+        Err(ClientError::Unavailable { .. })
+    ));
+    let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 0x25);
+    write_stored_credential(&temp.path().join("credential/device.crt.json"), &enrolled);
+    assert!(temp.path().join("credential/registration.pending.json").exists());
+    assert!(temp.path().join("identity/device.key.next").exists());
+    assert_ne!(
+        config
+            .identity_store
+            .load()
+            .expect("load pre-recovery identity")
+            .expect("pre-recovery identity")
+            .public_key_der(),
+        next.public_key_der()
+    );
+
+    let snapshot = InventorySnapshot::current(1, 1, 100, 50, []).expect("inventory snapshot");
+    let content_hash = snapshot.content_hash().expect("inventory content hash");
+    let telemetry = server_with_client_auth(
+        &pki,
+        vec![
+            Reply::Json(
+                StatusCode::OK,
+                json!({
+                    "name": format!("organizations/{ORGANIZATION_UID}/clusters/{CLUSTER_UID}/inventorySnapshots/{INVENTORY_UID}"),
+                    "uid": INVENTORY_UID,
+                    "contentHash": content_hash,
+                    "receivedAt": "2026-08-25T01:02:03Z",
+                }),
+            ),
+            Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:04Z")),
+        ],
+        true,
+    )
+    .await;
+    config.endpoint = telemetry.endpoint.clone();
+    config.schedule.cadence = Duration::from_secs(60);
+    let heartbeat_config = config.clone();
+    let shutdown = CancellationToken::new();
+    let inventory_snapshot = snapshot.clone();
+    let inventory = spawn_inventory_runtime(
+        Some(config),
+        InventorySchedule {
+            cadence: Duration::from_secs(60),
+            jitter: Duration::ZERO,
+        },
+        &shutdown,
+        move || std::future::ready(Ok(inventory_snapshot.clone())),
+    )
+    .expect("start inventory runtime")
+    .expect("configured inventory runtime");
+    let mut inventory_status = inventory.status();
+    assert!(matches!(
+        tokio::time::timeout(Duration::from_secs(3), async {
+            loop {
+                let current = inventory_status.borrow_and_update().clone();
+                if matches!(current, InventoryStatus::Online { .. }) {
+                    break current;
+                }
+                inventory_status.changed().await.expect("inventory status channel");
+            }
+        })
+        .await
+        .expect("inventory online status"),
+        InventoryStatus::Online { .. }
+    ));
+
+    let pending_path = temp.path().join("credential/registration.pending.json");
+    assert!(!pending_path.exists());
+    assert!(!temp.path().join("identity/device.key.next").exists());
+    assert_eq!(
+        heartbeat_config
+            .identity_store
+            .load()
+            .expect("load recovered identity")
+            .expect("recovered identity")
+            .public_key_der(),
+        next.public_key_der()
+    );
+    let completed: Value = serde_json::from_slice(
+        &fs::read(temp.path().join("credential/registration.completed.json")).expect("completed reenrollment receipt"),
+    )
+    .expect("completed reenrollment JSON");
+    assert_eq!(completed["tokenUid"], SECOND_TOKEN_UID);
+
+    let heartbeat = spawn_heartbeat_runtime(Some(heartbeat_config), &shutdown, || {
+        CoarseNodeSummary::new(1, 1, 0).expect("node summary")
+    })
+    .expect("start heartbeat runtime")
+    .expect("configured heartbeat runtime");
+    let mut heartbeat_status = heartbeat.status();
+    assert!(matches!(
+        wait_for_heartbeat_status(&mut heartbeat_status, |status| matches!(status, HeartbeatStatus::Online { .. })).await,
+        HeartbeatStatus::Online { .. }
+    ));
+    heartbeat.shutdown().await;
+    inventory.shutdown().await;
+
+    assert_eq!(failed.paths.lock().expect("registration paths").len(), 3);
+    assert_eq!(
+        telemetry.paths.lock().expect("telemetry paths").as_slice(),
+        [
+            format!("/agent/clusters/{CLUSTER_UID}/inventorySnapshots"),
+            format!("/agent/clusters/{CLUSTER_UID}/heartbeats"),
+        ]
+    );
+    let enrolled_fingerprint = certificate_fingerprint(enrolled["certificate"].as_str().expect("enrolled certificate"));
+    assert_eq!(
+        telemetry.client_certificates.lock().expect("client certificates").as_slice(),
+        [Some(enrolled_fingerprint.clone()), Some(enrolled_fingerprint)]
+    );
+}
+
+#[tokio::test]
+async fn heartbeat_runtime_cancels_an_in_flight_rotation() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let pki = TestPki::new();
+    let server = server_with_client_auth(&pki, vec![Reply::DelayedClose(Duration::from_secs(5))], true).await;
+    let (config, _, _) = due_runtime_config(&temp, &pki, &server.endpoint);
+    let shutdown = CancellationToken::new();
+    let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary"))
+        .expect("start heartbeat runtime")
+        .expect("configured runtime");
+
+    wait_for_requests(&server, 1).await;
+    tokio::time::timeout(Duration::from_millis(250), runtime.shutdown())
+        .await
+        .expect("rotation cancellation");
+    assert_eq!(
+        server.paths.lock().expect("paths lock").as_slice(),
+        [format!("/agent/clusterDevices/{DEVICE_UID}:rotateCredential")]
+    );
+}
+
+#[tokio::test]
+async fn heartbeat_runtime_stops_when_rotation_reports_revocation() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let pki = TestPki::new();
+    let server = server_with_client_auth(
+        &pki,
+        vec![Reply::Json(
+            StatusCode::UNAUTHORIZED,
+            json!({"details": [{"reason": "DEVICE_REVOKED"}]}),
+        )],
+        true,
+    )
+    .await;
+    let (config, _, _) = due_runtime_config(&temp, &pki, &server.endpoint);
+    let shutdown = CancellationToken::new();
+    let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary"))
+        .expect("start heartbeat runtime")
+        .expect("configured runtime");
+    let mut status = runtime.status();
+
+    assert_eq!(
+        wait_for_heartbeat_status(&mut status, |status| matches!(status, HeartbeatStatus::AuthenticationStopped { .. })).await,
+        HeartbeatStatus::AuthenticationStopped {
+            status: 401,
+            reason: Some("DEVICE_REVOKED".to_owned()),
+        }
+    );
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    assert_eq!(
+        server.paths.lock().expect("paths lock").as_slice(),
+        [format!("/agent/clusterDevices/{DEVICE_UID}:rotateCredential")]
+    );
+    runtime.shutdown().await;
+}
+
 #[tokio::test]
 async fn rotation_commit_recovers_after_each_durable_step() {
     let temp = tempfile::tempdir().expect("temp dir");
@@ -665,6 +1353,13 @@ async fn rotation_commit_recovers_after_each_durable_step() {
             .await,
         Err(ClientError::Unavailable { .. })
     ));
+    let failed_seen = failed.seen.lock().expect("seen lock");
+    assert_eq!(failed_seen.len(), 3, "public rotation keeps its bounded retry contract");
+    for request in &failed_seen[1..] {
+        assert_eq!(request["requestId"], failed_seen[0]["requestId"]);
+        assert_eq!(request["certificateRequest"], failed_seen[0]["certificateRequest"]);
+    }
+    drop(failed_seen);
 
     let pending_path = temp.path().join("credential/rotation.pending.json");
     let pending = fs::read(&pending_path).expect("read pending state");
@@ -754,19 +1449,31 @@ async fn pending_reenrollment_blocks_rotation_and_resumes_original_exchange() {
 async fn reenrollment_commit_recovers_after_each_durable_step() {
     let temp = tempfile::tempdir().expect("temp dir");
     let (identity_store, credential_store) = stores(&temp);
-    let current = identity_store.load_or_create().expect("create identity");
+    let initial = identity_store.load_or_create().expect("create identity");
     let pki = TestPki::new();
-    let issued = pki.credential(&current, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 13);
+    let issued = pki.credential(&initial, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 13);
     let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await;
     client(&registration, &pki, Duration::from_secs(2))
         .register(&identity_store, &credential_store, &token())
         .await
         .expect("register");
 
+    let first_next = identity_store
+        .load_or_create_next()
+        .expect("create first reenrollment identity");
+    let first_enrolled = pki.credential(&first_next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 14);
+    let first_reenrollment = server(&pki, vec![Reply::Json(StatusCode::CREATED, first_enrolled)]).await;
+    client(&first_reenrollment, &pki, Duration::from_secs(2))
+        .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID))
+        .await
+        .expect("complete prior reenrollment");
+    let completed_path = temp.path().join("credential/registration.completed.json");
+    let previous_completed = fs::read(&completed_path).expect("read prior completed receipt");
+
     let failed = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await;
     assert!(matches!(
         client(&failed, &pki, Duration::from_secs(2))
-            .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID))
+            .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID))
             .await,
         Err(ClientError::Unavailable { .. })
     ));
@@ -775,16 +1482,41 @@ async fn reenrollment_commit_recovers_after_each_durable_step() {
     let pending = fs::read(&pending_path).expect("read pending reenrollment");
     let next_der = fs::read(temp.path().join("identity/device.key.next")).expect("read next key");
     let next = rustfs::connect::DeviceIdentity::from_pkcs8_der(&next_der).expect("parse next key");
-    let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 14);
+    let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 15);
     write_stored_credential(&temp.path().join("credential/device.crt.json"), &enrolled);
 
     let idle = server(&pki, vec![]).await;
     let idle_client = client(&idle, &pki, Duration::from_secs(2));
+    for token_uid in [UNRELATED_TOKEN_UID, FRESH_TOKEN_UID] {
+        let mut document: Value = serde_json::from_slice(&pending).expect("pending reenrollment JSON");
+        document["tokenUid"] = json!(token_uid);
+        let tampered_pending = serde_json::to_vec(&document).expect("tampered pending reenrollment JSON");
+        fs::write(&pending_path, &tampered_pending).expect("tamper pending after credential save");
+        set_owner_only(&pending_path);
+        assert!(matches!(
+            idle_client
+                .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID))
+                .await,
+            Err(ClientError::PendingRegistration)
+        ));
+        assert_eq!(fs::read(&pending_path).expect("preserved tampered pending"), tampered_pending);
+        assert_eq!(fs::read(&completed_path).expect("preserved prior completed receipt"), previous_completed);
+        assert_eq!(
+            identity_store
+                .load()
+                .expect("load current key")
+                .expect("current key")
+                .public_key_der(),
+            first_next.public_key_der()
+        );
+    }
+    fs::write(&pending_path, &pending).expect("restore bound pending after credential save");
+    set_owner_only(&pending_path);
     let recovered = idle_client
-        .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID))
+        .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID))
         .await
         .expect("recover after reenrollment credential save");
-    assert_eq!(recovered.certificate_serial, "0e".repeat(16));
+    assert_eq!(recovered.certificate_serial, "0f".repeat(16));
     assert_eq!(
         identity_store
             .load()
@@ -794,26 +1526,50 @@ async fn reenrollment_commit_recovers_after_each_durable_step() {
         next.public_key_der()
     );
 
-    fs::remove_file(temp.path().join("credential/registration.completed.json")).expect("remove completed receipt");
+    fs::write(&completed_path, &previous_completed).expect("restore prior completed receipt after key commit");
-    fs::write(&pending_path, pending).expect("restore pending after key commit");
+    set_owner_only(&completed_path);
+    for token_uid in [UNRELATED_TOKEN_UID, FRESH_TOKEN_UID] {
+        let mut document: Value = serde_json::from_slice(&pending).expect("pending reenrollment JSON");
+        document["tokenUid"] = json!(token_uid);
+        let tampered_pending = serde_json::to_vec(&document).expect("tampered pending reenrollment JSON");
+        fs::write(&pending_path, &tampered_pending).expect("tamper pending after key commit");
+        set_owner_only(&pending_path);
+        assert!(matches!(
+            idle_client
+                .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID))
+                .await,
+            Err(ClientError::PendingRegistration)
+        ));
+        assert_eq!(fs::read(&pending_path).expect("preserved tampered pending"), tampered_pending);
+        assert_eq!(fs::read(&completed_path).expect("preserved prior completed receipt"), previous_completed);
+        assert_eq!(
+            identity_store
+                .load()
+                .expect("load current key")
+                .expect("current key")
+                .public_key_der(),
+            next.public_key_der()
+        );
+    }
+    fs::write(&pending_path, pending).expect("restore bound pending after key commit");
     set_owner_only(&pending_path);
     let recovered = idle_client
-        .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID))
+        .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID))
         .await
         .expect("recover after reenrollment key commit");
-    assert_eq!(recovered.certificate_serial, "0e".repeat(16));
+    assert_eq!(recovered.certificate_serial, "0f".repeat(16));
     assert!(!pending_path.exists());
     let recovered = idle_client
-        .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID))
+        .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID))
         .await
         .expect("completed reenrollment is idempotent after pending cleanup");
-    assert_eq!(recovered.certificate_serial, "0e".repeat(16));
+    assert_eq!(recovered.certificate_serial, "0f".repeat(16));
     assert!(idle.seen.lock().expect("seen lock").is_empty());
 
     let different = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await;
     assert!(matches!(
         client(&different, &pki, Duration::from_secs(2))
-            .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID))
+            .reenroll(&identity_store, &credential_store, &token_with_uid(TOKEN_UID))
             .await,
         Err(ClientError::Unavailable { .. })
     ));