From b3a1bfc4ce95300997cfb09e99b4bfcc36a91439 Mon Sep 17 00:00:00 2001 From: Deepak Maram Date: Thu, 9 Jul 2026 19:47:27 -0400 Subject: [PATCH 1/4] [Guardian] Improve S3 log write robustness --- crates/hashi-guardian/src/enclave.rs | 75 +++--- crates/hashi-guardian/src/lib.rs | 2 - crates/hashi-guardian/src/main.rs | 30 +-- crates/hashi-guardian/src/rpc.rs | 9 +- crates/hashi-guardian/src/s3_client.rs | 239 +++++++++++++++++- .../src/withdraw_mode/heartbeat.rs | 176 ++----------- .../hashi-guardian/src/withdraw_mode/mod.rs | 5 +- .../{standard.rs => standard_withdrawal.rs} | 45 ++-- crates/hashi-types/src/guardian/limiter.rs | 28 +- 9 files changed, 318 insertions(+), 291 deletions(-) rename crates/hashi-guardian/src/withdraw_mode/{standard.rs => standard_withdrawal.rs} (91%) diff --git a/crates/hashi-guardian/src/enclave.rs b/crates/hashi-guardian/src/enclave.rs index 8124b13cea..55b2c1ad0d 100644 --- a/crates/hashi-guardian/src/enclave.rs +++ b/crates/hashi-guardian/src/enclave.rs @@ -23,6 +23,7 @@ use std::sync::Arc; use std::sync::OnceLock; use std::sync::RwLock; use std::time::Duration; +use tokio::sync::OwnedMutexGuard; use tracing::info; use crate::s3_client::GuardianS3Client; @@ -283,19 +284,19 @@ impl EnclaveState { .map_err(|_| InvalidInputs("rate_limiter already initialized".into())) } - /// Acquire exclusive access to the limiter, consume tokens, and return a guard. - /// The guard holds the mutex lock — no other withdrawal can start until it is - /// committed or dropped (which reverts). /// Timeout for acquiring the limiter lock. If a withdrawal is in progress and /// takes longer than this, we bail rather than queue up requests indefinitely. const LIMITER_LOCK_TIMEOUT: Duration = Duration::from_secs(10); + /// Acquire exclusive access to the limiter, consume tokens, and return a guard. + /// The guard is held through signing and durable logging so no other withdrawal + /// can start until this one is durably logged or the enclave aborts. pub async fn consume_from_limiter( &self, seq: u64, timestamp: u64, amount_sats: u64, - ) -> GuardianResult { + ) -> GuardianResult> { let rate_limiter = self .rate_limiter .get() @@ -307,7 +308,7 @@ impl EnclaveState { .await .map_err(|_| InvalidInputs("timed out waiting for rate limiter lock".into()))?; guard.consume(seq, timestamp, amount_sats)?; - Ok(LimiterGuard::new(guard)) + Ok(guard) } pub async fn limiter_state(&self) -> Option { @@ -321,39 +322,6 @@ impl EnclaveState { } } -/// RAII guard that holds the limiter mutex via an owned guard, returned by -/// [`EnclaveState::consume_from_limiter`]. Reverts on drop unless committed. -pub struct LimiterGuard { - guard: tokio::sync::OwnedMutexGuard, - committed: bool, -} - -impl LimiterGuard { - pub(crate) fn new(guard: tokio::sync::OwnedMutexGuard) -> Self { - Self { - guard, - committed: false, - } - } - - /// Mark this withdrawal as successful. Prevents revert on drop. - pub fn commit(mut self) { - self.committed = true; - } - - pub fn state(&self) -> &LimiterState { - self.guard.state() - } -} - -impl Drop for LimiterGuard { - fn drop(&mut self) { - if !self.committed { - self.guard.revert(); - } - } -} - impl Enclave { // ======================================================================== // Construction & Initialization Status @@ -526,29 +494,48 @@ impl Enclave { self.config.s3_logger()?.write_log_record(log).await } + async fn write_log_or_abort(&self, message: LogMessage) -> GuardianResult<()> { + let log = LogRecord::new( + self.s3_session_id(), + message, + &self.config.eph_keys.signing_keys, + ); + + self.config + .s3_logger()? + .write_log_record_or_abort(log) + .await + } + + /// Only init skips grace-period retries, providing quick fail-stop for basic + /// S3 write/access issues. The incomplete enclave cannot serve and will restart, + /// so S3 being ahead after a lost acknowledgement is acceptable. pub async fn log_init(&self, msg: InitLogMessage) -> GuardianResult<()> { self.write_log(LogMessage::Init(Box::new(msg))).await } pub async fn log_withdraw(&self, msg: WithdrawalLogMessage) -> GuardianResult<()> { - self.write_log(LogMessage::Withdrawal(Box::new(msg))).await + self.write_log_or_abort(LogMessage::Withdrawal(Box::new(msg))) + .await } pub async fn log_committee_update(&self, msg: CommitteeUpdateLogMessage) -> GuardianResult<()> { - self.write_log(LogMessage::CommitteeUpdate(Box::new(msg))) + self.write_log_or_abort(LogMessage::CommitteeUpdate(Box::new(msg))) .await } pub async fn log_genesis(&self, msg: GenesisLogMessage) -> GuardianResult<()> { - self.write_log(LogMessage::Genesis(Box::new(msg))).await + self.write_log_or_abort(LogMessage::Genesis(Box::new(msg))) + .await } pub async fn log_heartbeat(&self, seq: u64) -> GuardianResult<()> { - self.write_log(LogMessage::Heartbeat { seq }).await + self.write_log_or_abort(LogMessage::Heartbeat { seq }).await } pub async fn log_ceremony(&self, state: CeremonyLogMessage) -> GuardianResult<()> { - self.write_log(LogMessage::Ceremony(Box::new(state))).await + self.write_log_or_abort(LogMessage::Ceremony(Box::new(state))) + .await } /// Persist the current encrypted KP share state to `kp-shares/` for recovery. @@ -560,7 +547,7 @@ impl Enclave { cert_seq: u64, encrypted_shares: KPEncryptedShares, ) -> GuardianResult<()> { - self.write_log(LogMessage::KpShareState(Box::new(KpShareState::new( + self.write_log_or_abort(LogMessage::KpShareState(Box::new(KpShareState::new( sharing_seq, cert_seq, encrypted_shares, diff --git a/crates/hashi-guardian/src/lib.rs b/crates/hashi-guardian/src/lib.rs index f9adaa7528..7f5c9c06c0 100644 --- a/crates/hashi-guardian/src/lib.rs +++ b/crates/hashi-guardian/src/lib.rs @@ -5,8 +5,6 @@ use std::time::Duration; // TODO: Leave as consts or make them configurable? pub const HEARTBEAT_INTERVAL: Duration = Duration::from_mins(1); -pub const HEARTBEAT_RETRY_INTERVAL: Duration = Duration::from_secs(10); -pub const MAX_HEARTBEAT_FAILURES_INTERVAL: Duration = Duration::from_mins(5); pub mod attestation; pub mod ceremony_mode; diff --git a/crates/hashi-guardian/src/main.rs b/crates/hashi-guardian/src/main.rs index 08e30a84de..a6fac2f334 100644 --- a/crates/hashi-guardian/src/main.rs +++ b/crates/hashi-guardian/src/main.rs @@ -6,8 +6,6 @@ use hashi_guardian::rpc::GuardianGrpc; use hashi_guardian::withdraw_mode::heartbeat::HeartbeatWriter; use hashi_guardian::Enclave; use hashi_guardian::HEARTBEAT_INTERVAL; -use hashi_guardian::HEARTBEAT_RETRY_INTERVAL; -use hashi_guardian::MAX_HEARTBEAT_FAILURES_INTERVAL; use hashi_types::guardian::EnclaveMode; use hashi_types::guardian::GuardianEncKeyPair; use hashi_types::guardian::GuardianSignKeyPair; @@ -68,31 +66,19 @@ async fn main() -> Result<()> { .set_serving::>() .await; - let server_future = Server::builder() - .add_service(health_service) - .add_service(GuardianServiceServer::new(svc)) - .serve(addr); - // Don't emit heartbeats in ceremony mode: their primary function is // to allow KPs to detect old sessions that might still be running // in order to bypass limiter. Not a concern for ceremony mode. - if ceremony_mode { - return server_future - .await - .map_err(|e| anyhow::anyhow!("Server error: {}", e)); + if !ceremony_mode { + let _ = tokio::spawn(HeartbeatWriter::new(enclave).run(HEARTBEAT_INTERVAL)); } - let heartbeat_future = HeartbeatWriter::new(enclave, MAX_HEARTBEAT_FAILURES_INTERVAL) - .run(HEARTBEAT_INTERVAL, HEARTBEAT_RETRY_INTERVAL); - - tokio::select! { - res = server_future => { - res.map_err(|e| anyhow::anyhow!("Server error: {}", e)) - } - res = heartbeat_future => { - panic!("Heartbeat failed: {:?}", res) - } - } + Server::builder() + .add_service(health_service) + .add_service(GuardianServiceServer::new(svc)) + .serve(addr) + .await + .map_err(|e| anyhow::anyhow!("Server error: {}", e)) } /// Make any panic abort the process instead of unwinding to the tokio task diff --git a/crates/hashi-guardian/src/rpc.rs b/crates/hashi-guardian/src/rpc.rs index bb84a2e8a1..7112a15ba9 100644 --- a/crates/hashi-guardian/src/rpc.rs +++ b/crates/hashi-guardian/src/rpc.rs @@ -9,7 +9,7 @@ use crate::operator_init; use crate::withdraw_mode::committee_update; use crate::withdraw_mode::genesis; use crate::withdraw_mode::provisioner_init; -use crate::withdraw_mode::standard; +use crate::withdraw_mode::standard_withdrawal; use crate::Enclave; use hashi_types::guardian::proto_conversions; use hashi_types::guardian::proto_conversions::pb_to_signed_committee_transition; @@ -193,9 +193,10 @@ impl proto::guardian_service_server::GuardianService for GuardianGrpc { .map_err(to_status)?; // core withdraw call - let response = standard::standard_withdrawal(self.enclave.clone(), validated_req) - .await - .map_err(to_status)?; + let response = + standard_withdrawal::standard_withdrawal(self.enclave.clone(), validated_req) + .await + .map_err(to_status)?; // domain to proto let resp_pb = proto_conversions::standard_withdrawal_response_signed_to_pb(response); diff --git a/crates/hashi-guardian/src/s3_client.rs b/crates/hashi-guardian/src/s3_client.rs index e61dc40da0..5e809a5e83 100644 --- a/crates/hashi-guardian/src/s3_client.rs +++ b/crates/hashi-guardian/src/s3_client.rs @@ -28,8 +28,14 @@ use hashi_types::guardian::VerifiedSessionInfo; use serde::de::DeserializeOwned; use serde::Serialize; use tracing::info; +use tracing::warn; +/// Maximum attempts the AWS SDK makes before returning one write failure. const MAX_RETRY_ATTEMPTS: u32 = 5; +/// Delay between application-level retries of an immutable S3 log write. +const S3_WRITE_RETRY_INTERVAL: Duration = Duration::from_secs(10); +/// Maximum write failure interval before the enclave aborts. +const MAX_S3_WRITE_FAILURE_INTERVAL: Duration = Duration::from_mins(5); #[derive(Clone)] pub struct GuardianS3Client { @@ -103,12 +109,59 @@ impl GuardianS3Client { // S3 Write // ======================================================================== + /// Attempt one immutable log write, relying on the AWS SDK retry policy. pub async fn write_log_record(&self, log: LogRecord) -> GuardianResult<()> { let object_lock_duration = log.object_lock_duration(); let key = log.object_key(); self.write_at_key(&key, &log, object_lock_duration).await } + /// Retry an immutable log write through the grace period, then abort. The + /// worker is detached so caller cancellation cannot abandon an in-flight PUT. + pub async fn write_log_record_or_abort(&self, log: LogRecord) -> GuardianResult<()> { + let writer = self.clone(); + tokio::spawn(async move { + writer + .write_log_record_or_abort_inner( + log, + MAX_S3_WRITE_FAILURE_INTERVAL, + S3_WRITE_RETRY_INTERVAL, + ) + .await + }) + .await + .expect("S3 log writer task failed") + } + + async fn write_log_record_or_abort_inner( + &self, + log: LogRecord, + max_failure_interval: Duration, + retry_interval: Duration, + ) -> GuardianResult<()> { + let object_lock_duration = log.object_lock_duration(); + let key = log.object_key(); + let write_until_success = async { + loop { + match self.write_at_key(&key, &log, object_lock_duration).await { + Ok(()) => return Ok(()), + Err(error) => { + warn!(%key, ?error, "S3 log write failed; retrying"); + tokio::time::sleep(retry_interval).await; + } + } + } + }; + + match tokio::time::timeout(max_failure_interval, write_until_success).await { + Ok(result) => result, + Err(_) => panic!( + "S3 log {} was not written within {:?}", + key, max_failure_interval + ), + } + } + /// Write a value to S3 at an explicit key. /// /// This is intended for ordered log streams where the caller determines the key. @@ -127,27 +180,37 @@ impl GuardianS3Client { .checked_add(object_lock_duration) .expect("Cant overflow"); - let body = serde_json::to_string(value).expect("Cant serialize to JSON"); - info!("Log message: {}", body); + let body = serde_json::to_vec(value).expect("Cant serialize to JSON"); + info!("Log message: {}", String::from_utf8_lossy(&body)); - // TODO(integration-test): Verify on a real S3 bucket with Object Lock enabled that an object written with `Compliance` + `retain_until` cannot be deleted/overwritten before expiry. - let _result = s3_client + // `If-None-Match: *` makes retries safe: a lost-ack write that already + // landed returns 412 instead of creating another version. A 412 is only + // success if the existing immutable object is exactly this record. + let result = s3_client .put_object() .bucket(s3_config.bucket_name()) .key(key) .content_type("application/json") .object_lock_mode(ObjectLockMode::Compliance) .object_lock_retain_until_date(DateTime::from(expiry_time)) - .body(ByteStream::from(body.into_bytes())) + .if_none_match("*") + .body(ByteStream::from(body.clone())) .send() - .await - // DisplayErrorContext displays the full error returned by the SDK - .map_err(|e| { - S3Error(format!( + .await; + if let Err(e) = result { + let already_written = e + .raw_response() + .is_some_and(|resp| resp.status().as_u16() == 412); + if !already_written { + // DisplayErrorContext displays the full error returned by the SDK + return Err(S3Error(format!( "Failed to write to s3: {}", DisplayErrorContext(&e) - )) - })?; + ))); + } + self.verify_existing_write(key, &body).await?; + info!("Object {} already contains the intended record", key); + } info!("Logged entry to immutable storage"); info!("Object locked until: {:?}", expiry_time); @@ -160,6 +223,17 @@ impl GuardianS3Client { Ok(()) } + async fn verify_existing_write(&self, key: &str, expected_body: &[u8]) -> GuardianResult<()> { + let actual_body = self.get_locked_object_bytes(key).await?; + if actual_body != expected_body { + // A 412 revealed different content at this write-once key. Retrying + // cannot replace it, so continuing would violate log durability. + panic!("existing object {key} differs from the intended record"); + } + + Ok(()) + } + // ======================================================================== // S3 Connectivity Tests // ======================================================================== @@ -422,8 +496,7 @@ impl GuardianS3Client { Ok(out) } - /// Point read. This method is unsafe to use since the bucket operator might've overwritten objects. - async fn get_object_unsafe(&self, key: &str) -> GuardianResult { + async fn get_locked_object_bytes(&self, key: &str) -> GuardianResult> { let s3_client = &self.client; let s3_config = &self.config; @@ -459,7 +532,14 @@ impl GuardianS3Client { )) })?; - serde_json::from_slice::(&bytes.into_bytes()).map_err(|e| { + Ok(bytes.into_bytes().to_vec()) + } + + /// Point read. This method is unsafe to use since the bucket operator might've overwritten objects. + async fn get_object_unsafe(&self, key: &str) -> GuardianResult { + let bytes = self.get_locked_object_bytes(key).await?; + + serde_json::from_slice::(&bytes).map_err(|e| { S3Error(format!( "Failed to deserialize object {} into target type: {}", key, e @@ -592,11 +672,13 @@ impl GuardianS3Client { #[cfg(test)] mod tests { use super::*; + use aws_sdk_s3::operation::get_object::GetObjectOutput; use aws_sdk_s3::operation::put_object::PutObjectOutput; use aws_sdk_s3::Client; use aws_smithy_mocks::mock; use aws_smithy_mocks::mock_client; use aws_smithy_mocks::RuleMode; + use hashi_types::guardian::GuardianSignKeyPair; fn mk_logger_with_client(client: Client) -> GuardianS3Client { let config = S3Config { @@ -624,6 +706,7 @@ mod tests { && req.content_type() == Some("application/json") && req.object_lock_mode() == Some(&ObjectLockMode::Compliance) && req.object_lock_retain_until_date().is_some() + && req.if_none_match() == Some("*") }) .then_output(|| PutObjectOutput::builder().build()); @@ -641,6 +724,134 @@ mod tests { assert_eq!(put_ok.num_calls(), 1); } + #[tokio::test] + async fn test_412_accepts_identical_locked_object() { + let put_precondition_failed = mock!(Client::put_object) + .match_requests(|req| req.bucket() == Some("bucket")) + .sequence() + .http_status(412, None) + .build(); + let get_existing = mock!(Client::get_object) + .match_requests(|req| req.bucket() == Some("bucket") && req.key() == Some("key")) + .then_output(|| { + GetObjectOutput::builder() + .object_lock_mode(ObjectLockMode::Compliance) + .object_lock_retain_until_date(DateTime::from( + SystemTime::now() + Duration::from_mins(5), + )) + .body(ByteStream::from_static(br#"{"a":1}"#)) + .build() + }); + + let client = mock_client!( + aws_sdk_s3, + RuleMode::MatchAny, + &[&put_precondition_failed, &get_existing], + |builder| builder.retry_config(RetryConfig::standard().with_max_attempts(1)) + ); + let logger = mk_logger_with_client(client); + logger + .write_at_key("key", &TestPayload { a: 1 }, Duration::from_mins(5)) + .await + .unwrap(); + + assert_eq!(put_precondition_failed.num_calls(), 1); + assert_eq!(get_existing.num_calls(), 1); + } + + #[tokio::test] + #[should_panic(expected = "differs from the intended record")] + async fn test_412_mismatch_panics() { + let put_precondition_failed = mock!(Client::put_object) + .match_requests(|req| req.bucket() == Some("bucket")) + .sequence() + .http_status(412, None) + .build(); + let get_existing = mock!(Client::get_object) + .match_requests(|req| req.bucket() == Some("bucket") && req.key() == Some("key")) + .then_output(|| { + GetObjectOutput::builder() + .object_lock_mode(ObjectLockMode::Compliance) + .object_lock_retain_until_date(DateTime::from( + SystemTime::now() + Duration::from_mins(5), + )) + .body(ByteStream::from_static(br#"{"a":2}"#)) + .build() + }); + + let client = mock_client!( + aws_sdk_s3, + RuleMode::MatchAny, + &[&put_precondition_failed, &get_existing], + |builder| builder.retry_config(RetryConfig::standard().with_max_attempts(1)) + ); + let logger = mk_logger_with_client(client); + logger + .write_at_key("key", &TestPayload { a: 1 }, Duration::from_mins(5)) + .await + .unwrap(); + } + + #[tokio::test] + async fn test_log_writer_retries_beyond_sdk_attempts() { + let put_flaky = mock!(Client::put_object) + .match_requests(|req| req.bucket() == Some("bucket")) + .sequence() + .http_status(500, None) + .times(5) + .output(|| PutObjectOutput::builder().build()) + .build(); + let client = mock_client!(aws_sdk_s3, RuleMode::Sequential, &[&put_flaky], |b| b + .retry_config(RetryConfig::standard().with_max_attempts(1))); + let logger = mk_logger_with_client(client); + let signing_key = GuardianSignKeyPair::new(rand::thread_rng()); + let log = LogRecord::new( + "session".to_string(), + LogMessage::Heartbeat { seq: 7 }, + &signing_key, + ); + + // The generous deadline avoids CI timing sensitivity without slowing success; + // zero delay keeps the application-level retry sequence immediate. + logger + .write_log_record_or_abort_inner(log, Duration::from_secs(10), Duration::ZERO) + .await + .unwrap(); + assert_eq!(put_flaky.num_calls(), 6); + } + + #[tokio::test] + #[should_panic(expected = "was not written within")] + async fn test_log_writer_panics_after_failure_interval() { + let put_fail = mock!(Client::put_object) + .match_requests(|req| req.bucket() == Some("bucket")) + .sequence() + .http_status(500, None) + .times(100) + .build(); + let client = mock_client!(aws_sdk_s3, RuleMode::MatchAny, &[&put_fail], |builder| { + builder.retry_config(RetryConfig::standard().with_max_attempts(1)) + }); + let logger = mk_logger_with_client(client); + let signing_key = GuardianSignKeyPair::new(rand::thread_rng()); + let log = LogRecord::new( + "session".to_string(), + LogMessage::Heartbeat { seq: 7 }, + &signing_key, + ); + + // The first PUT runs immediately, then retries every 5ms until the 50ms + // deadline panics (roughly 10 attempts; scheduling makes the count inexact). + logger + .write_log_record_or_abort_inner( + log, + Duration::from_millis(50), + Duration::from_millis(5), + ) + .await + .unwrap(); + } + #[tokio::test] async fn test_write_retries_on_transient_failures() { // Two transient failures followed by success. diff --git a/crates/hashi-guardian/src/withdraw_mode/heartbeat.rs b/crates/hashi-guardian/src/withdraw_mode/heartbeat.rs index 9bc303abed..455350228b 100644 --- a/crates/hashi-guardian/src/withdraw_mode/heartbeat.rs +++ b/crates/hashi-guardian/src/withdraw_mode/heartbeat.rs @@ -2,34 +2,21 @@ // SPDX-License-Identifier: Apache-2.0 use crate::Enclave; -use hashi_types::guardian::GuardianError; use hashi_types::guardian::GuardianResult; use std::sync::Arc; use std::time::Duration; -use std::time::Instant; /// Stateful heartbeat writer. pub struct HeartbeatWriter { pub enclave: Arc, - pub max_failures_interval: Duration, - /// a local record to track how many heartbeat attempts failed - pub consecutive_failures: u32, - /// records the last success timestamp - pub last_success_at: Option, - /// records the first failure timestamp - pub failure_window_start_at: Option, /// the next sequence number used in s3 logs pub next_seq: u64, } impl HeartbeatWriter { - pub fn new(enclave: Arc, max_failures_interval: Duration) -> Self { + pub fn new(enclave: Arc) -> Self { Self { enclave, - max_failures_interval, - consecutive_failures: 0, - last_success_at: None, - failure_window_start_at: None, next_seq: 0, } } @@ -37,60 +24,27 @@ impl HeartbeatWriter { /// Attempt to send one heartbeat. /// /// - If operator init is not complete, this is a no-op. - /// - On success, resets failure state. - /// - On S3 error, retries are driven by `run`; errors after elapsed failure window. - pub async fn tick(&mut self, now: Instant) -> GuardianResult<()> { + /// The shared S3 writer retries failures and aborts the process if its grace + /// period expires. + pub async fn tick(&mut self) -> GuardianResult<()> { if !self.enclave.is_operator_init_complete() { return Ok(()); } - match self.enclave.log_heartbeat(self.next_seq).await { - Ok(()) => { - self.consecutive_failures = 0; - self.last_success_at = Some(now); - self.failure_window_start_at = None; - self.next_seq += 1; - } - Err(e) => { - // other errors shouldn't occur due to checks in operator_init_complete - debug_assert!(matches!(e, GuardianError::S3Error(_))); - - self.consecutive_failures += 1; - if self.failure_window_start_at.is_none() { - self.failure_window_start_at = Some(now); - } - - let reference = self - .last_success_at - .or(self.failure_window_start_at) - .unwrap_or(now); - if now.duration_since(reference) > self.max_failures_interval { - return Err(GuardianError::InternalError(format!( - "Heartbeat has no successful write for {:?}; last success {:?}, failure window start {:?}, consecutive failures {}, latest error {:?}", - self.max_failures_interval, - self.last_success_at, - self.failure_window_start_at, - self.consecutive_failures, - e - ))); - } - } - } - + self.enclave.log_heartbeat(self.next_seq).await?; + self.next_seq += 1; Ok(()) } /// Run the periodic heartbeat loop. - pub async fn run(mut self, interval: Duration, retry_interval: Duration) -> GuardianResult<()> { + pub async fn run(mut self, interval: Duration) { let mut delay = Duration::ZERO; loop { tokio::time::sleep(delay).await; - self.tick(Instant::now()).await?; - delay = if self.consecutive_failures > 0 { - retry_interval - } else { - interval - }; + self.tick() + .await + .expect("heartbeat write failed unexpectedly"); + delay = interval; } } } @@ -99,108 +53,18 @@ impl HeartbeatWriter { mod tests { use super::*; - use crate::s3_client::GuardianS3Client; - use crate::OperatorInitTestArgs; - use aws_sdk_s3::operation::put_object::PutObjectOutput; - use aws_sdk_s3::Client; - use aws_smithy_mocks::mock; - use aws_smithy_mocks::mock_client; - use aws_smithy_mocks::RuleMode; - use hashi_types::guardian::S3Config; - - fn mk_s3_logger(client: Client) -> GuardianS3Client { - GuardianS3Client::from_client_for_tests(S3Config::mock_for_testing(), client) - } - - async fn mk_operator_initialized_enclave(s3_logger: GuardianS3Client) -> Arc { - Enclave::create_operator_initialized_with( - OperatorInitTestArgs::default().with_s3_logger(s3_logger), - ) - .await - } - #[tokio::test] - async fn test_heartbeat_fails_after_max_failure_interval() { - // Mock S3 client that always fails put_object. - let put_fail = mock!(Client::put_object) - .match_requests(|req| req.bucket() == Some("test-bucket")) - .sequence() - .http_status(500, None) - .times(10) - .build(); - - // Disable retries so each heartbeat attempt makes exactly one put_object call. - let client = mock_client!(aws_sdk_s3, RuleMode::MatchAny, &[&put_fail], |b| b - .retry_config( - aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1) - )); - - let enclave = mk_operator_initialized_enclave(mk_s3_logger(client)).await; - - let max_failure_interval = Duration::from_secs(20); - let mut writer = HeartbeatWriter::new(enclave, max_failure_interval); - let t0 = Instant::now(); - - assert!(writer.tick(t0).await.is_ok()); - assert_eq!(writer.consecutive_failures, 1); - assert_eq!(writer.last_success_at, None); - assert_eq!(writer.failure_window_start_at, Some(t0)); - - assert!(writer.tick(t0 + Duration::from_secs(10)).await.is_ok()); - assert_eq!(writer.consecutive_failures, 2); - assert_eq!(writer.last_success_at, None); - - assert!(writer.tick(t0 + Duration::from_secs(21)).await.is_err()); - assert_eq!(put_fail.num_calls(), 3usize); + async fn heartbeat_is_a_noop_before_operator_init() { + let mut writer = HeartbeatWriter::new(Enclave::create_with_random_keys()); + writer.tick().await.unwrap(); + assert_eq!(writer.next_seq, 0); } #[tokio::test] - async fn test_heartbeat_resets_failure_state_on_success() { - // fail twice, succeed, then fail again - let put_flaky = mock!(Client::put_object) - .match_requests(|req| req.bucket() == Some("test-bucket")) - .sequence() - .http_status(500, None) - .times(2) - .output(|| PutObjectOutput::builder().build()) - .http_status(500, None) - .build(); - - // Disable retries so each heartbeat attempt makes exactly one put_object call. - let client = mock_client!(aws_sdk_s3, RuleMode::Sequential, &[&put_flaky], |b| b - .retry_config( - aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1) - )); - - let enclave = mk_operator_initialized_enclave(mk_s3_logger(client)).await; - - let max_failure_interval = Duration::from_secs(30); - let mut writer = HeartbeatWriter::new(enclave, max_failure_interval); - let t0 = Instant::now(); - - assert!(writer.tick(t0).await.is_ok()); - assert_eq!(writer.consecutive_failures, 1); - assert_eq!(writer.last_success_at, None); - assert_eq!(writer.failure_window_start_at, Some(t0)); - - assert!(writer.tick(t0 + Duration::from_secs(10)).await.is_ok()); - assert_eq!(writer.consecutive_failures, 2); - assert_eq!(writer.last_success_at, None); - - assert!(writer.tick(t0 + Duration::from_secs(15)).await.is_ok()); - assert_eq!(writer.consecutive_failures, 0, "should reset"); - assert_eq!(writer.next_seq, 1, "success should advance sequence"); - assert_eq!(writer.last_success_at, Some(t0 + Duration::from_secs(15))); - assert_eq!(writer.failure_window_start_at, None); - - assert!(writer.tick(t0 + Duration::from_secs(40)).await.is_ok()); - assert_eq!(writer.consecutive_failures, 1); - assert_eq!(writer.last_success_at, Some(t0 + Duration::from_secs(15))); - assert_eq!( - writer.failure_window_start_at, - Some(t0 + Duration::from_secs(40)) - ); - - assert_eq!(put_flaky.num_calls(), 4usize); + async fn heartbeat_advances_after_durable_write() { + let enclave = Enclave::create_operator_initialized_with(Default::default()).await; + let mut writer = HeartbeatWriter::new(enclave); + writer.tick().await.unwrap(); + assert_eq!(writer.next_seq, 1); } } diff --git a/crates/hashi-guardian/src/withdraw_mode/mod.rs b/crates/hashi-guardian/src/withdraw_mode/mod.rs index d3717f69b4..6c6e2fb6be 100644 --- a/crates/hashi-guardian/src/withdraw_mode/mod.rs +++ b/crates/hashi-guardian/src/withdraw_mode/mod.rs @@ -3,13 +3,14 @@ //! Withdraw-mode flows (enabled when CEREMONY_MODE=false): standard withdrawal, //! committee updates, provisioner init, and heartbeats. `verify_hashi_cert` is -//! the committee-certificate check shared by `standard` and `committee_update`. +//! the committee-certificate check shared by `standard_withdrawal` and +//! `committee_update`. pub mod committee_update; pub mod genesis; pub mod heartbeat; pub mod provisioner_init; -pub mod standard; +pub mod standard_withdrawal; use hashi_types::guardian::GuardianError::InvalidInputs; use hashi_types::guardian::GuardianResult; diff --git a/crates/hashi-guardian/src/withdraw_mode/standard.rs b/crates/hashi-guardian/src/withdraw_mode/standard_withdrawal.rs similarity index 91% rename from crates/hashi-guardian/src/withdraw_mode/standard.rs rename to crates/hashi-guardian/src/withdraw_mode/standard_withdrawal.rs index 3edf717e0a..19c40a1456 100644 --- a/crates/hashi-guardian/src/withdraw_mode/standard.rs +++ b/crates/hashi-guardian/src/withdraw_mode/standard_withdrawal.rs @@ -2,7 +2,6 @@ // SPDX-License-Identifier: Apache-2.0 use super::verify_hashi_cert; -use crate::enclave::LimiterGuard; use crate::Enclave; use bitcoin::Txid; use hashi_types::committee::certificate_threshold; @@ -14,12 +13,14 @@ use hashi_types::guardian::GuardianError::InvalidInputs; use hashi_types::guardian::GuardianResult; use hashi_types::guardian::GuardianSigned; use hashi_types::guardian::HashiSigned; +use hashi_types::guardian::RateLimiter; use hashi_types::guardian::StandardWithdrawalRequest; use hashi_types::guardian::StandardWithdrawalRequestWire; use hashi_types::guardian::StandardWithdrawalResponse; use hashi_types::guardian::WithdrawalID; use hashi_types::guardian::WithdrawalLogMessage; use std::sync::Arc; +use tokio::sync::OwnedMutexGuard; use tracing::error; use tracing::info; @@ -44,7 +45,7 @@ pub async fn standard_withdrawal( response: response.clone(), post_state, }; - log_withdrawal_success(enclave.as_ref(), wid, msg, limiter_guard).await?; + log_withdrawal_success(enclave.clone(), wid, msg, limiter_guard).await?; // <-- Limiter guard drops upon log_withdrawal_success return. Next withdrawal can begin. Ok(enclave.sign(response)) } @@ -64,7 +65,11 @@ pub async fn standard_withdrawal( async fn normal_withdrawal_inner( enclave: Arc, signed_request: HashiSigned, -) -> GuardianResult<(Txid, StandardWithdrawalResponse, LimiterGuard)> { +) -> GuardianResult<( + Txid, + StandardWithdrawalResponse, + OwnedMutexGuard, +)> { // 0) Validation if !enclave.is_fully_initialized() { return Err(EnclaveUninitialized); @@ -82,7 +87,7 @@ async fn normal_withdrawal_inner( // 2) Rate limits: acquire exclusive lock on limiter, consume tokens. // The returned guard holds the mutex — no other withdrawal can proceed - // until this one is committed or reverted. + // until this one is durably logged or the enclave aborts. // // Reject timestamps too far in the future (clock skew protection). // Old timestamps are safe — the limiter's monotonicity check prevents replay, @@ -127,25 +132,25 @@ async fn normal_withdrawal_inner( } async fn log_withdrawal_success( - enclave: &Enclave, + enclave: Arc, wid: WithdrawalID, msg: WithdrawalLogMessage, - limiter_guard: LimiterGuard, + limiter_guard: OwnedMutexGuard, ) -> GuardianResult<()> { - match enclave.log_withdraw(msg).await { - Ok(_) => { - info!("Withdrawal {} logged.", wid); - // Commit limiter consumption only after we've successfully logged. - limiter_guard.commit(); - Ok(()) - } - Err(e) => { - // Logging failed => return Err (do not return signatures). - // Note that LimiterGuard::Drop will revert the limiter - error!("Logging withdrawal {} to S3 failed: {:?}", wid, e); - Err(e) - } - } + // The task owns the limiter guard and continues if the RPC future is cancelled. + // In production, exhausting the write grace period panics and the process-wide + // panic hook aborts the enclave. + tokio::spawn(async move { + enclave + .log_withdraw(msg) + .await + .expect("withdrawal log write failed"); + info!("Withdrawal {} logged.", wid); + drop(limiter_guard); + }) + .await + .expect("withdrawal log task failed"); + Ok(()) } async fn log_withdrawal_failure( diff --git a/crates/hashi-types/src/guardian/limiter.rs b/crates/hashi-types/src/guardian/limiter.rs index 60e9d5d4c7..42a89c62ec 100644 --- a/crates/hashi-types/src/guardian/limiter.rs +++ b/crates/hashi-types/src/guardian/limiter.rs @@ -52,8 +52,6 @@ impl LimiterState { pub struct RateLimiter { config: LimiterConfig, state: LimiterState, - /// Snapshot of state before the most recent `consume`, used for revert. - prev_state: LimiterState, } impl RateLimiter { @@ -63,12 +61,7 @@ impl RateLimiter { "num_tokens_available exceeds max_bucket_capacity".into(), )); } - let prev_state = state; - Ok(Self { - config, - state, - prev_state, - }) + Ok(Self { config, state }) } pub fn config(&self) -> &LimiterConfig { @@ -114,17 +107,11 @@ impl RateLimiter { return Err(RateLimitExceeded); } - self.prev_state = self.state; self.state.last_updated_at = timestamp; self.state.num_tokens_available = capacity - amount_sats; self.state.next_seq += 1; Ok(()) } - - /// Revert a previous `consume`. Restores state to pre-consume snapshot. - pub fn revert(&mut self) { - self.state = self.prev_state; - } } #[cfg(test)] @@ -181,19 +168,6 @@ mod test { ); } - #[test] - fn test_revert_restores_pre_refill_state() { - let (config, state) = make_limiter(); - let mut limiter = RateLimiter::new(config, state).unwrap(); - // Consume after refill, then revert — should restore original state. - limiter.consume(0, 100, 50_000).unwrap(); - assert_eq!(limiter.state().num_tokens_available, 50_000); // 100*1000 - 50_000 - limiter.revert(); - assert_eq!(limiter.state().num_tokens_available, 0); - assert_eq!(limiter.state().last_updated_at, 0); - assert_eq!(limiter.state().next_seq, 0); - } - #[test] fn test_rejects_wrong_seq_and_old_timestamp() { let (config, state) = make_limiter(); From c0ed07bd5f247d102da481df345dbe8fe8073acf Mon Sep 17 00:00:00 2001 From: Deepak Maram Date: Fri, 10 Jul 2026 19:49:49 -0400 Subject: [PATCH 2/4] [Guardian] Fix status-lock stall and retry log spam Two incidental issues from the S3 durable-or-abort refactor: - limiter_state()/limiter_config() took a bare untimed lock, so a get_guardian_info status poll could block for the full 5-min write grace behind an in-flight durable write. Bound both reads by LIMITER_LOCK_TIMEOUT via a shared helper, returning None on timeout (info() already treats None as unavailable). - write_at_key logged the full serialized record (request data + BTC signatures) at info on every attempt; now inside the retry loop, that re-emitted the record ~18x during an outage. The record is already durably written to S3, so drop the redundant trace-stream dump. Also fixes a doc_lazy_continuation clippy error in heartbeat.rs. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/hashi-guardian/src/enclave.rs | 20 +++++++++++++++---- crates/hashi-guardian/src/s3_client.rs | 1 - .../src/withdraw_mode/heartbeat.rs | 1 + 3 files changed, 17 insertions(+), 5 deletions(-) diff --git a/crates/hashi-guardian/src/enclave.rs b/crates/hashi-guardian/src/enclave.rs index 710e86c228..6728ed3dcd 100644 --- a/crates/hashi-guardian/src/enclave.rs +++ b/crates/hashi-guardian/src/enclave.rs @@ -312,13 +312,25 @@ impl EnclaveState { } pub async fn limiter_state(&self) -> Option { - let limiter = self.rate_limiter.get()?; - Some(*limiter.lock().await.state()) + Some(*self.lock_limiter_for_read().await?.state()) } pub async fn limiter_config(&self) -> Option { - let limiter = self.rate_limiter.get()?; - Some(*limiter.lock().await.config()) + Some(*self.lock_limiter_for_read().await?.config()) + } + + /// Lock the limiter for a read-only status query, bounded by + /// `LIMITER_LOCK_TIMEOUT` so an in-flight durable write cannot stall `info` + /// for minutes. Returns None if uninitialized or the lock is still held at + /// the deadline. + async fn lock_limiter_for_read(&self) -> Option> { + let rate_limiter = self.rate_limiter.get()?; + tokio::time::timeout( + Self::LIMITER_LOCK_TIMEOUT, + rate_limiter.clone().lock_owned(), + ) + .await + .ok() } } diff --git a/crates/hashi-guardian/src/s3_client.rs b/crates/hashi-guardian/src/s3_client.rs index c9178561ee..5c9dbed1f1 100644 --- a/crates/hashi-guardian/src/s3_client.rs +++ b/crates/hashi-guardian/src/s3_client.rs @@ -182,7 +182,6 @@ impl GuardianS3Client { .expect("Cant overflow"); let body = serde_json::to_vec(value).expect("Cant serialize to JSON"); - info!("Log message: {}", String::from_utf8_lossy(&body)); // `If-None-Match: *` makes retries safe: a lost-ack write that already // landed returns 412 instead of creating another version. A 412 is only diff --git a/crates/hashi-guardian/src/withdraw_mode/heartbeat.rs b/crates/hashi-guardian/src/withdraw_mode/heartbeat.rs index 455350228b..e57e63f394 100644 --- a/crates/hashi-guardian/src/withdraw_mode/heartbeat.rs +++ b/crates/hashi-guardian/src/withdraw_mode/heartbeat.rs @@ -24,6 +24,7 @@ impl HeartbeatWriter { /// Attempt to send one heartbeat. /// /// - If operator init is not complete, this is a no-op. + /// /// The shared S3 writer retries failures and aborts the process if its grace /// period expires. pub async fn tick(&mut self) -> GuardianResult<()> { From 445014574eeea70f3ee842a13b56f82fcc95ab6f Mon Sep 17 00:00:00 2001 From: Deepak Maram Date: Mon, 13 Jul 2026 18:11:04 -0400 Subject: [PATCH 3/4] [Guardian] Harden 412 write verification --- crates/hashi-guardian/src/s3_client.rs | 65 +++++++++++++++++++++++--- 1 file changed, 59 insertions(+), 6 deletions(-) diff --git a/crates/hashi-guardian/src/s3_client.rs b/crates/hashi-guardian/src/s3_client.rs index 5c9dbed1f1..39c5efa384 100644 --- a/crates/hashi-guardian/src/s3_client.rs +++ b/crates/hashi-guardian/src/s3_client.rs @@ -224,12 +224,40 @@ impl GuardianS3Client { } async fn verify_existing_write(&self, key: &str, expected_body: &[u8]) -> GuardianResult<()> { - let actual_body = self.get_locked_object_bytes(key).await?; - if actual_body != expected_body { + let response = self + .client + .get_object() + .bucket(self.config.bucket_name()) + .key(key) + .send() + .await + .map_err(|e| { + S3Error(format!( + "Failed to get object {}: {}", + key, + DisplayErrorContext(&e) + )) + })?; + let has_compliance_lock = response.object_lock_mode() == Some(&ObjectLockMode::Compliance) + && response.object_lock_retain_until_date().is_some(); + let actual_body = response.body.collect().await.map_err(|e| { + S3Error(format!( + "Failed to read object body for key {}: {}", + key, + DisplayErrorContext(&e) + )) + })?; + + if actual_body.into_bytes().as_ref() != expected_body { // A 412 revealed different content at this write-once key. Retrying // cannot replace it, so continuing would violate log durability. panic!("existing object {key} differs from the intended record"); } + if !has_compliance_lock { + // The intended record exists but is not immutable. Retrying cannot + // replace it, so it cannot satisfy the durable-write requirement. + panic!("existing object {key} is missing a valid compliance lock"); + } Ok(()) } @@ -772,10 +800,6 @@ mod tests { .match_requests(|req| req.bucket() == Some("bucket") && req.key() == Some("key")) .then_output(|| { GetObjectOutput::builder() - .object_lock_mode(ObjectLockMode::Compliance) - .object_lock_retain_until_date(DateTime::from( - SystemTime::now() + Duration::from_mins(5), - )) .body(ByteStream::from_static(br#"{"a":2}"#)) .build() }); @@ -793,6 +817,35 @@ mod tests { .unwrap(); } + #[tokio::test] + #[should_panic(expected = "is missing a valid compliance lock")] + async fn test_412_identical_unlocked_object_panics() { + let put_precondition_failed = mock!(Client::put_object) + .match_requests(|req| req.bucket() == Some("bucket")) + .sequence() + .http_status(412, None) + .build(); + let get_existing = mock!(Client::get_object) + .match_requests(|req| req.bucket() == Some("bucket") && req.key() == Some("key")) + .then_output(|| { + GetObjectOutput::builder() + .body(ByteStream::from_static(br#"{"a":1}"#)) + .build() + }); + + let client = mock_client!( + aws_sdk_s3, + RuleMode::MatchAny, + &[&put_precondition_failed, &get_existing], + |builder| builder.retry_config(RetryConfig::standard().with_max_attempts(1)) + ); + let logger = mk_logger_with_client(client); + logger + .write_at_key("key", &TestPayload { a: 1 }, Duration::from_mins(5)) + .await + .unwrap(); + } + #[tokio::test] async fn test_log_writer_retries_beyond_sdk_attempts() { let put_flaky = mock!(Client::put_object) From 260dbaab12b90511d649c9e07a07982c96ce905a Mon Sep 17 00:00:00 2001 From: Deepak Maram Date: Mon, 13 Jul 2026 18:23:42 -0400 Subject: [PATCH 4/4] [Guardian] Enforce heartbeat fail-stop bound --- crates/hashi-guardian/src/lib.rs | 15 ++++++++++- crates/hashi-guardian/src/main.rs | 4 ++- crates/hashi-guardian/src/s3_client.rs | 25 +++++++------------ .../src/s3_reader/heartbeat_checks.rs | 11 ++------ 4 files changed, 28 insertions(+), 27 deletions(-) diff --git a/crates/hashi-guardian/src/lib.rs b/crates/hashi-guardian/src/lib.rs index 7f5c9c06c0..b0da4468b0 100644 --- a/crates/hashi-guardian/src/lib.rs +++ b/crates/hashi-guardian/src/lib.rs @@ -3,8 +3,21 @@ use std::time::Duration; -// TODO: Leave as consts or make them configurable? +/// Interval between successful heartbeat writes. pub const HEARTBEAT_INTERVAL: Duration = Duration::from_mins(1); +/// Maximum write failure interval before the enclave aborts. +pub const MAX_S3_WRITE_FAILURE_INTERVAL: Duration = Duration::from_mins(5); +/// Maximum acceptable heartbeat age for the activating standby session. +pub const LIVE_SESSION_MAX_AGE: Duration = Duration::from_mins(3); +/// Silence required before another session is considered inactive. +pub const OTHER_SESSION_QUIET_PERIOD: Duration = Duration::from_mins(10); + +// An enclave that cannot write its next heartbeat must abort before another +// session is allowed to treat it as quiet. +const _: () = assert!( + HEARTBEAT_INTERVAL.as_secs() + MAX_S3_WRITE_FAILURE_INTERVAL.as_secs() + < OTHER_SESSION_QUIET_PERIOD.as_secs() +); pub mod attestation; pub mod ceremony_mode; diff --git a/crates/hashi-guardian/src/main.rs b/crates/hashi-guardian/src/main.rs index a6fac2f334..6255f3242e 100644 --- a/crates/hashi-guardian/src/main.rs +++ b/crates/hashi-guardian/src/main.rs @@ -70,7 +70,9 @@ async fn main() -> Result<()> { // to allow KPs to detect old sessions that might still be running // in order to bypass limiter. Not a concern for ceremony mode. if !ceremony_mode { - let _ = tokio::spawn(HeartbeatWriter::new(enclave).run(HEARTBEAT_INTERVAL)); + drop(tokio::spawn( + HeartbeatWriter::new(enclave).run(HEARTBEAT_INTERVAL), + )); } Server::builder() diff --git a/crates/hashi-guardian/src/s3_client.rs b/crates/hashi-guardian/src/s3_client.rs index 39c5efa384..c5e8c764d3 100644 --- a/crates/hashi-guardian/src/s3_client.rs +++ b/crates/hashi-guardian/src/s3_client.rs @@ -1,6 +1,7 @@ // Copyright (c) Mysten Labs, Inc. // SPDX-License-Identifier: Apache-2.0 +use crate::MAX_S3_WRITE_FAILURE_INTERVAL; use aws_credential_types::provider::SharedCredentialsProvider; use aws_credential_types::CredentialsBuilder; use aws_sdk_s3::error::DisplayErrorContext; @@ -34,8 +35,6 @@ use tracing::warn; const MAX_RETRY_ATTEMPTS: u32 = 5; /// Delay between application-level retries of an immutable S3 log write. const S3_WRITE_RETRY_INTERVAL: Duration = Duration::from_secs(10); -/// Maximum write failure interval before the enclave aborts. -const MAX_S3_WRITE_FAILURE_INTERVAL: Duration = Duration::from_mins(5); #[derive(Clone)] pub struct GuardianS3Client { @@ -223,6 +222,8 @@ impl GuardianS3Client { Ok(()) } + /// Similar to `get_object_unsafe`, but compares the raw bytes and treats + /// invalid lock metadata as a fatal conflict at this write-once key. async fn verify_existing_write(&self, key: &str, expected_body: &[u8]) -> GuardianResult<()> { let response = self .client @@ -524,13 +525,12 @@ impl GuardianS3Client { Ok(out) } - async fn get_locked_object_bytes(&self, key: &str) -> GuardianResult> { - let s3_client = &self.client; - let s3_config = &self.config; - - let response = s3_client + /// Point read. This method is unsafe to use since the bucket operator might've overwritten objects. + async fn get_object_unsafe(&self, key: &str) -> GuardianResult { + let response = self + .client .get_object() - .bucket(s3_config.bucket_name()) + .bucket(self.config.bucket_name()) .key(key) .send() .await @@ -560,14 +560,7 @@ impl GuardianS3Client { )) })?; - Ok(bytes.into_bytes().to_vec()) - } - - /// Point read. This method is unsafe to use since the bucket operator might've overwritten objects. - async fn get_object_unsafe(&self, key: &str) -> GuardianResult { - let bytes = self.get_locked_object_bytes(key).await?; - - serde_json::from_slice::(&bytes).map_err(|e| { + serde_json::from_slice::(&bytes.into_bytes()).map_err(|e| { S3Error(format!( "Failed to deserialize object {} into target type: {}", key, e diff --git a/crates/hashi-guardian/src/s3_reader/heartbeat_checks.rs b/crates/hashi-guardian/src/s3_reader/heartbeat_checks.rs index a089d43216..d3cf5ea47e 100644 --- a/crates/hashi-guardian/src/s3_reader/heartbeat_checks.rs +++ b/crates/hashi-guardian/src/s3_reader/heartbeat_checks.rs @@ -3,6 +3,8 @@ use super::heartbeat_cursor; use super::GuardianReader; +use crate::LIVE_SESSION_MAX_AGE; +use crate::OTHER_SESSION_QUIET_PERIOD; use hashi_types::guardian::time_utils::now_timestamp_secs; use hashi_types::guardian::time_utils::unix_millis_to_seconds; use hashi_types::guardian::time_utils::UnixSeconds; @@ -10,17 +12,8 @@ use hashi_types::guardian::LogMessage; use hashi_types::guardian::SessionID; use hashi_types::guardian::VerifiedLogRecord; use std::collections::BTreeMap; -use std::time::Duration; use tracing::info; -/// Maximum acceptable heartbeat age for the activating standby session. -/// Set to HEARTBEAT_INTERVAL + grace period to account for clock skew and retries. -const LIVE_SESSION_MAX_AGE: Duration = Duration::from_mins(3); - -/// A non-ignored session with a heartbeat inside this window is considered -/// active, so standby activation must fail. -const OTHER_SESSION_QUIET_PERIOD: Duration = Duration::from_mins(10); - impl GuardianReader { /// Enforces that `live_session` has heartbeated recently, while every other /// guardian session has been quiet long enough to no longer be considered