From fe67af352418f1ace52b02e6ef2a72f8bb8fd79b Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 20 Jul 2026 17:40:46 +0800 Subject: [PATCH] fix(heal): coordinate cluster-wide control operations (#5003) --- .../src/cluster/rpc/peer_rest_client.rs | 42 +- crates/heal/src/heal/channel.rs | 204 +++++- crates/heal/src/heal/manager.rs | 17 + crates/heal/src/heal/task.rs | 27 +- .../src/generated/proto_gen/node_service.rs | 2 + crates/protos/src/heal_control.rs | 162 ++++- crates/protos/src/lib.rs | 129 +++- crates/protos/src/node.proto | 1 + rustfs/src/admin/handlers/heal.rs | 634 +++++++++++----- rustfs/src/server/http.rs | 33 +- rustfs/src/storage/rpc/node_service.rs | 679 ++++++++++++++++-- rustfs/src/storage/rpc/node_service/heal.rs | 37 +- rustfs/src/storage/storage_api.rs | 5 + 13 files changed, 1680 insertions(+), 292 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index c8820cf2a..dd9ee7a5e 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -99,6 +99,11 @@ fn validate_heal_control_capability_proof(canonical_ack: &[u8], proof: &[u8]) -> .map_err(|_| Error::other("peer returned an invalid heal control capability proof")) } +fn validate_heal_control_response_proof(canonical_response: &[u8], proof: &[u8]) -> Result<()> { + verify_tonic_rpc_response_proof(canonical_response, proof) + .map_err(|_| Error::other("peer returned an invalid heal control response proof")) +} + #[derive(Clone, Debug)] pub struct PeerLiveEventsBatch { pub events: Vec, @@ -737,6 +742,7 @@ impl PeerRestClient { if command.len() > HEAL_CONTROL_PAYLOAD_MAX_SIZE { return Err(Error::other("heal control command exceeds size limit")); } + let capability_probe = rustfs_protos::is_heal_control_capability_probe(&command); self.finalize_result( async { let mut client = self @@ -748,9 +754,10 @@ impl PeerRestClient { .map_err(|_| Error::other("heal control request length cannot be represented"))?; let mut request = Request::new(HealControlRequest { version, - topology_fingerprint, - command: command.into(), + topology_fingerprint: topology_fingerprint.clone(), + command: command.clone().into(), }); + request.set_timeout(rustfs_protos::heal_control_execution_timeout()); set_tonic_canonical_body_digest(&mut request, &canonical_body)?; let response = client.heal_control(request).await?.into_inner(); if !response.success { @@ -760,6 +767,16 @@ impl PeerRestClient { .unwrap_or_else(|| "peer heal control failed without an error".to_string()), )); } + if !capability_probe { + let canonical_response = rustfs_protos::canonical_heal_control_response_body( + version, + &topology_fingerprint, + &command, + &response.result, + ) + .map_err(|_| Error::other("heal control response length cannot be represented"))?; + validate_heal_control_response_proof(&canonical_response, &response.response_proof)?; + } Ok(response.result.to_vec()) } .await, @@ -767,7 +784,8 @@ impl PeerRestClient { .await } - /// Confirms that a peer supports heal-control v1 and has the same storage + /// Confirms that a peer supports the current heal-control coordination + /// contract and has the same storage /// topology. Every non-success response is an error so old or divergent /// peers cannot be mistaken for compatible ones. pub async fn probe_heal_control(&self, topology_fingerprint: String) -> Result<()> { @@ -1407,6 +1425,24 @@ mod tests { assert!(err.to_string().contains("invalid heal control capability proof")); } + #[test] + fn heal_control_response_proof_binds_command_and_result() { + runtime_sources::ensure_test_rpc_secret(); + let canonical = rustfs_protos::canonical_heal_control_response_body(2, "fingerprint", b"query", b"result") + .expect("small response should encode"); + let proof = crate::cluster::rpc::sign_tonic_rpc_response_proof(&canonical).expect("test proof should sign"); + assert!(validate_heal_control_response_proof(&canonical, &proof).is_ok()); + + for tampered in [ + rustfs_protos::canonical_heal_control_response_body(2, "fingerprint", b"cancel", b"result").unwrap(), + rustfs_protos::canonical_heal_control_response_body(2, "fingerprint", b"query", b"tampered").unwrap(), + ] { + let err = validate_heal_control_response_proof(&tampered, &proof) + .expect_err("proof must not authenticate a different command or result"); + assert!(err.to_string().contains("invalid heal control response proof")); + } + } + #[tokio::test] async fn peer_rest_client_prepare_retry_clears_offline_gate() { // finalize_result sets the offline gate on a network error; without diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 317dc5950..9c6385aa7 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -34,6 +34,7 @@ const LOG_SUBSYSTEM_CHANNEL: &str = "channel"; const EVENT_HEAL_CHANNEL_STATE: &str = "heal_channel_state"; const EVENT_HEAL_CHANNEL_REQUEST: &str = "heal_channel_request"; const EVENT_HEAL_CHANNEL_RESPONSE: &str = "heal_channel_response"; +const MAX_HEAL_STATUS_PAYLOAD_SIZE: usize = 8 * 1024 * 1024; fn admission_response(request_id: String, admission: HealAdmissionResult) -> HealChannelResponse { let (success, error) = match admission { @@ -61,11 +62,56 @@ pub struct HealChannelProcessor { } #[derive(Serialize)] -struct HealTaskStatusPayload { - summary: String, - items: Vec, +struct HealTaskStatusPayload<'a> { + summary: &'a str, + items: &'a [HealResultItem], + truncated: bool, #[serde(skip_serializing_if = "Option::is_none")] - progress: Option, + progress: Option<&'a HealProgress>, +} + +fn encode_heal_task_status_payload( + summary: &str, + mut items: Vec, + progress: Option<&HealProgress>, + mut truncated: bool, +) -> Result<(Vec, bool)> { + loop { + let data = serde_json::to_vec(&HealTaskStatusPayload { + summary, + items: &items, + truncated, + progress, + }) + .map_err(|e| Error::Serialization(format!("failed to serialize heal task status: {e}")))?; + if data.len() <= MAX_HEAL_STATUS_PAYLOAD_SIZE { + return Ok((data, truncated)); + } + if items.is_empty() { + return Err(Error::Serialization("heal task status metadata exceeds size limit".to_string())); + } + truncated = true; + items.truncate(items.len() / 2); + } +} + +fn heal_status_detail(detail: Option, truncated: bool) -> Option { + if !truncated { + return detail; + } + let truncation = "heal result items were truncated"; + Some(detail.map_or_else(|| truncation.to_string(), |detail| format!("{detail}; {truncation}"))) +} + +fn encode_heal_status_response( + summary: &str, + items: Vec, + progress: Option<&HealProgress>, + detail: Option, + truncated: bool, +) -> Result<(Vec, Option)> { + let (data, truncated) = encode_heal_task_status_payload(summary, items, progress, truncated)?; + Ok((data, heal_status_detail(detail, truncated))) } impl HealChannelProcessor { @@ -79,6 +125,37 @@ impl HealChannelProcessor { } } + /// Execute a start directly against the manager without entering the + /// process-global unbounded command queue. + pub async fn execute_start_request(&self, request: HealChannelRequest) -> Result { + let (response_tx, response_rx) = oneshot::channel(); + self.process_start_request(request, false, true, response_tx).await?; + response_rx + .await + .map_err(|err| Error::other(format!("heal receipt channel closed: {err}")))? + .map_err(Error::other) + } + + /// Execute a token query directly against the manager. + pub async fn execute_query_request(&self, heal_path: String, client_token: String) -> Result { + let (response_tx, response_rx) = oneshot::channel(); + self.process_query_request(heal_path, client_token, response_tx).await?; + response_rx + .await + .map_err(|err| Error::other(format!("heal query channel closed: {err}")))? + .map_err(Error::other) + } + + /// Execute cancellation directly against the manager. + pub async fn execute_cancel_request(&self, heal_path: String, client_token: String) -> Result { + let (response_tx, response_rx) = oneshot::channel(); + self.process_cancel_request(heal_path, client_token, response_tx).await?; + response_rx + .await + .map_err(|err| Error::other(format!("heal cancel channel closed: {err}")))? + .map_err(Error::other) + } + /// Start processing legacy heal channel requests. pub async fn start(&mut self, receiver: HealChannelReceiver) -> Result<()> { let (receipt_sender, receipt_receiver) = mpsc::unbounded_channel(); @@ -326,46 +403,66 @@ impl HealChannelProcessor { self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await }; - let (summary, detail, items, progress) = match report { + let (summary, detail, items, truncated, progress) = match report { Ok(HealTaskReport { status: HealTaskStatus::Pending | HealTaskStatus::Running, result_items, + result_items_truncated, progress, - }) => ("running".to_string(), None, result_items, progress), + }) => ("running".to_string(), None, result_items, result_items_truncated, progress), Ok(HealTaskReport { status: HealTaskStatus::Retrying { error, retry_attempt }, result_items, + result_items_truncated, progress, }) => ( "running".to_string(), Some(format!("heal task retrying after recoverable failure, attempt {retry_attempt}: {error}")), result_items, + result_items_truncated, progress, ), Ok(HealTaskReport { status: HealTaskStatus::Completed, result_items, + result_items_truncated, progress, - }) => ("finished".to_string(), None, result_items, progress), + }) => ("finished".to_string(), None, result_items, result_items_truncated, progress), Ok(HealTaskReport { status: HealTaskStatus::Cancelled, result_items, + result_items_truncated, progress, - }) => ("stopped".to_string(), Some("heal task cancelled".to_string()), result_items, progress), + }) => ( + "stopped".to_string(), + Some("heal task cancelled".to_string()), + result_items, + result_items_truncated, + progress, + ), Ok(HealTaskReport { status: HealTaskStatus::Timeout, result_items, + result_items_truncated, progress, - }) => ("stopped".to_string(), Some("heal task timed out".to_string()), result_items, progress), + }) => ( + "stopped".to_string(), + Some("heal task timed out".to_string()), + result_items, + result_items_truncated, + progress, + ), Ok(HealTaskReport { status: HealTaskStatus::Failed { error }, result_items, + result_items_truncated, progress, - }) => ("stopped".to_string(), Some(error), result_items, progress), + }) => ("stopped".to_string(), Some(error), result_items, result_items_truncated, progress), Err(crate::Error::TaskNotFound { .. }) => ( "notFound".to_string(), Some("heal task not found or expired".to_string()), Vec::new(), + false, None, ), Err(crate::Error::InvalidClientToken) => { @@ -393,12 +490,7 @@ impl HealChannelProcessor { } }; - let data = serde_json::to_vec(&HealTaskStatusPayload { - summary, - items, - progress, - }) - .map_err(|e| crate::Error::Serialization(format!("failed to serialize heal task status: {e}")))?; + let (data, detail) = encode_heal_status_response(&summary, items, progress.as_ref(), detail, truncated)?; let response = HealChannelResponse { request_id: client_token, @@ -697,6 +789,22 @@ mod tests { // If we can get the sender, processor was created correctly } + #[test] + fn oversized_status_items_are_truncated_before_transport() { + let items = vec![HealResultItem { + detail: "x".repeat(MAX_HEAL_STATUS_PAYLOAD_SIZE + 1), + ..Default::default() + }]; + + let (data, detail) = encode_heal_status_response("running", items, None, None, false).unwrap(); + + assert!(data.len() <= MAX_HEAL_STATUS_PAYLOAD_SIZE); + let payload: serde_json::Value = serde_json::from_slice(&data).unwrap(); + assert_eq!(payload["truncated"], true); + assert!(payload["items"].as_array().unwrap().is_empty()); + assert_eq!(detail.as_deref(), Some("heal result items were truncated")); + } + #[test] fn admission_response_preserves_all_admission_outcomes() { let cases = [ @@ -1339,6 +1447,70 @@ mod tests { .expect("receipt processor should stop cleanly"); } + #[tokio::test] + async fn direct_control_execution_preserves_target_dedup_and_token_ownership() { + let manager = create_test_heal_manager(); + let processor = HealChannelProcessor::new(manager); + let original_id = uuid::Uuid::new_v4().to_string(); + let request = HealChannelRequest { + id: original_id.clone(), + bucket: "bucket-a".to_string(), + object_prefix: Some("object".to_string()), + priority: HealChannelPriority::High, + source: HealRequestSource::Admin, + ..Default::default() + }; + + let accepted = processor + .execute_start_request(request.clone()) + .await + .expect("first direct start should be admitted"); + assert_eq!(accepted.result, HealAdmissionResult::Accepted); + assert_eq!(accepted.task_id, original_id); + + let mut duplicate = request; + duplicate.id = uuid::Uuid::new_v4().to_string(); + let merged = processor + .execute_start_request(duplicate) + .await + .expect("duplicate direct start should merge"); + assert_eq!(merged.result, HealAdmissionResult::Merged); + assert_eq!(merged.task_id, accepted.task_id); + + let other = processor + .execute_start_request(HealChannelRequest { + id: uuid::Uuid::new_v4().to_string(), + bucket: "bucket-b".to_string(), + object_prefix: Some("object".to_string()), + priority: HealChannelPriority::High, + source: HealRequestSource::Admin, + ..Default::default() + }) + .await + .expect("different target should be admitted independently"); + assert_eq!(other.result, HealAdmissionResult::Accepted); + assert_ne!(other.task_id, accepted.task_id); + + let status = processor + .execute_query_request(String::new(), accepted.task_id.clone()) + .await + .expect("canonical token should be queryable"); + assert!(status.success); + let cancelled = processor + .execute_cancel_request(String::new(), accepted.task_id.clone()) + .await + .expect("canonical token should be cancellable"); + assert!(cancelled.success); + let stopped = processor + .execute_query_request(String::new(), accepted.task_id) + .await + .expect("cancelled task status should remain queryable"); + assert!(stopped.success); + assert_eq!(stopped.error.as_deref(), Some("heal task not found or expired")); + let status: serde_json::Value = serde_json::from_slice(stopped.data.as_deref().unwrap()).unwrap(); + assert_eq!(status["summary"], "notFound"); + } + #[tokio::test] async fn test_process_start_request_returns_error_on_invalid_request() { let heal_manager = create_test_heal_manager(); diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 1f1ab81e0..f1ff61beb 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -179,6 +179,7 @@ struct CompletedHealStatus { heal_type: HealType, status: HealTaskStatus, result_items: Vec, + result_items_truncated: bool, completed_at: SystemTime, } @@ -198,6 +199,7 @@ struct RetryingHeal { pub struct HealTaskReport { pub status: HealTaskStatus, pub result_items: Vec, + pub result_items_truncated: bool, pub progress: Option, } @@ -1600,6 +1602,7 @@ impl HealManager { return Ok(HealTaskReport { status: task.get_status().await, result_items: task.get_result_items().await, + result_items_truncated: task.result_items_truncated(), progress: Some(task.get_progress().await), }); } @@ -1611,6 +1614,7 @@ impl HealManager { return Ok(HealTaskReport { status: retrying.status(), result_items: Vec::new(), + result_items_truncated: false, progress: None, }); } @@ -1625,6 +1629,7 @@ impl HealManager { return Ok(HealTaskReport { status: completed.status.clone(), result_items: completed.result_items.clone(), + result_items_truncated: completed.result_items_truncated, progress: None, }); } @@ -1636,6 +1641,7 @@ impl HealManager { return Ok(HealTaskReport { status: HealTaskStatus::Pending, result_items: Vec::new(), + result_items_truncated: false, progress: None, }); } @@ -1647,6 +1653,7 @@ impl HealManager { return Ok(HealTaskReport { status: completed.status.clone(), result_items: completed.result_items.clone(), + result_items_truncated: completed.result_items_truncated, progress: None, }); } @@ -1666,6 +1673,7 @@ impl HealManager { return Ok(HealTaskReport { status: task.get_status().await, result_items: task.get_result_items().await, + result_items_truncated: task.result_items_truncated(), progress: Some(task.get_progress().await), }); } @@ -1679,6 +1687,7 @@ impl HealManager { return Ok(HealTaskReport { status: retrying.status(), result_items: Vec::new(), + result_items_truncated: false, progress: None, }); } @@ -1694,6 +1703,7 @@ impl HealManager { return Ok(HealTaskReport { status: completed.status.clone(), result_items: completed.result_items.clone(), + result_items_truncated: completed.result_items_truncated, progress: None, }); } @@ -1705,6 +1715,7 @@ impl HealManager { return Ok(HealTaskReport { status: HealTaskStatus::Pending, result_items: Vec::new(), + result_items_truncated: false, progress: None, }); } @@ -1719,6 +1730,7 @@ impl HealManager { return Ok(HealTaskReport { status: completed.status.clone(), result_items: completed.result_items.clone(), + result_items_truncated: completed.result_items_truncated, progress: None, }); } @@ -2618,6 +2630,7 @@ impl HealManager { heal_type: completed_task.heal_type.clone(), status: completed_status.clone(), result_items: completed_task.get_result_items().await, + result_items_truncated: completed_task.result_items_truncated(), completed_at: SystemTime::now(), }; let mut completed_heals_guard = completed_heals_clone.lock().await; @@ -3848,6 +3861,7 @@ mod tests { retry_attempt: request.retry_attempts, }, result_items: Vec::new(), + result_items_truncated: false, completed_at: SystemTime::now(), }, ); @@ -4411,6 +4425,7 @@ mod tests { }, status: HealTaskStatus::Completed, result_items: Vec::new(), + result_items_truncated: false, completed_at: SystemTime::now(), }, ); @@ -4444,6 +4459,7 @@ mod tests { object_size: 1024, ..Default::default() }], + result_items_truncated: true, completed_at: SystemTime::now(), }, ); @@ -4452,6 +4468,7 @@ mod tests { .get_task_report_for_path("bucket/object", "completed-token") .await .expect("recent completed task report should be queryable"); + assert!(report.result_items_truncated); assert_eq!(report.status, HealTaskStatus::Completed); assert_eq!(report.result_items.len(), 1); diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index f64b2a489..dbd505bb3 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -43,6 +43,7 @@ const LOG_SUBSYSTEM_OBJECT: &str = "object"; const EVENT_HEAL_TASK_STATE: &str = "heal_task_state"; const EVENT_HEAL_OBJECT_STAGE: &str = "heal_object_stage"; const EVENT_HEAL_OBJECT_MISSING: &str = "heal_object_missing"; +const MAX_RETAINED_HEAL_RESULT_ITEMS: usize = 1024; const EVENT_HEAL_OBJECT_RESULT: &str = "heal_object_result"; const MAX_BUCKET_OBJECT_HEAL_RETRIES: u32 = 3; const MAX_BUCKET_FAILURE_LOG_SAMPLES: u64 = 5; @@ -306,6 +307,7 @@ pub struct HealTask { pub progress: Arc>, /// Result items collected from storage heal calls. pub result_items: Arc>>, + result_items_truncated: Arc, batch_failure: Arc>>, batch_failure_recorded: Arc, /// Created time @@ -337,6 +339,7 @@ impl HealTask { status: Arc::new(RwLock::new(HealTaskStatus::Pending)), progress: Arc::new(RwLock::new(HealProgress::new())), result_items: Arc::new(RwLock::new(Vec::new())), + result_items_truncated: Arc::new(AtomicBool::new(false)), batch_failure: Arc::new(RwLock::new(None)), batch_failure_recorded: Arc::new(AtomicBool::new(false)), created_at: request.created_at, @@ -741,8 +744,17 @@ impl HealTask { self.result_items.read().await.clone() } + pub fn result_items_truncated(&self) -> bool { + self.result_items_truncated.load(Ordering::Relaxed) + } + async fn record_result_item(&self, result: HealResultItem) { - self.result_items.write().await.push(result); + let mut result_items = self.result_items.write().await; + if result_items.len() < MAX_RETAINED_HEAL_RESULT_ITEMS { + result_items.push(result); + } else { + self.result_items_truncated.store(true, Ordering::Relaxed); + } } // specific heal implementation method @@ -2656,6 +2668,19 @@ mod tests { assert_eq!(result_items.iter().filter(|item| item.object_size == 1).count(), 2); } + #[tokio::test] + async fn result_items_are_bounded_and_report_truncation() { + let storage = Arc::new(MockStorage::default()); + let task = HealTask::from_request(HealRequest::bucket("bucket-a".to_string()), storage); + + for _ in 0..=MAX_RETAINED_HEAL_RESULT_ITEMS { + task.record_result_item(HealResultItem::default()).await; + } + + assert_eq!(task.get_result_items().await.len(), MAX_RETAINED_HEAL_RESULT_ITEMS); + assert!(task.result_items_truncated()); + } + #[tokio::test] async fn test_recursive_bucket_heal_skips_object_dir_candidates() { let storage = Arc::new(MockStorage { diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index f637b09b1..0a0f1eb77 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -1114,6 +1114,8 @@ pub struct HealControlResponse { pub result: ::prost::bytes::Bytes, #[prost(string, optional, tag = "3")] pub error_info: ::core::option::Option<::prost::alloc::string::String>, + #[prost(bytes = "bytes", tag = "4")] + pub response_proof: ::prost::bytes::Bytes, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct GetMetacacheListingRequest { diff --git a/crates/protos/src/heal_control.rs b/crates/protos/src/heal_control.rs index 9b09565a0..7847f3bc0 100644 --- a/crates/protos/src/heal_control.rs +++ b/crates/protos/src/heal_control.rs @@ -29,8 +29,10 @@ use std::{fmt, io::Cursor, io::Write}; const ENVELOPE_VERSION: u8 = 1; pub const ENVELOPE_MAX_SIZE: usize = 64 * 1024; +pub const RESULT_MAX_SIZE: usize = 16 * 1024 * 1024; pub const NONCE_SIZE: usize = 16; pub const MAX_LIFETIME_MS: i64 = 30_000; +const MAX_CLOCK_SKEW_MS: i64 = 5_000; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct RequestMetadata { @@ -136,6 +138,38 @@ impl TryFrom for StartCommand { } } +impl StartCommand { + fn into_channel_request(self, request_id: String) -> Result { + Ok(HealChannelRequest { + id: request_id, + disk: self.disk, + bucket: self.bucket, + object_prefix: self.object_prefix, + object_version_id: self.object_version_id, + force_start: self.force_start, + priority: self.priority.into(), + pool_index: self + .pool_index + .map(usize::try_from) + .transpose() + .map_err(|_| "heal pool index exceeds platform range".to_string())?, + set_index: self + .set_index + .map(usize::try_from) + .transpose() + .map_err(|_| "heal set index exceeds platform range".to_string())?, + scan_mode: self.scan_mode, + remove_corrupted: self.remove_corrupted, + recreate_missing: self.recreate_missing, + update_parity: self.update_parity, + recursive: self.recursive, + dry_run: self.dry_run, + timeout_seconds: self.timeout_seconds, + source: self.source, + }) + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "action", rename_all = "snake_case", deny_unknown_fields)] pub enum Command { @@ -144,6 +178,13 @@ pub enum Command { Cancel { heal_path: String, client_token: String }, } +#[derive(Debug)] +pub enum ExecutableCommand { + Start { request: HealChannelRequest }, + Query { heal_path: String, client_token: String }, + Cancel { heal_path: String, client_token: String }, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct Envelope { @@ -214,13 +255,42 @@ impl Envelope { } match &self.command { Command::Start { .. } => {} - Command::Query { client_token, .. } | Command::Cancel { client_token, .. } if client_token.is_empty() => { + Command::Query { client_token, .. } if client_token.is_empty() => { return Err("heal control client token is empty".to_string()); } Command::Query { .. } | Command::Cancel { .. } => {} } Ok(()) } + + pub fn validate_execution(&self, now_unix_ms: i64, expected_coordinator_epoch: u64) -> Result<(), String> { + self.validate()?; + if self.coordinator_epoch != expected_coordinator_epoch { + return Err("heal control coordinator epoch does not match".to_string()); + } + if self.issued_at_unix_ms > now_unix_ms.saturating_add(MAX_CLOCK_SKEW_MS) { + return Err("heal control request was issued in the future".to_string()); + } + if self.expires_at_unix_ms <= now_unix_ms { + return Err("heal control request expired".to_string()); + } + Ok(()) + } + + pub fn into_execution(self) -> Result<(String, u64, ExecutableCommand), String> { + let command = match self.command { + Command::Start { request } => ExecutableCommand::Start { + request: request.into_channel_request(self.request_id.clone())?, + }, + Command::Query { heal_path, client_token } => ExecutableCommand::Query { heal_path, client_token }, + Command::Cancel { heal_path, client_token } => ExecutableCommand::Cancel { heal_path, client_token }, + }; + Ok((self.request_id, self.coordinator_epoch, command)) + } + + pub const fn expires_at_unix_ms(&self) -> i64 { + self.expires_at_unix_ms + } } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -249,6 +319,16 @@ impl Admission { pub const fn is_admitted(self) -> bool { matches!(self, Self::Accepted | Self::Merged) } + + pub const fn into_heal_admission_result(self) -> HealAdmissionResult { + match self { + Self::Accepted => HealAdmissionResult::Accepted, + Self::Merged => HealAdmissionResult::Merged, + Self::Full => HealAdmissionResult::Full, + Self::DroppedQueueFull => HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull), + Self::DroppedPolicy => HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped), + } + } } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -274,20 +354,20 @@ impl<'de> Visitor<'de> for BoundedBytesVisitor { type Value = BoundedBytes; fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(formatter, "at most {ENVELOPE_MAX_SIZE} bytes") + write!(formatter, "at most {RESULT_MAX_SIZE} bytes") } fn visit_seq(self, mut sequence: A) -> Result where A: SeqAccess<'de>, { - if sequence.size_hint().is_some_and(|length| length > ENVELOPE_MAX_SIZE) { + if sequence.size_hint().is_some_and(|length| length > RESULT_MAX_SIZE) { return Err(serde::de::Error::custom("heal control response data exceeds size limit")); } let mut bytes = Vec::new(); while let Some(byte) = sequence.next_element()? { - if bytes.len() == ENVELOPE_MAX_SIZE { + if bytes.len() == RESULT_MAX_SIZE { return Err(serde::de::Error::custom("heal control response data exceeds size limit")); } bytes.push(byte); @@ -358,13 +438,6 @@ impl ResultEnvelope { validate_uuid(&self.request_id, "result request")?; match &self.outcome { Outcome::Start { task_id, .. } => validate_uuid(task_id, "result task")?, - Outcome::Channel { - success: true, - error: Some(_), - .. - } => { - return Err("successful heal control result contains an error".to_string()); - } Outcome::Channel { success: false, error: None, @@ -390,6 +463,17 @@ impl ResultEnvelope { } Ok(()) } + + pub fn into_outcome(self, expected_request_id: &str, expected_epoch: u64) -> Result { + self.validate()?; + if self.request_id != expected_request_id { + return Err("heal control result request ID does not match".to_string()); + } + if self.coordinator_epoch != expected_epoch { + return Err("heal control result coordinator epoch does not match".to_string()); + } + Ok(self.outcome) + } } fn validate_uuid(value: &str, field: &str) -> Result<(), String> { @@ -400,8 +484,8 @@ fn validate_uuid(value: &str, field: &str) -> Result<(), String> { Ok(()) } -fn decode Deserialize<'de>>(data: &[u8], value_name: &str) -> Result { - if data.len() > ENVELOPE_MAX_SIZE { +fn decode Deserialize<'de>>(data: &[u8], value_name: &str, max_size: usize) -> Result { + if data.len() > max_size { return Err(format!("{value_name} exceeds size limit")); } let mut deserializer = Deserializer::new(Cursor::new(data)); @@ -414,35 +498,36 @@ fn decode Deserialize<'de>>(data: &[u8], value_name: &str) -> Result pub fn encode_envelope(envelope: &Envelope) -> Result, String> { envelope.validate()?; - encode_bounded(envelope, "heal control envelope") + encode_bounded(envelope, "heal control envelope", ENVELOPE_MAX_SIZE) } pub fn decode_envelope(data: &[u8]) -> Result { - let envelope: Envelope = decode(data, "heal control envelope")?; + let envelope: Envelope = decode(data, "heal control envelope", ENVELOPE_MAX_SIZE)?; envelope.validate()?; Ok(envelope) } pub fn encode_result(result: &ResultEnvelope) -> Result, String> { result.validate()?; - encode_bounded(result, "heal control result") + encode_bounded(result, "heal control result", RESULT_MAX_SIZE) } pub fn decode_result(data: &[u8]) -> Result { - let result: ResultEnvelope = decode(data, "heal control result")?; + let result: ResultEnvelope = decode(data, "heal control result", RESULT_MAX_SIZE)?; result.validate()?; Ok(result) } -fn encode_bounded(value: &impl Serialize, value_name: &str) -> Result, String> { +fn encode_bounded(value: &impl Serialize, value_name: &str, max_size: usize) -> Result, String> { struct BoundedWriter { bytes: Vec, exceeded: bool, + max_size: usize, } impl Write for BoundedWriter { fn write(&mut self, data: &[u8]) -> std::io::Result { - let remaining = ENVELOPE_MAX_SIZE.saturating_sub(self.bytes.len()); + let remaining = self.max_size.saturating_sub(self.bytes.len()); if data.len() > remaining { self.exceeded = true; return Err(std::io::Error::other("heal control value exceeds size limit")); @@ -459,6 +544,7 @@ fn encode_bounded(value: &impl Serialize, value_name: &str) -> Result, S let mut writer = BoundedWriter { bytes: Vec::with_capacity(1024), exceeded: false, + max_size, }; let result = value.serialize(&mut rmp_serde::Serializer::new(&mut writer).with_struct_map()); if writer.exceeded { @@ -471,7 +557,8 @@ fn encode_bounded(value: &impl Serialize, value_name: &str) -> Result, S #[cfg(test)] mod tests { use super::{ - Admission, ENVELOPE_MAX_SIZE, Envelope, Outcome, RequestMetadata, ResultEnvelope, decode_envelope, decode_result, + Admission, ENVELOPE_MAX_SIZE, Envelope, Outcome, RESULT_MAX_SIZE, RequestMetadata, ResultEnvelope, decode_envelope, + decode_result, encode_result, }; use rustfs_common::heal_channel::{HealChannelRequest, HealChannelResponse, HealRequestSource}; use serde::de::{DeserializeSeed, SeqAccess, Visitor, value::Error as ValueError}; @@ -640,11 +727,7 @@ mod tests { assert!(Envelope::start(test_request(request_id.clone()), metadata(1, 0)).is_err()); assert!(Envelope::start(test_request(request_id.clone()), RequestMetadata::new([1; 16], 1_000, 31_001, 7),).is_err()); assert!(Envelope::query(request_id.clone(), metadata(1, 7), String::new(), String::new()).is_err()); - assert!( - Envelope::cancel(request_id.clone(), metadata(1, 7), String::new(), String::new()) - .unwrap_err() - .contains("token is empty") - ); + assert!(Envelope::cancel(request_id.clone(), metadata(1, 7), String::new(), String::new()).is_ok()); let mut noncanonical_request = test_request(request_id.to_uppercase()); assert!(Envelope::start(noncanonical_request.clone(), metadata(1, 7)).is_err()); @@ -689,6 +772,15 @@ mod tests { let unknown = rmp_serde::to_vec_named(&unknown).unwrap(); assert!(decode_envelope(&unknown).unwrap_err().contains("unknown field")); + let executable = + Envelope::start(test_request(request_id.clone()), RequestMetadata::new([1; 16], 10_000, 20_000, 7)).unwrap(); + assert!(executable.validate_execution(15_000, 7).is_ok()); + assert!(executable.validate_execution(20_000, 7).unwrap_err().contains("expired")); + assert!(executable.validate_execution(4_999, 7).unwrap_err().contains("future")); + let wrong_epoch = + Envelope::start(test_request(request_id.clone()), RequestMetadata::new([1; 16], 10_000, 20_000, 8)).unwrap(); + assert!(wrong_epoch.validate_execution(15_000, 7).unwrap_err().contains("epoch")); + assert!( ResultEnvelope::channel( uuid::Uuid::new_v4().to_string(), @@ -711,11 +803,25 @@ mod tests { Outcome::Channel { success: true, data: None, - error: Some("unexpected".to_string()), + error: Some("status detail".to_string()), }, ) - .is_err() + .is_ok() ); + + let large_result = ResultEnvelope::new( + request_id.clone(), + 7, + Outcome::Channel { + success: true, + data: Some(vec![7; ENVELOPE_MAX_SIZE + 1]), + error: None, + }, + ) + .unwrap(); + let large_result = encode_result(&large_result).expect("status results may exceed the request envelope limit"); + assert!(large_result.len() > ENVELOPE_MAX_SIZE); + assert!(decode_result(&large_result).is_ok()); assert!( ResultEnvelope::new( request_id.clone(), @@ -823,7 +929,7 @@ mod tests { } fn size_hint(&self) -> Option { - Some(ENVELOPE_MAX_SIZE + 1) + Some(RESULT_MAX_SIZE + 1) } } diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 2bf93b030..2cf2db8d3 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -97,10 +97,31 @@ fn internode_http2_keep_alive_timeout() -> Duration { } fn internode_rpc_timeout() -> Duration { - Duration::from_secs(rustfs_utils::get_env_u64( + normalize_internode_rpc_timeout(Duration::from_secs(rustfs_utils::get_env_u64( rustfs_config::ENV_INTERNODE_RPC_TIMEOUT_SECS, rustfs_config::DEFAULT_INTERNODE_RPC_TIMEOUT_SECS, - )) + ))) +} + +fn normalize_internode_rpc_timeout(timeout: Duration) -> Duration { + timeout.max(Duration::from_secs(1)) +} + +/// Budget for one heal-control execution, kept below the transport timeout so +/// the coordinator stops waiting for admission before the caller gives up. +pub fn heal_control_execution_timeout() -> Duration { + heal_control_execution_timeout_for(internode_rpc_timeout()) +} + +fn heal_control_execution_timeout_for(transport_timeout: Duration) -> Duration { + const TRANSPORT_GUARD: Duration = Duration::from_secs(1); + let transport_timeout = transport_timeout.max(Duration::from_secs(1)); + transport_timeout + .saturating_sub(TRANSPORT_GUARD.min(transport_timeout / 2)) + .max(Duration::from_millis(1)) + .min(Duration::from_millis( + u64::try_from(heal_control::MAX_LIFETIME_MS).expect("positive heal control lifetime must fit u64"), + )) } fn internode_rpc_tcp_nodelay() -> bool { @@ -141,9 +162,20 @@ pub fn internode_rpc_max_message_size() -> usize { rustfs_utils::get_env_usize(rustfs_config::ENV_INTERNODE_RPC_MAX_MESSAGE_SIZE, DEFAULT_GRPC_SERVER_MESSAGE_LEN) } -pub const HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE: usize = 65 * 1024; -pub const HEAL_CONTROL_PROTOCOL_VERSION: u32 = 1; -pub const HEAL_CONTROL_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-heal-control-capability-v1\0"; +pub const HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE: usize = heal_control::RESULT_MAX_SIZE + 1024; +pub const HEAL_CONTROL_PROTOCOL_VERSION: u32 = 2; +pub const HEAL_CONTROL_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-heal-control-capability-v2\0"; + +pub fn heal_control_coordinator_epoch(topology_fingerprint: &str) -> Result { + let prefix = topology_fingerprint + .get(..16) + .ok_or("heal control topology fingerprint is too short")?; + let epoch = u64::from_str_radix(prefix, 16).map_err(|_| "heal control topology fingerprint is not hexadecimal")?; + if epoch == 0 { + return Err("heal control topology epoch is zero"); + } + Ok(epoch) +} pub fn heal_control_capability_probe(nonce: &[u8; 16]) -> Vec { let mut probe = Vec::with_capacity(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX.len() + nonce.len()); @@ -165,7 +197,7 @@ pub fn canonical_heal_control_request_body( topology_fingerprint: &str, command: &[u8], ) -> Result, std::num::TryFromIntError> { - const DOMAIN: &[u8] = b"rustfs-heal-control-v1\0"; + const DOMAIN: &[u8] = b"rustfs-heal-control-v2\0"; let fingerprint = topology_fingerprint.as_bytes(); let mut body = Vec::with_capacity(DOMAIN.len() + 4 + 8 + fingerprint.len() + 8 + command.len()); @@ -185,7 +217,7 @@ pub fn canonical_heal_control_capability_ack( topology_fingerprint: &str, probe: &[u8], ) -> Result, std::num::TryFromIntError> { - const DOMAIN: &[u8] = b"rustfs-heal-control-capability-ack-v1\0"; + const DOMAIN: &[u8] = b"rustfs-heal-control-capability-ack-v2\0"; let fingerprint = topology_fingerprint.as_bytes(); let mut body = Vec::with_capacity(DOMAIN.len() + 4 + 8 + fingerprint.len() + 8 + probe.len()); @@ -198,17 +230,42 @@ pub fn canonical_heal_control_capability_ack( Ok(body) } +pub fn canonical_heal_control_response_body( + version: u32, + topology_fingerprint: &str, + command: &[u8], + result: &[u8], +) -> Result, std::num::TryFromIntError> { + const DOMAIN: &[u8] = b"rustfs-heal-control-response-v2\0"; + + let fingerprint = topology_fingerprint.as_bytes(); + let mut body = Vec::with_capacity(DOMAIN.len() + 4 + 8 + fingerprint.len() + 8 + command.len() + 8 + result.len()); + body.extend_from_slice(DOMAIN); + body.extend_from_slice(&version.to_be_bytes()); + body.extend_from_slice(&u64::try_from(fingerprint.len())?.to_be_bytes()); + body.extend_from_slice(fingerprint); + body.extend_from_slice(&u64::try_from(command.len())?.to_be_bytes()); + body.extend_from_slice(command); + body.extend_from_slice(&u64::try_from(result.len())?.to_be_bytes()); + body.extend_from_slice(result); + Ok(body) +} + #[cfg(test)] mod heal_control_tests { use super::{ - HEAL_CONTROL_CAPABILITY_PROBE_PREFIX, canonical_heal_control_capability_ack, canonical_heal_control_request_body, - heal_control_capability_probe, is_heal_control_capability_probe, + HEAL_CONTROL_CAPABILITY_PROBE_PREFIX, HEAL_CONTROL_PROTOCOL_VERSION, canonical_heal_control_capability_ack, + canonical_heal_control_request_body, canonical_heal_control_response_body, heal_control_capability_probe, + heal_control_coordinator_epoch, heal_control_execution_timeout, heal_control_execution_timeout_for, + internode_rpc_timeout, is_heal_control_capability_probe, normalize_internode_rpc_timeout, }; + use crate::heal_control; + use std::time::Duration; #[test] fn canonical_heal_control_body_binds_every_field_and_boundary() { let baseline = canonical_heal_control_request_body(1, "ab", b"c").expect("small request should encode"); - let mut golden = b"rustfs-heal-control-v1\0".to_vec(); + let mut golden = b"rustfs-heal-control-v2\0".to_vec(); golden.extend_from_slice(&1_u32.to_be_bytes()); golden.extend_from_slice(&2_u64.to_be_bytes()); golden.extend_from_slice(b"ab"); @@ -236,9 +293,11 @@ mod heal_control_tests { #[test] fn canonical_capability_ack_binds_version_and_topology() { + assert_eq!(HEAL_CONTROL_PROTOCOL_VERSION, 2); + assert!(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX.starts_with(b"rustfs-heal-control-capability-v2")); let probe = heal_control_capability_probe(&[7; 16]); let ack = canonical_heal_control_capability_ack(1, "ab", &probe).expect("small acknowledgement should encode"); - let mut golden = b"rustfs-heal-control-capability-ack-v1\0".to_vec(); + let mut golden = b"rustfs-heal-control-capability-ack-v2\0".to_vec(); golden.extend_from_slice(&1_u32.to_be_bytes()); golden.extend_from_slice(&2_u64.to_be_bytes()); golden.extend_from_slice(b"ab"); @@ -254,6 +313,54 @@ mod heal_control_tests { assert!(is_heal_control_capability_probe(&probe)); assert!(!is_heal_control_capability_probe(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX)); } + + #[test] + fn canonical_response_binds_request_and_result() { + let baseline = canonical_heal_control_response_body(2, "abcdef", b"query", b"result").unwrap(); + assert_ne!(baseline, canonical_heal_control_response_body(1, "abcdef", b"query", b"result").unwrap()); + assert_ne!(baseline, canonical_heal_control_response_body(2, "bbcdef", b"query", b"result").unwrap()); + assert_ne!(baseline, canonical_heal_control_response_body(2, "abcdef", b"cancel", b"result").unwrap()); + assert_ne!( + baseline, + canonical_heal_control_response_body(2, "abcdef", b"query", b"tampered").unwrap() + ); + } + + #[test] + fn coordinator_epoch_is_stable_and_rejects_invalid_fingerprints() { + assert_eq!(heal_control_coordinator_epoch("0123456789abcdefextra"), Ok(0x0123_4567_89ab_cdef)); + assert_eq!( + heal_control_coordinator_epoch("0000000000000000"), + Err("heal control topology epoch is zero") + ); + assert_eq!( + heal_control_coordinator_epoch("short"), + Err("heal control topology fingerprint is too short") + ); + assert_eq!( + heal_control_coordinator_epoch("not-hex-value!!!!"), + Err("heal control topology fingerprint is not hexadecimal") + ); + } + + #[test] + fn execution_budget_precedes_transport_timeout() { + let execution = heal_control_execution_timeout(); + assert!(!execution.is_zero()); + assert!(execution < internode_rpc_timeout()); + assert!(execution <= std::time::Duration::from_millis(heal_control::MAX_LIFETIME_MS as u64)); + } + + #[test] + fn execution_budget_is_nonzero_for_zero_transport_configuration() { + let normalized_transport = Duration::from_secs(1); + assert_eq!(normalize_internode_rpc_timeout(Duration::ZERO), normalized_transport); + for configured_transport in [Duration::ZERO, normalized_transport] { + let execution = heal_control_execution_timeout_for(configured_transport); + assert!(execution > Duration::ZERO); + assert!(execution < normalized_transport); + } + } } /// Whether internode metadata RPCs should send only the msgpack `_bin` payloads and leave the JSON diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index 694c6027f..14b5eb808 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -783,6 +783,7 @@ message HealControlResponse { bool success = 1; bytes result = 2; optional string error_info = 3; + bytes response_proof = 4; } message GetMetacacheListingRequest { diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index ac6316217..70e45fdd4 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -15,21 +15,23 @@ use crate::admin::auth::{authenticate_request, validate_admin_request}; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::runtime_sources::{app_context_from_req, object_store_from_extensions}; -use crate::admin::storage_api::access::spawn_traced; use crate::admin::storage_api::bucket::is_reserved_or_invalid_bucket; use crate::admin::storage_api::bucket::utils::is_valid_object_prefix; use crate::admin::storage_api::contract::heal::HealOperations as _; use crate::server::ADMIN_PREFIX; use crate::server::RemoteAddr; use crate::storage::rpc::node_service::heal::{ - NodeHealProgress, NodeHealStatusSnapshot, capture_node_heal_status, decode_node_heal_status, + HealControlCoordinator, NodeHealProgress, NodeHealStatusSnapshot, capture_node_heal_status, decode_node_heal_status, + heal_control_coordinator, heal_topology_fingerprint, }; use bytes::Bytes; use futures_util::future::join_all; use http::{HeaderMap, HeaderValue, Uri}; use hyper::{Method, StatusCode}; use matchit::Params; -use rustfs_common::heal_channel::{HealChannelPriority, HealChannelRequest, HealOpts, HealRequestSource, HealScanMode}; +use rustfs_common::heal_channel::{ + HealAdmissionReceipt, HealChannelPriority, HealChannelRequest, HealOpts, HealRequestSource, HealScanMode, +}; use rustfs_config::MAX_HEAL_REQUEST_SIZE; use rustfs_heal::heal::utils::format_set_disk_id; use rustfs_policy::policy::action::{Action, AdminAction}; @@ -39,9 +41,10 @@ use s3s::header::{CONTENT_LENGTH, CONTENT_TYPE}; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; use std::collections::HashSet; +use std::future::Future; use std::path::PathBuf; +use std::sync::Arc; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; -use tokio::sync::mpsc; use tokio::time::{Duration, timeout}; use tracing::{info, warn}; @@ -175,6 +178,7 @@ pub fn register_heal_route(r: &mut S3Router) -> std::io::Result< Ok(()) } +#[cfg(test)] #[derive(Default)] struct HealResp { resp_bytes: Vec, @@ -200,6 +204,8 @@ struct HealTaskStatus { heal_settings: HealOpts, #[serde(skip_serializing_if = "Vec::is_empty")] items: Vec, + #[serde(skip_serializing_if = "std::ops::Not::not")] + truncated: bool, #[serde(skip_serializing_if = "Option::is_none")] progress: Option, } @@ -479,15 +485,281 @@ async fn read_cluster_heal_status( merge_peer_heal_statuses(snapshots, peer_statuses, expected_nodes) } +fn cluster_heal_control_unavailable(reason: &str) -> s3s::S3Error { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "heal_control", + result = "failed", + reason, + "cluster heal coordination unavailable" + ); + s3_error!(InternalError, "cluster heal coordination unavailable") +} + +struct PreparedHealControlRoute { + remote_grid_hosts: Vec, + fingerprint: String, + coordinator_epoch: u64, + coordinator: HealControlCoordinator, +} + +fn prepare_heal_control_route(context: &crate::admin::runtime_sources::AppContext) -> S3Result { + let endpoints = context + .endpoints() + .handle() + .ok_or_else(|| cluster_heal_control_unavailable("endpoint_topology_unavailable"))?; + let fingerprint = heal_topology_fingerprint(&endpoints) + .map_err(|_| cluster_heal_control_unavailable("topology_fingerprint_unavailable"))?; + let coordinator_epoch = rustfs_protos::heal_control_coordinator_epoch(&fingerprint) + .map_err(|_| cluster_heal_control_unavailable("topology_epoch_unavailable"))?; + let coordinator = + heal_control_coordinator(&endpoints).map_err(|_| cluster_heal_control_unavailable("coordinator_unavailable"))?; + let remote_grid_hosts = endpoints + .get_nodes() + .into_iter() + .filter(|node| !node.is_local) + .map(|node| node.grid_host) + .collect(); + Ok(PreparedHealControlRoute { + remote_grid_hosts, + fingerprint, + coordinator_epoch, + coordinator, + }) +} + +fn new_heal_control_metadata(route: &PreparedHealControlRoute) -> S3Result { + let now = OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000; + let now = i64::try_from(now).map_err(|_| cluster_heal_control_unavailable("clock_out_of_range"))?; + let lifetime_ms = i64::try_from(rustfs_protos::heal_control_execution_timeout().as_millis()) + .map_err(|_| cluster_heal_control_unavailable("execution_timeout_out_of_range"))?; + Ok(rustfs_protos::heal_control::RequestMetadata::new( + *uuid::Uuid::new_v4().as_bytes(), + now, + now.saturating_add(lifetime_ms), + route.coordinator_epoch, + )) +} + +async fn require_cluster_heal_control_capability( + context: &crate::admin::runtime_sources::AppContext, + route: &PreparedHealControlRoute, +) -> S3Result<()> { + if route.remote_grid_hosts.is_empty() { + return Ok(()); + } + let notification_system = context + .notification_system() + .handle() + .ok_or_else(|| cluster_heal_control_unavailable("notification_system_unavailable"))?; + let clients = route + .remote_grid_hosts + .iter() + .map(|grid_host| { + notification_system + .peer_client_for_grid_host(grid_host) + .ok_or_else(|| cluster_heal_control_unavailable("peer_capability_client_unavailable")) + }) + .collect::>>()?; + let results = join_all(clients.into_iter().map(|client| { + let fingerprint = route.fingerprint.clone(); + async move { client.probe_heal_control(fingerprint).await } + })) + .await; + if let Some(err) = results.into_iter().find_map(Result::err) { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "heal_control", + result = "failed", + reason = "cluster_capability_unavailable", + error = %err, + "cluster heal coordination failed" + ); + return Err(cluster_heal_control_unavailable("cluster_capability_unavailable")); + } + Ok(()) +} + +async fn route_cluster_heal_control( + context: &crate::admin::runtime_sources::AppContext, + route: &PreparedHealControlRoute, + envelope: rustfs_protos::heal_control::Envelope, + request_id: &str, + coordinator_capability_verified: bool, +) -> S3Result { + let response = if route.coordinator.is_local { + crate::storage::rpc::node_service::execute_heal_control_envelope(envelope, route.coordinator_epoch) + .await + .map_err(|err| { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "heal_control", + result = "failed", + reason = "local_coordinator_failed", + error = %err, + "cluster heal coordination failed" + ); + cluster_heal_control_unavailable("local_coordinator_failed") + })? + } else { + let notification_system = context + .notification_system() + .handle() + .ok_or_else(|| cluster_heal_control_unavailable("notification_system_unavailable"))?; + let client = notification_system + .peer_client_for_grid_host(&route.coordinator.grid_host) + .ok_or_else(|| cluster_heal_control_unavailable("coordinator_client_unavailable"))?; + if !coordinator_capability_verified { + client.probe_heal_control(route.fingerprint.clone()).await.map_err(|err| { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "heal_control", + result = "failed", + reason = "coordinator_capability_unavailable", + error = %err, + "cluster heal coordination failed" + ); + cluster_heal_control_unavailable("coordinator_capability_unavailable") + })?; + } + let command = rustfs_protos::heal_control::encode_envelope(&envelope) + .map_err(|err| s3_error!(InternalError, "encode heal control request failed: {err}"))?; + client + .heal_control(rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, route.fingerprint.clone(), command) + .await + .map_err(|err| { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "heal_control", + result = "failed", + reason = "coordinator_request_failed", + error = %err, + "cluster heal coordination failed" + ); + cluster_heal_control_unavailable("coordinator_request_failed") + })? + }; + rustfs_protos::heal_control::decode_result(&response) + .and_then(|result| result.into_outcome(request_id, route.coordinator_epoch)) + .map_err(|err| { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "heal_control", + result = "failed", + reason = "coordinator_response_invalid", + error = %err, + "cluster heal coordination failed" + ); + cluster_heal_control_unavailable("coordinator_response_invalid") + }) +} + +async fn execute_after_heal_control_capability(probe: P, execute: E) -> S3Result +where + P: FnOnce() -> PF, + PF: Future>, + E: FnOnce() -> EF, + EF: Future>, +{ + probe().await?; + execute().await +} + +async fn submit_cluster_heal_start( + context: Arc, + hip: &HealInitParams, +) -> S3Result { + let route = prepare_heal_control_route(&context)?; + let result = execute_after_heal_control_capability( + || require_cluster_heal_control_capability(&context, &route), + || async { + let heal_request = build_heal_channel_request(hip); + let request_id = heal_request.id.clone(); + let envelope = rustfs_protos::heal_control::Envelope::start(heal_request, new_heal_control_metadata(&route)?) + .map_err(|err| s3_error!(InternalError, "encode heal control request failed: {err}"))?; + route_cluster_heal_control(&context, &route, envelope, &request_id, true).await + }, + ) + .await?; + match result { + rustfs_protos::heal_control::Outcome::Start { task_id, admission } => Ok(HealAdmissionReceipt { + result: admission.into_heal_admission_result(), + task_id, + }), + rustfs_protos::heal_control::Outcome::Channel { .. } => { + Err(s3_error!(InternalError, "coordinator returned an unexpected heal result")) + } + } +} + +fn reject_heal_admission(result: rustfs_common::heal_channel::HealAdmissionResult) -> s3s::S3Error { + use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult}; + + match result { + HealAdmissionResult::Full | HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull) => s3_error!( + SlowDown, + "heal request not admitted: admission={}, reason={}", + result.result_label(), + result.reason_label() + ), + HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped) => s3_error!( + OperationAborted, + "heal request not admitted: admission={}, reason={}", + result.result_label(), + result.reason_label() + ), + HealAdmissionResult::Accepted | HealAdmissionResult::Merged => { + s3_error!(InternalError, "admitted heal request was rejected unexpectedly") + } + } +} + +async fn submit_cluster_heal_channel_command( + context: Arc, + route: PreparedHealControlRoute, + envelope: rustfs_protos::heal_control::Envelope, + request_id: &str, + response_id: String, +) -> S3Result { + match route_cluster_heal_control(&context, &route, envelope, request_id, false).await? { + rustfs_protos::heal_control::Outcome::Channel { success, data, error } => { + Ok(rustfs_common::heal_channel::HealChannelResponse { + request_id: response_id, + success, + data, + error, + }) + } + rustfs_protos::heal_control::Outcome::Start { .. } => { + Err(s3_error!(InternalError, "coordinator returned an unexpected heal result")) + } + } +} + #[derive(Debug, Deserialize)] struct HealTaskStatusPayload { summary: String, #[serde(default)] items: Vec, #[serde(default)] + truncated: bool, + #[serde(default)] progress: Option, } +#[cfg(test)] fn map_heal_response(result: Option) -> S3Result<(StatusCode, Vec)> { match result { Some(result) => { @@ -528,6 +800,7 @@ fn encode_heal_task_status( failure_detail: String, heal_settings: HealOpts, items: Vec, + truncated: bool, progress: Option, ) -> S3Result> { encode_json(&HealTaskStatus { @@ -536,6 +809,7 @@ fn encode_heal_task_status( start_time: current_rfc3339_time()?, heal_settings, items, + truncated, progress, }) } @@ -577,15 +851,15 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { fn heal_channel_response_status( response: &rustfs_common::heal_channel::HealChannelResponse, -) -> (String, Vec, Option) { +) -> (String, Vec, bool, Option) { let Some(data) = response.data.as_deref() else { - return ("running".to_string(), Vec::new(), None); + return ("running".to_string(), Vec::new(), false, None); }; if let Ok(payload) = serde_json::from_slice::(data) && !payload.summary.is_empty() { - return (payload.summary, payload.items, payload.progress); + return (payload.summary, payload.items, payload.truncated, payload.progress); } let summary = std::str::from_utf8(data) @@ -593,7 +867,7 @@ fn heal_channel_response_status( .filter(|summary| !summary.is_empty()) .unwrap_or("running") .to_string(); - (summary, Vec::new(), None) + (summary, Vec::new(), false, None) } #[cfg(test)] @@ -610,7 +884,7 @@ fn heal_channel_response_items( #[cfg(test)] fn heal_channel_response_progress(response: &rustfs_common::heal_channel::HealChannelResponse) -> Option { - heal_channel_response_status(response).2 + heal_channel_response_status(response).3 } fn encode_background_heal_status( @@ -732,6 +1006,7 @@ impl Operation for HealHandler { .get::>() .and_then(|opt| opt.map(|addr| addr.0.to_string())) .unwrap_or_default(); + let app_context = app_context_from_req(&req); let mut input = req.input; let bytes = match input.store_all_limited(MAX_HEAL_REQUEST_SIZE).await { Ok(b) => b, @@ -804,179 +1079,120 @@ impl Operation for HealHandler { }; let heal_path = path_join(&[PathBuf::from(hip.bucket.clone()), PathBuf::from(hip.obj_prefix.clone())]); - let (tx, mut rx) = mpsc::channel(1); - if !hip.client_token.is_empty() && !hip.force_start && !hip.force_stop { - // Query heal status - let tx_clone = tx.clone(); let heal_path_str = heal_path.to_str().unwrap_or_default().to_string(); let client_token = hip.client_token.clone(); - spawn_traced(async move { - match rustfs_common::heal_channel::query_heal_status(heal_path_str, client_token).await { - Ok(response) if response.success => { - let (summary, items, progress) = heal_channel_response_status(&response); - let resp_bytes = encode_heal_task_status( - summary, - response.error.unwrap_or_default(), - HealOpts::default(), - items, - progress, - ); - match resp_bytes { - Ok(resp_bytes) => { - let _ = tx_clone - .send(HealResp { - resp_bytes, - ..Default::default() - }) - .await; - } - Err(e) => { - let _ = tx_clone - .send(HealResp { - api_err: Some(e.to_string()), - ..Default::default() - }) - .await; - } - } - } - Ok(response) => { - let _ = tx_clone - .send(HealResp { - api_err: Some(response.error.unwrap_or_else(|| "query heal status failed".to_string())), - ..Default::default() - }) - .await; - } - Err(e) => { - let _ = tx_clone - .send(HealResp { - api_err: Some(format!("query heal status failed: {e}")), - ..Default::default() - }) - .await; - } - } - }); + let request_id = uuid::Uuid::new_v4().to_string(); + let context = app_context + .clone() + .ok_or_else(|| cluster_heal_control_unavailable("app_context_unavailable"))?; + let route = prepare_heal_control_route(&context)?; + let envelope = rustfs_protos::heal_control::Envelope::query( + request_id.clone(), + new_heal_control_metadata(&route)?, + heal_path_str, + client_token.clone(), + ) + .map_err(|err| s3_error!(InternalError, "encode heal control query failed: {err}"))?; + let response = submit_cluster_heal_channel_command(context, route, envelope, &request_id, client_token).await?; + if !response.success { + return Err(s3_error!( + InternalError, + "{}", + response.error.unwrap_or_else(|| "query heal status failed".to_string()) + )); + } + let (summary, items, truncated, progress) = heal_channel_response_status(&response); + let body = encode_heal_task_status( + summary, + response.error.unwrap_or_default(), + HealOpts::default(), + items, + truncated, + progress, + )?; + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = response_operation, + result = "success", + status_code = StatusCode::OK.as_u16(), + "admin response emitted" + ); + return Ok(json_response(StatusCode::OK, body)); } else if hip.force_stop { - // Cancel heal task - let tx_clone = tx.clone(); let heal_path_str = heal_path.to_str().unwrap_or_default().to_string(); let client_token = hip.client_token.clone(); - let client_address = client_address.clone(); - let heal_settings = hip.hs; - spawn_traced(async move { - match rustfs_common::heal_channel::cancel_heal_task(heal_path_str, client_token.clone()).await { - Ok(response) if response.success => { - let resp_bytes = if client_token.is_empty() { - encode_heal_start_success(response.request_id, client_address) - } else { - let (summary, items, progress) = heal_channel_response_status(&response); - encode_heal_task_status(summary, response.error.unwrap_or_default(), heal_settings, items, progress) - }; - match resp_bytes { - Ok(resp_bytes) => { - let _ = tx_clone - .send(HealResp { - resp_bytes, - ..Default::default() - }) - .await; - } - Err(e) => { - let _ = tx_clone - .send(HealResp { - api_err: Some(e.to_string()), - ..Default::default() - }) - .await; - } - } - } - Ok(response) => { - let _ = tx_clone - .send(HealResp { - api_err: Some(response.error.unwrap_or_else(|| "cancel heal task failed".to_string())), - ..Default::default() - }) - .await; - } - Err(e) => { - let _ = tx_clone - .send(HealResp { - api_err: Some(format!("cancel heal task failed: {e}")), - ..Default::default() - }) - .await; - } - } - }); + let request_id = uuid::Uuid::new_v4().to_string(); + let context = app_context + .clone() + .ok_or_else(|| cluster_heal_control_unavailable("app_context_unavailable"))?; + let route = prepare_heal_control_route(&context)?; + let envelope = rustfs_protos::heal_control::Envelope::cancel( + request_id.clone(), + new_heal_control_metadata(&route)?, + heal_path_str, + client_token.clone(), + ) + .map_err(|err| s3_error!(InternalError, "encode heal control cancel failed: {err}"))?; + let response = submit_cluster_heal_channel_command( + context, + route, + envelope, + &request_id, + if client_token.is_empty() { + heal_path.to_string_lossy().into_owned() + } else { + client_token.clone() + }, + ) + .await?; + if !response.success { + return Err(s3_error!( + InternalError, + "{}", + response.error.unwrap_or_else(|| "cancel heal task failed".to_string()) + )); + } + let body = if client_token.is_empty() { + encode_heal_start_success(response.request_id, client_address)? + } else { + let (summary, items, truncated, progress) = heal_channel_response_status(&response); + encode_heal_task_status(summary, response.error.unwrap_or_default(), hip.hs, items, truncated, progress)? + }; + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = response_operation, + result = "success", + status_code = StatusCode::OK.as_u16(), + "admin response emitted" + ); + return Ok(json_response(StatusCode::OK, body)); } else if hip.client_token.is_empty() { - // Use new heal channel mechanism - let tx_clone = tx.clone(); - let client_address = client_address.clone(); - spawn_traced(async move { - // Create heal request through channel - let heal_request = build_heal_channel_request(&hip); - match rustfs_common::heal_channel::send_heal_request_with_receipt(heal_request).await { - Ok(receipt) if receipt.result.is_admitted() => { - let resp_bytes = encode_heal_start_success(receipt.task_id, client_address); - match resp_bytes { - Ok(resp_bytes) => { - let _ = tx_clone - .send(HealResp { - resp_bytes, - ..Default::default() - }) - .await; - } - Err(e) => { - let _ = tx_clone - .send(HealResp { - api_err: Some(e.to_string()), - ..Default::default() - }) - .await; - } - } - } - Ok(receipt) => { - let _ = tx_clone - .send(HealResp { - api_err: Some(format!( - "heal request not admitted: admission={}, reason={}", - receipt.result.result_label(), - receipt.result.reason_label() - )), - ..Default::default() - }) - .await; - } - Err(e) => { - // Error - send error response - let _ = tx_clone - .send(HealResp { - api_err: Some(format!("send heal request failed: {e}")), - ..Default::default() - }) - .await; - } - } - }); + let Some(context) = app_context else { + return Err(cluster_heal_control_unavailable("app_context_unavailable")); + }; + let receipt = submit_cluster_heal_start(context, &hip).await?; + if !receipt.result.is_admitted() { + return Err(reject_heal_admission(receipt.result)); + } + let body = encode_heal_start_success(receipt.task_id, client_address)?; + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = response_operation, + result = "success", + status_code = StatusCode::OK.as_u16(), + "admin response emitted" + ); + return Ok(json_response(StatusCode::OK, body)); } - let (status, body) = map_heal_response(rx.recv().await)?; - info!( - event = EVENT_ADMIN_RESPONSE_EMITTED, - component = LOG_COMPONENT_ADMIN_API, - subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, - operation = response_operation, - result = "success", - status_code = status.as_u16(), - "admin response emitted" - ); - Ok(json_response(status, body)) + Err(s3_error!(InvalidRequest, "invalid heal control request")) } } @@ -1037,9 +1253,10 @@ mod tests { use super::{ BackgroundHealProgress, HealInitParams, HealResp, HealRuntimeState, aggregate_cluster_heal_status, background_heal_runtime_state, build_heal_channel_request, encode_background_heal_status, encode_heal_start_success, - encode_heal_task_status, heal_channel_response_items, heal_channel_response_progress, heal_channel_response_summary, - json_response, map_heal_response, map_root_heal_status, merge_peer_heal_statuses, peer_topology_complete, - query_peer_heal_status, should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target, + encode_heal_task_status, execute_after_heal_control_capability, heal_channel_response_items, + heal_channel_response_progress, heal_channel_response_summary, json_response, map_heal_response, map_root_heal_status, + merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status, reject_heal_admission, + should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target, }; use crate::admin::storage_api::error::StorageError; use crate::storage::rpc::node_service::heal::{NodeHealProgress, NodeHealStatusSnapshot}; @@ -1047,17 +1264,61 @@ mod tests { use http::StatusCode; use http::Uri; use matchit::Router; - use rustfs_common::heal_channel::{HealChannelPriority, HealOpts, HealRequestSource, HealScanMode}; + use rustfs_common::heal_channel::{ + HealAdmissionDropReason, HealAdmissionResult, HealChannelPriority, HealOpts, HealRequestSource, HealScanMode, + }; use rustfs_scanner::scanner::BackgroundHealInfo; use s3s::{ S3ErrorCode, header::{CONTENT_LENGTH, CONTENT_TYPE}, }; use serde_json::json; + use std::sync::atomic::{AtomicBool, Ordering}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::sync::mpsc; use tokio::time::Duration; + #[tokio::test] + async fn cluster_capability_gate_runs_before_execution() { + let executed = AtomicBool::new(false); + let rejected = execute_after_heal_control_capability( + || async { Err(super::cluster_heal_control_unavailable("test_capability_failure")) }, + || async { + executed.store(true, Ordering::SeqCst); + Ok(()) + }, + ) + .await; + assert!(rejected.is_err()); + assert!(!executed.load(Ordering::SeqCst)); + + execute_after_heal_control_capability( + || async { Ok(()) }, + || async { + executed.store(true, Ordering::SeqCst); + Ok(()) + }, + ) + .await + .unwrap(); + assert!(executed.load(Ordering::SeqCst)); + } + + #[test] + fn test_reject_heal_admission_preserves_retry_semantics() { + for admission in [ + HealAdmissionResult::Full, + HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull), + ] { + assert_eq!(reject_heal_admission(admission).code(), &S3ErrorCode::SlowDown); + } + assert_eq!( + reject_heal_admission(HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped)).code(), + &S3ErrorCode::OperationAborted + ); + assert_eq!(reject_heal_admission(HealAdmissionResult::Accepted).code(), &S3ErrorCode::InternalError); + } + #[test] fn test_heal_opts_serialization() { // Test that HealOpts can be properly deserialized @@ -1775,6 +2036,7 @@ mod tests { String::new(), HealOpts::default(), Vec::new(), + false, None, ) .expect("status response should serialize"); @@ -1783,11 +2045,28 @@ mod tests { assert_eq!(json["summary"], "Heal status query accepted"); assert_eq!(json["detail"], ""); assert!(json["items"].is_null()); + assert!(json["truncated"].is_null()); assert!(json["settings"].is_object()); let start_time = json["startTime"].as_str().expect("startTime should be a string"); OffsetDateTime::parse(start_time, &Rfc3339).expect("startTime should be RFC3339"); } + #[test] + fn test_encode_heal_task_status_reports_truncated_items() { + let encoded = encode_heal_task_status( + "running".to_string(), + "heal result items were truncated".to_string(), + HealOpts::default(), + Vec::new(), + true, + None, + ) + .expect("truncated status response should serialize"); + let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize"); + + assert_eq!(json["truncated"], true); + } + #[test] fn test_encode_heal_task_status_preserves_progress() { let progress = json!({ @@ -1800,6 +2079,7 @@ mod tests { String::new(), HealOpts::default(), Vec::new(), + false, Some(progress.clone()), ) .expect("status response should serialize"); diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 1654efcf2..1b48e657f 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -2319,7 +2319,7 @@ mod tests { #[tokio::test] #[serial_test::serial] - async fn peer_rest_heal_control_uses_production_auth_and_stays_online_when_disabled() { + async fn peer_rest_heal_control_uses_production_auth_and_keeps_validation_errors_online() { let _ = rustfs_credentials::set_global_rpc_secret("rpc-http-test-secret".to_string()); let listener = match TcpListener::bind("127.0.0.1:0").await { Ok(listener) => listener, @@ -2376,11 +2376,11 @@ mod tests { for _ in 0..2 { let error = client - .heal_control(1, "fingerprint".to_string(), b"query".to_vec()) + .heal_control(rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, "fingerprint".to_string(), b"query".to_vec()) .await - .expect_err("dormant heal control must report disabled"); + .expect_err("a divergent topology must fail closed"); let message = error.to_string(); - assert!(message.contains("routing is not enabled")); + assert!(message.contains("topology does not match")); assert!(!message.contains("temporarily offline")); } client @@ -2395,16 +2395,21 @@ mod tests { assert!(mismatch.to_string().contains("topology does not match")); assert!(!mismatch.to_string().contains("temporarily offline")); - let unsupported = client - .heal_control( - rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION + 1, - fingerprint.clone(), - rustfs_protos::heal_control_capability_probe(&[9; 16]), - ) - .await - .expect_err("an unsupported protocol version must fail closed"); - assert!(unsupported.to_string().contains("unsupported heal control protocol version")); - assert!(!unsupported.to_string().contains("temporarily offline")); + for unsupported_version in [ + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION - 1, + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION + 1, + ] { + let unsupported = client + .heal_control( + unsupported_version, + fingerprint.clone(), + rustfs_protos::heal_control_capability_probe(&[9; 16]), + ) + .await + .expect_err("an unsupported protocol version must fail closed"); + assert!(unsupported.to_string().contains("unsupported heal control protocol version")); + assert!(!unsupported.to_string().contains("temporarily offline")); + } client .probe_heal_control(fingerprint) diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 745f04933..da706455d 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -39,9 +39,17 @@ use rustfs_protos::{ proto_gen::node_service::{node_service_server::NodeService as Node, *}, }; use serde::Deserialize; -use std::{collections::HashMap, io::Cursor, pin::Pin, sync::Arc}; +use sha2::{Digest, Sha256}; +use std::{ + collections::HashMap, + io::Cursor, + pin::Pin, + sync::{Arc, OnceLock}, +}; +use time::OffsetDateTime; use tokio::spawn; use tokio::sync::mpsc; +use tokio::time::{Duration, timeout}; use tokio_stream::wrappers::ReceiverStream; use tokio_util::sync::CancellationToken; use tonic::{Request, Response, Status, Streaming}; @@ -59,6 +67,95 @@ const EVENT_RPC_REQUEST_FAILED: &str = "rpc_request_failed"; const EVENT_RPC_RESPONSE_EMITTED: &str = "rpc_response_emitted"; const EVENT_RPC_BACKGROUND_TASK_SPAWNED: &str = "rpc_background_task_spawned"; const EVENT_RPC_BACKGROUND_TASK_FAILED: &str = "rpc_background_task_failed"; +const HEAL_CONTROL_REPLAY_CACHE_MAX_ENTRIES: usize = 4096; + +#[derive(Debug)] +struct HealControlReplayEntry { + command_digest: [u8; 32], + expires_at_unix_ms: i64, + result: tokio::sync::Mutex>>, +} + +fn remove_heal_control_replay( + replay_cache: &mut HashMap>, + request_id: &str, + replay_entry: &Arc, +) { + if replay_cache + .get(request_id) + .is_some_and(|cached| Arc::ptr_eq(cached, replay_entry)) + { + replay_cache.remove(request_id); + } +} + +static HEAL_CONTROL_REPLAY_CACHE: OnceLock>>> = OnceLock::new(); + +fn admit_heal_control_replay( + replay_cache: &mut HashMap>, + request_id: &str, + command_digest: &[u8; 32], + expires_at_unix_ms: i64, + now_unix_ms: i64, +) -> Result, Status> { + replay_cache.retain(|_, entry| entry.expires_at_unix_ms > now_unix_ms || Arc::strong_count(entry) > 1); + if let Some(cached) = replay_cache.get(request_id) { + if &cached.command_digest != command_digest { + return Err(Status::already_exists("heal control request ID was reused with a different command")); + } + return Ok(Arc::clone(cached)); + } + if replay_cache.len() >= HEAL_CONTROL_REPLAY_CACHE_MAX_ENTRIES { + return Err(Status::resource_exhausted("heal control replay cache is full")); + } + let entry = Arc::new(HealControlReplayEntry { + command_digest: *command_digest, + expires_at_unix_ms, + result: tokio::sync::Mutex::new(None), + }); + replay_cache.insert(request_id.to_string(), Arc::clone(&entry)); + Ok(entry) +} + +fn heal_control_now_unix_ms() -> Result { + i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000) + .map_err(|_| Status::internal("heal control clock is out of range")) +} + +fn heal_control_remaining(expires_at_unix_ms: i64, now_unix_ms: i64) -> Result { + let remaining_ms = expires_at_unix_ms.saturating_sub(now_unix_ms); + let remaining_ms = u64::try_from(remaining_ms).map_err(|_| Status::deadline_exceeded("heal control request expired"))?; + if remaining_ms == 0 { + return Err(Status::deadline_exceeded("heal control request expired")); + } + Ok(Duration::from_millis(remaining_ms)) +} + +fn validate_admin_heal_control_start(request: &rustfs_common::heal_channel::HealChannelRequest) -> Result<(), Status> { + if request.source != rustfs_common::heal_channel::HealRequestSource::Admin { + return Err(Status::permission_denied("heal control start source must be admin")); + } + if request.pool_index.is_some() != request.set_index.is_some() { + return Err(Status::invalid_argument("heal control start requires both pool and set")); + } + if request.bucket.is_empty() { + if request.object_prefix.as_deref().is_some_and(|prefix| !prefix.is_empty()) { + return Err(Status::invalid_argument("root heal control start cannot contain an object prefix")); + } + if request.recursive != Some(true) { + return Err(Status::invalid_argument("root heal control start must be recursive")); + } + let erasure_set_target = request.pool_index.is_some(); + if request.disk.is_some() != erasure_set_target { + return Err(Status::invalid_argument("root erasure-set heal control target is inconsistent")); + } + } else if request.disk.is_some() { + return Err(Status::invalid_argument( + "bucket heal control start cannot contain an erasure-set disk target", + )); + } + Ok(()) +} fn scanner_activity_response(namespace_generation: u64) -> ScannerActivityResponse { ScannerActivityResponse { @@ -265,12 +362,17 @@ impl HealControlRpcService { Err(Status::failed_precondition("heal control topology is not initialized")) } - #[cfg(test)] async fn endpoint_pools(&self) -> Option { + #[cfg(test)] if let Some(source) = self.endpoint_pools_source.as_ref() { return source.read().await.clone(); } - self.endpoint_pools.clone() + #[cfg(test)] + if self.endpoint_pools.is_some() { + return self.endpoint_pools.clone(); + } + let context = runtime_sources::current_app_context()?; + context.endpoints().handle() } } @@ -288,6 +390,100 @@ pub(crate) async fn initialize_heal_topology_fingerprint( Ok(()) } +pub(crate) async fn execute_heal_control_envelope( + envelope: rustfs_protos::heal_control::Envelope, + expected_coordinator_epoch: u64, +) -> Result, Status> { + execute_heal_control_envelope_with_manager(envelope, expected_coordinator_epoch, None).await +} + +async fn execute_heal_control_envelope_with_manager( + envelope: rustfs_protos::heal_control::Envelope, + expected_coordinator_epoch: u64, + manager: Option>, +) -> Result, Status> { + let now = heal_control_now_unix_ms()?; + envelope + .validate_execution(now, expected_coordinator_epoch) + .map_err(Status::failed_precondition)?; + let expires_at_unix_ms = envelope.expires_at_unix_ms(); + let canonical_envelope = rustfs_protos::heal_control::encode_envelope(&envelope).map_err(Status::invalid_argument)?; + let command_digest = Sha256::digest(&canonical_envelope).into(); + let (request_id, coordinator_epoch, command) = envelope.into_execution().map_err(Status::invalid_argument)?; + + let replay_cache = HEAL_CONTROL_REPLAY_CACHE.get_or_init(|| tokio::sync::Mutex::new(HashMap::new())); + let replay_entry = { + let mut replay_cache = timeout(heal_control_remaining(expires_at_unix_ms, now)?, replay_cache.lock()) + .await + .map_err(|_| Status::deadline_exceeded("heal control request expired while awaiting replay admission"))?; + admit_heal_control_replay(&mut replay_cache, &request_id, &command_digest, expires_at_unix_ms, now)? + }; + let mut replay_result = timeout(heal_control_remaining(expires_at_unix_ms, now)?, replay_entry.result.lock()) + .await + .map_err(|_| Status::deadline_exceeded("heal control request expired while awaiting matching execution"))?; + let now = heal_control_now_unix_ms()?; + if expires_at_unix_ms <= now { + return Err(Status::deadline_exceeded("heal control request expired before execution")); + } + if let Some(cached) = replay_result.as_ref() { + return Ok(cached.clone()); + } + + if let rustfs_protos::heal_control::ExecutableCommand::Start { request } = &command { + validate_admin_heal_control_start(request)?; + } + let retain_completed_result = !matches!(&command, rustfs_protos::heal_control::ExecutableCommand::Query { .. }); + + let manager = manager + .or_else(|| rustfs_heal::get_heal_manager().cloned()) + .ok_or_else(|| Status::failed_precondition("heal manager is not initialized"))?; + let processor = rustfs_heal::HealChannelProcessor::new(manager.clone()); + let remaining = heal_control_remaining(expires_at_unix_ms, now)?; + let outcome = match command { + rustfs_protos::heal_control::ExecutableCommand::Start { request } => { + let receipt = timeout(remaining, processor.execute_start_request(request)) + .await + .map_err(|_| Status::deadline_exceeded("heal control start expired before admission"))? + .map_err(|_| Status::internal("heal control start admission failed"))?; + rustfs_protos::heal_control::Outcome::Start { + task_id: receipt.task_id, + admission: receipt.result.into(), + } + } + rustfs_protos::heal_control::ExecutableCommand::Query { heal_path, client_token } => { + let response = timeout(remaining, processor.execute_query_request(heal_path, client_token)) + .await + .map_err(|_| Status::deadline_exceeded("heal control query expired before execution"))? + .map_err(|_| Status::internal("heal control query failed"))?; + rustfs_protos::heal_control::Outcome::Channel { + success: response.success, + data: response.data, + error: response.error, + } + } + rustfs_protos::heal_control::ExecutableCommand::Cancel { heal_path, client_token } => { + let response = timeout(remaining, processor.execute_cancel_request(heal_path, client_token)) + .await + .map_err(|_| Status::deadline_exceeded("heal control cancel expired before execution"))? + .map_err(|_| Status::internal("heal control cancel failed"))?; + rustfs_protos::heal_control::Outcome::Channel { + success: response.success, + data: response.data, + error: response.error, + } + } + }; + let result = rustfs_protos::heal_control::ResultEnvelope::new(request_id.clone(), coordinator_epoch, outcome) + .and_then(|result| rustfs_protos::heal_control::encode_result(&result)) + .map_err(Status::internal)?; + *replay_result = Some(result.clone()); + if !retain_completed_result { + let mut replay_cache = replay_cache.lock().await; + remove_heal_control_replay(&mut replay_cache, &request_id, &replay_entry); + } + Ok(result) +} + #[tonic::async_trait] impl heal_control_service_server::HealControlService for HealControlRpcService { async fn heal_control(&self, request: Request) -> Result, Status> { @@ -307,11 +503,11 @@ impl heal_control_service_server::HealControlService for HealControlRpcService { if request.get_ref().version != rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION { return Err(Status::failed_precondition("unsupported heal control protocol version")); } + let fingerprint = self.capability_fingerprint().await?; + if request.get_ref().topology_fingerprint != *fingerprint { + return Err(Status::failed_precondition("heal control topology does not match")); + } if rustfs_protos::is_heal_control_capability_probe(&request.get_ref().command) { - let fingerprint = self.capability_fingerprint().await?; - if request.get_ref().topology_fingerprint != *fingerprint { - return Err(Status::failed_precondition("heal control topology does not match")); - } let canonical_ack = rustfs_protos::canonical_heal_control_capability_ack( request.get_ref().version, fingerprint, @@ -324,9 +520,39 @@ impl heal_control_service_server::HealControlService for HealControlRpcService { success: true, result: result.into(), error_info: None, + response_proof: Bytes::new(), })); } - Err(Status::unimplemented("heal control routing is not enabled")) + let endpoints = self + .endpoint_pools() + .await + .ok_or_else(|| Status::failed_precondition("heal control topology is not initialized"))?; + if !heal::heal_control_coordinator(&endpoints) + .map_err(Status::failed_precondition)? + .is_local + { + return Err(Status::failed_precondition("heal control request reached a non-coordinator node")); + } + let envelope = + rustfs_protos::heal_control::decode_envelope(&request.get_ref().command).map_err(Status::invalid_argument)?; + let coordinator_epoch = + rustfs_protos::heal_control_coordinator_epoch(fingerprint).map_err(Status::failed_precondition)?; + let result = execute_heal_control_envelope(envelope, coordinator_epoch).await?; + let canonical_response = rustfs_protos::canonical_heal_control_response_body( + request.get_ref().version, + &request.get_ref().topology_fingerprint, + &request.get_ref().command, + &result, + ) + .map_err(|_| Status::internal("heal control response length cannot be represented"))?; + let response_proof = sign_tonic_rpc_response_proof(&canonical_response) + .map_err(|_| Status::internal("heal control response proof is unavailable"))?; + Ok(Response::new(HealControlResponse { + success: true, + result: result.into(), + error_info: None, + response_proof: response_proof.into(), + })) } } @@ -1388,18 +1614,21 @@ impl Node for NodeService { #[allow(unused_imports)] mod tests { use super::{ - CollectMetricsOpts, Error, HEAL_CONTROL_PAYLOAD_MAX_SIZE, MetricType, Node as _, NodeService, PEER_RESTSIGNAL, + CollectMetricsOpts, DiskStore, Error, HEAL_CONTROL_PAYLOAD_MAX_SIZE, MetricType, Node as _, NodeService, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, STORAGE_CLASS_SUB_SYS, - background_rebalance_start_error_message, initialize_heal_topology_fingerprint, make_heal_control_server, - make_heal_control_server_with_cache, make_server, scanner_activity_response, stop_rebalance_response, + admit_heal_control_replay, background_rebalance_start_error_message, execute_heal_control_envelope_with_manager, + initialize_heal_topology_fingerprint, make_heal_control_server, make_heal_control_server_with_cache, make_server, + remove_heal_control_replay, scanner_activity_response, stop_rebalance_response, }; use crate::storage::rpc::node_service::heal::heal_topology_fingerprint; + use crate::storage::storage_api::rpc_consumer::node_service::{HealBucketInfo, HealEndpoint}; use crate::storage::storage_api::set_tonic_canonical_body_digest; use crate::storage::storage_api::{ Endpoint, ecstore_layout::{EndpointServerPools, Endpoints, PoolEndpoints}, }; use bytes::Bytes; + use rustfs_heal::heal::{manager::HealManager, storage::HealStorageAPI}; use rustfs_protos::models::PingBodyBuilder; use rustfs_protos::proto_gen::node_service::{ BackgroundHealStatusRequest, CheckPartsRequest, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest, @@ -1422,14 +1651,273 @@ mod tests { node_service_server::NodeServiceServer, }; use std::{collections::HashMap, sync::Arc}; + use time::OffsetDateTime; use tokio::net::TcpListener; + use tokio::time::Duration; use tokio_stream::wrappers::TcpListenerStream; use tonic::{Request, Response, Status}; + struct HealControlMockStorage; + + #[async_trait::async_trait] + impl HealStorageAPI for HealControlMockStorage { + async fn get_object_meta( + &self, + _bucket: &str, + _object: &str, + ) -> rustfs_heal::Result> { + Ok(None) + } + + async fn get_object_data(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result>> { + Ok(None) + } + + async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> rustfs_heal::Result<()> { + Ok(()) + } + + async fn delete_object(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<()> { + Ok(()) + } + + async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result { + Ok(true) + } + + async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { + Ok(Vec::new()) + } + + async fn get_disk_status(&self, _endpoint: &HealEndpoint) -> rustfs_heal::Result { + Ok(rustfs_heal::heal::storage::DiskStatus::Ok) + } + + async fn format_disk(&self, _endpoint: &HealEndpoint) -> rustfs_heal::Result<()> { + Ok(()) + } + + async fn get_bucket_info(&self, _bucket: &str) -> rustfs_heal::Result> { + Ok(None) + } + + async fn heal_bucket_metadata(&self, _bucket: &str) -> rustfs_heal::Result<()> { + Ok(()) + } + + async fn list_buckets(&self) -> rustfs_heal::Result> { + Ok(Vec::new()) + } + + async fn object_exists(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result { + Ok(false) + } + + async fn get_object_size(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { + Ok(None) + } + + async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { + Ok(None) + } + + async fn heal_object( + &self, + _bucket: &str, + _object: &str, + _version_id: Option<&str>, + _opts: &rustfs_common::heal_channel::HealOpts, + ) -> rustfs_heal::Result<(rustfs_madmin::heal_commands::HealResultItem, Option)> { + Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None)) + } + + async fn heal_bucket( + &self, + _bucket: &str, + _opts: &rustfs_common::heal_channel::HealOpts, + ) -> rustfs_heal::Result { + Ok(rustfs_madmin::heal_commands::HealResultItem::default()) + } + + async fn heal_format( + &self, + _dry_run: bool, + ) -> rustfs_heal::Result<(rustfs_madmin::heal_commands::HealResultItem, Option)> { + Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None)) + } + + async fn list_objects_for_heal( + &self, + _bucket: &str, + _prefix: &str, + ) -> rustfs_heal::Result> { + Ok(Vec::new()) + } + + async fn list_objects_for_heal_page( + &self, + _bucket: &str, + _prefix: &str, + _continuation_token: Option<&str>, + ) -> rustfs_heal::Result<(Vec, Option, bool)> { + Ok((Vec::new(), None, false)) + } + + async fn get_disk_for_resume(&self, _set_disk_id: &str) -> rustfs_heal::Result { + Err(rustfs_heal::Error::other("not implemented in heal control test")) + } + } + fn create_test_node_service() -> NodeService { make_server() } + #[tokio::test] + async fn heal_control_replay_cache_singleflights_only_matching_request_ids() { + let mut cache = HashMap::new(); + let first = admit_heal_control_replay(&mut cache, "request-1", &[1; 32], 200, 100).unwrap(); + let exact = admit_heal_control_replay(&mut cache, "request-1", &[1; 32], 200, 100).unwrap(); + assert!(Arc::ptr_eq(&first, &exact)); + + let collision = admit_heal_control_replay(&mut cache, "request-1", &[2; 32], 200, 100) + .expect_err("one request ID must not identify two commands"); + assert_eq!(collision.code(), tonic::Code::AlreadyExists); + + remove_heal_control_replay(&mut cache, "request-1", &first); + assert!(!cache.contains_key("request-1"), "completed query results must not remain cached"); + + let second = admit_heal_control_replay(&mut cache, "request-2", &[2; 32], 300, 100).unwrap(); + let first_execution = first.result.lock().await; + let _second_execution = tokio::time::timeout(Duration::from_millis(50), second.result.lock()) + .await + .expect("a different request ID must not wait behind the first request"); + + drop(first_execution); + drop(_second_execution); + drop(exact); + drop(first); + drop(second); + let _third = admit_heal_control_replay(&mut cache, "request-3", &[3; 32], 400, 300).unwrap(); + assert!(!cache.contains_key("request-1"), "expired idle entries must be purged before admission"); + } + + #[tokio::test] + async fn heal_control_executor_preserves_canonical_token_and_drops_query_results() { + let manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None)); + let coordinator_epoch = 7; + let now = OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000; + let now = i64::try_from(now).expect("test clock should fit in i64"); + let metadata = || rustfs_protos::heal_control::RequestMetadata::new(rand::random(), now, now + 30_000, coordinator_epoch); + let start = |request_id: String| { + let mut request = + rustfs_common::heal_channel::create_heal_request("bucket".to_string(), Some("prefix".to_string()), false, None); + request.id = request_id; + request.source = rustfs_common::heal_channel::HealRequestSource::Admin; + request + }; + + let canonical_token = uuid::Uuid::new_v4().to_string(); + let first = rustfs_protos::heal_control::Envelope::start(start(canonical_token.clone()), metadata()).unwrap(); + let first_result = execute_heal_control_envelope_with_manager(first, coordinator_epoch, Some(Arc::clone(&manager))) + .await + .unwrap(); + let first_outcome = rustfs_protos::heal_control::decode_result(&first_result) + .and_then(|result| result.into_outcome(&canonical_token, coordinator_epoch)) + .unwrap(); + assert!(matches!( + first_outcome, + rustfs_protos::heal_control::Outcome::Start { + task_id, + admission: rustfs_protos::heal_control::Admission::Accepted, + } if task_id == canonical_token + )); + + let duplicate_id = uuid::Uuid::new_v4().to_string(); + let duplicate = rustfs_protos::heal_control::Envelope::start(start(duplicate_id.clone()), metadata()).unwrap(); + let duplicate_result = + execute_heal_control_envelope_with_manager(duplicate, coordinator_epoch, Some(Arc::clone(&manager))) + .await + .unwrap(); + let duplicate_outcome = rustfs_protos::heal_control::decode_result(&duplicate_result) + .and_then(|result| result.into_outcome(&duplicate_id, coordinator_epoch)) + .unwrap(); + assert!(matches!( + duplicate_outcome, + rustfs_protos::heal_control::Outcome::Start { + task_id, + admission: rustfs_protos::heal_control::Admission::Merged, + } if task_id == canonical_token + )); + + let query_id = uuid::Uuid::new_v4().to_string(); + let query = rustfs_protos::heal_control::Envelope::query( + query_id.clone(), + metadata(), + "bucket/prefix".to_string(), + canonical_token.clone(), + ) + .unwrap(); + let query_result = execute_heal_control_envelope_with_manager(query, coordinator_epoch, Some(Arc::clone(&manager))) + .await + .unwrap(); + let query_outcome = rustfs_protos::heal_control::decode_result(&query_result) + .and_then(|result| result.into_outcome(&query_id, coordinator_epoch)) + .unwrap(); + assert!(matches!( + query_outcome, + rustfs_protos::heal_control::Outcome::Channel { success: true, .. } + )); + + let cancel_id = uuid::Uuid::new_v4().to_string(); + let cancel = rustfs_protos::heal_control::Envelope::cancel( + cancel_id.clone(), + metadata(), + "bucket/prefix".to_string(), + canonical_token.clone(), + ) + .unwrap(); + let cancel_result = execute_heal_control_envelope_with_manager(cancel, coordinator_epoch, Some(Arc::clone(&manager))) + .await + .unwrap(); + let cancel_outcome = rustfs_protos::heal_control::decode_result(&cancel_result) + .and_then(|result| result.into_outcome(&cancel_id, coordinator_epoch)) + .unwrap(); + assert!(matches!( + cancel_outcome, + rustfs_protos::heal_control::Outcome::Channel { success: true, .. } + )); + + let stopped_query_id = uuid::Uuid::new_v4().to_string(); + let stopped_query = rustfs_protos::heal_control::Envelope::query( + stopped_query_id.clone(), + metadata(), + "bucket/prefix".to_string(), + canonical_token, + ) + .unwrap(); + let stopped_result = execute_heal_control_envelope_with_manager(stopped_query, coordinator_epoch, Some(manager)) + .await + .unwrap(); + let stopped_outcome = rustfs_protos::heal_control::decode_result(&stopped_result) + .and_then(|result| result.into_outcome(&stopped_query_id, coordinator_epoch)) + .unwrap(); + assert!(matches!( + stopped_outcome, + rustfs_protos::heal_control::Outcome::Channel { + success: true, + error: Some(detail), + .. + } if detail == "heal task not found or expired" + )); + + let replay_cache = super::HEAL_CONTROL_REPLAY_CACHE.get().unwrap().lock().await; + assert!(!replay_cache.contains_key(&query_id), "completed query results must not remain cached"); + assert!( + !replay_cache.contains_key(&stopped_query_id), + "completed stopped queries must not remain cached" + ); + } + #[tokio::test] async fn test_make_server() { let service = make_server(); @@ -1439,19 +1927,24 @@ mod tests { fn heal_control_request(command: &[u8]) -> Request { Request::new(HealControlRequest { - version: 1, + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, topology_fingerprint: "fingerprint".to_string(), command: Bytes::copy_from_slice(command), }) } fn heal_control_test_endpoints(last_host: &str) -> EndpointServerPools { + heal_control_test_endpoints_with_coordinator(last_host, false) + } + + fn heal_control_test_endpoints_with_coordinator(last_host: &str, coordinator_local: bool) -> EndpointServerPools { let endpoints = ["node-a", "node-b", "node-c", last_host] .into_iter() .enumerate() .map(|(index, host)| { let mut endpoint = Endpoint::try_from(format!("http://{host}:9000/disk{}", index + 1).as_str()) .expect("test endpoint should parse"); + endpoint.is_local = coordinator_local && index == 0; endpoint.set_pool_index(0); endpoint.set_set_index(index / 2); endpoint.set_disk_index(index % 2); @@ -1475,7 +1968,7 @@ mod tests { } #[tokio::test] - async fn heal_control_requires_body_bound_auth_before_reporting_disabled() { + async fn heal_control_requires_body_bound_auth_before_topology_validation() { let service = make_heal_control_server(); let unsigned = service .heal_control(heal_control_request(b"query")) @@ -1484,20 +1977,31 @@ mod tests { assert_eq!(unsigned.code(), tonic::Code::PermissionDenied); let mut tampered = heal_control_request(b"query"); - let other_body = - rustfs_protos::canonical_heal_control_request_body(1, "fingerprint", b"cancel").expect("small request should encode"); + let other_body = rustfs_protos::canonical_heal_control_request_body( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + "fingerprint", + b"cancel", + ) + .expect("small request should encode"); set_tonic_canonical_body_digest(&mut tampered, &other_body).expect("digest metadata should encode"); mark_v2_authenticated(&mut tampered); let tampered = service.heal_control(tampered).await.expect_err("tampered request must fail"); assert_eq!(tampered.code(), tonic::Code::PermissionDenied); let mut signed = heal_control_request(b"query"); - let body = - rustfs_protos::canonical_heal_control_request_body(1, "fingerprint", b"query").expect("small request should encode"); + let body = rustfs_protos::canonical_heal_control_request_body( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + "fingerprint", + b"query", + ) + .expect("small request should encode"); set_tonic_canonical_body_digest(&mut signed, &body).expect("digest metadata should encode"); mark_v2_authenticated(&mut signed); - let disabled = service.heal_control(signed).await.expect_err("routing must remain disabled"); - assert_eq!(disabled.code(), tonic::Code::Unimplemented); + let unavailable = service + .heal_control(signed) + .await + .expect_err("authenticated commands still require initialized topology"); + assert_eq!(unavailable.code(), tonic::Code::FailedPrecondition); } #[tokio::test] @@ -1511,15 +2015,12 @@ mod tests { } #[tokio::test] - async fn heal_control_probe_requires_exact_topology_and_keeps_commands_disabled() { + async fn heal_control_probe_requires_exact_topology_and_coordinator() { let _ = rustfs_credentials::set_global_rpc_secret("heal-control-node-service-test-secret".to_string()); let endpoints = heal_control_test_endpoints("node-d"); let fingerprint = heal_topology_fingerprint(&endpoints).expect("test topology should hash"); - let cache = Arc::new(tokio::sync::OnceCell::new()); - initialize_heal_topology_fingerprint(Arc::clone(&cache), endpoints) - .await - .expect("valid topology should initialize"); - let service = make_heal_control_server_with_cache(cache); + let (service, source) = super::make_heal_control_server_for_source(); + *source.write().await = Some(endpoints); let probe_command = rustfs_protos::heal_control_capability_probe(&[7; 16]); let mut probe = Request::new(HealControlRequest { @@ -1548,6 +2049,25 @@ mod tests { crate::storage::storage_api::verify_tonic_rpc_response_proof(&canonical_ack, &response.into_inner().result) .expect("response proof should authenticate the exact acknowledgement"); + let mut old_probe = Request::new(HealControlRequest { + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION - 1, + topology_fingerprint: fingerprint.clone(), + command: Bytes::from(probe_command.clone()), + }); + let old_body = rustfs_protos::canonical_heal_control_request_body( + old_probe.get_ref().version, + &old_probe.get_ref().topology_fingerprint, + &old_probe.get_ref().command, + ) + .expect("old probe should encode for rejection test"); + set_tonic_canonical_body_digest(&mut old_probe, &old_body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut old_probe); + let old_version = service + .heal_control(old_probe) + .await + .expect_err("old coordination capability must fail closed"); + assert_eq!(old_version.code(), tonic::Code::FailedPrecondition); + let divergent_probe = rustfs_protos::heal_control_capability_probe(&[8; 16]); let mut divergent = Request::new(HealControlRequest { version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, @@ -1569,15 +2089,81 @@ mod tests { .expect_err("divergent topology must fail closed"); assert_eq!(mismatch.code(), tonic::Code::FailedPrecondition); - let mut command = heal_control_request(b"start"); - let body = rustfs_protos::canonical_heal_control_request_body(1, "fingerprint", b"start").expect("command should encode"); + let mut command = Request::new(HealControlRequest { + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + topology_fingerprint: fingerprint.clone(), + command: Bytes::from_static(b"start"), + }); + let body = rustfs_protos::canonical_heal_control_request_body( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + &fingerprint, + b"start", + ) + .expect("command should encode"); set_tonic_canonical_body_digest(&mut command, &body).expect("digest metadata should encode"); mark_v2_authenticated(&mut command); - let disabled = service + let non_coordinator = service .heal_control(command) .await - .expect_err("commands must remain disabled"); - assert_eq!(disabled.code(), tonic::Code::Unimplemented); + .expect_err("commands must be rejected by a non-coordinator node"); + assert_eq!(non_coordinator.code(), tonic::Code::FailedPrecondition); + } + + #[tokio::test] + async fn heal_control_coordinator_rejects_expired_and_non_admin_starts() { + let _ = rustfs_credentials::set_global_rpc_secret("heal-control-node-service-test-secret".to_string()); + let endpoints = heal_control_test_endpoints_with_coordinator("node-d", true); + let fingerprint = heal_topology_fingerprint(&endpoints).expect("test topology should hash"); + let coordinator_epoch = + rustfs_protos::heal_control_coordinator_epoch(&fingerprint).expect("test topology should have an epoch"); + let (service, source) = super::make_heal_control_server_for_source(); + *source.write().await = Some(endpoints); + + fn signed_command(fingerprint: &str, command: Vec) -> Request { + let mut request = Request::new(HealControlRequest { + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + topology_fingerprint: fingerprint.to_string(), + command: command.into(), + }); + let body = rustfs_protos::canonical_heal_control_request_body( + request.get_ref().version, + &request.get_ref().topology_fingerprint, + &request.get_ref().command, + ) + .expect("command should encode"); + set_tonic_canonical_body_digest(&mut request, &body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut request); + request + } + + let expired_request = rustfs_common::heal_channel::create_heal_request("bucket".to_string(), None, false, None); + let expired = rustfs_protos::heal_control::Envelope::start( + expired_request, + rustfs_protos::heal_control::RequestMetadata::new([1; 16], 1, 2, coordinator_epoch), + ) + .and_then(|envelope| rustfs_protos::heal_control::encode_envelope(&envelope)) + .expect("expired command should encode structurally"); + let expired = service + .heal_control(signed_command(&fingerprint, expired)) + .await + .expect_err("expired commands must fail before admission"); + assert_eq!(expired.code(), tonic::Code::FailedPrecondition); + + let mut non_admin_request = rustfs_common::heal_channel::create_heal_request("bucket".to_string(), None, false, None); + non_admin_request.source = rustfs_common::heal_channel::HealRequestSource::Scanner; + let now = OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000; + let now = i64::try_from(now).expect("test clock should fit in i64"); + let non_admin = rustfs_protos::heal_control::Envelope::start( + non_admin_request, + rustfs_protos::heal_control::RequestMetadata::new([2; 16], now, now + 1_000, coordinator_epoch), + ) + .and_then(|envelope| rustfs_protos::heal_control::encode_envelope(&envelope)) + .expect("non-admin command should encode structurally"); + let non_admin = service + .heal_control(signed_command(&fingerprint, non_admin)) + .await + .expect_err("non-admin commands must fail before admission"); + assert_eq!(non_admin.code(), tonic::Code::PermissionDenied); } #[tokio::test] @@ -3389,32 +3975,43 @@ mod tests { } #[tokio::test] - async fn heal_control_transport_enforces_codec_limit_and_reaches_disabled_handler() { + async fn heal_control_transport_enforces_codec_limit_and_fails_closed() { let Some(mut client) = connect_test_heal_control_client().await else { return; }; let mut request = heal_control_request(b"query"); - let body = - rustfs_protos::canonical_heal_control_request_body(1, "fingerprint", b"query").expect("small request should encode"); + let body = rustfs_protos::canonical_heal_control_request_body( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + "fingerprint", + b"query", + ) + .expect("small request should encode"); set_tonic_canonical_body_digest(&mut request, &body).expect("digest metadata should encode"); mark_v2_authenticated(&mut request); - let disabled = client.heal_control(request).await.expect_err("routing must remain disabled"); - assert_eq!(disabled.code(), tonic::Code::Unimplemented); + let rejected = client + .heal_control(request) + .await + .expect_err("invalid command must fail closed"); + assert_eq!(rejected.code(), tonic::Code::FailedPrecondition); let max_command = vec![0; HEAL_CONTROL_PAYLOAD_MAX_SIZE]; let mut max_request = heal_control_request(&max_command); - let max_body = rustfs_protos::canonical_heal_control_request_body(1, "fingerprint", &max_command) - .expect("maximum request should encode"); + let max_body = rustfs_protos::canonical_heal_control_request_body( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + "fingerprint", + &max_command, + ) + .expect("maximum request should encode"); set_tonic_canonical_body_digest(&mut max_request, &max_body).expect("digest metadata should encode"); mark_v2_authenticated(&mut max_request); - let disabled = client + let rejected = client .heal_control(max_request) .await - .expect_err("maximum valid request must reach disabled handler"); - assert_eq!(disabled.code(), tonic::Code::Unimplemented); + .expect_err("maximum valid transport payload must reach validation"); + assert_eq!(rejected.code(), tonic::Code::FailedPrecondition); let oversized = Request::new(HealControlRequest { - version: 1, + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, topology_fingerprint: "fingerprint".to_string(), command: Bytes::from(vec![0; rustfs_protos::HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE]), }); diff --git a/rustfs/src/storage/rpc/node_service/heal.rs b/rustfs/src/storage/rpc/node_service/heal.rs index 783ce8823..34ec0991b 100644 --- a/rustfs/src/storage/rpc/node_service/heal.rs +++ b/rustfs/src/storage/rpc/node_service/heal.rs @@ -28,6 +28,24 @@ const NODE_HEAL_STATUS_VERSION: u8 = 1; const NODE_HEAL_STATUS_MAX_SIZE: usize = 64 * 1024; const HEAL_TOPOLOGY_FINGERPRINT_DOMAIN: &[u8] = b"rustfs-heal-topology-v1\0"; +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct HealControlCoordinator { + pub grid_host: String, + pub is_local: bool, +} + +pub(crate) fn heal_control_coordinator(endpoint_pools: &EndpointServerPools) -> Result { + endpoint_pools + .get_nodes() + .into_iter() + .next() + .map(|node| HealControlCoordinator { + grid_host: node.grid_host, + is_local: node.is_local, + }) + .ok_or_else(|| "heal control topology has no coordinator".to_string()) +} + fn hash_sized(hasher: &mut Sha256, value: &[u8]) -> Result<(), String> { let len = u64::try_from(value.len()).map_err(|_| "heal topology field length cannot be represented".to_string())?; hasher.update(len.to_be_bytes()); @@ -246,7 +264,7 @@ pub(crate) fn decode_node_heal_status(data: &[u8]) -> Result = super::super::Result; + #[cfg(test)] + pub(crate) type HealEndpoint = super::super::ecstore_disk::endpoint::Endpoint; + #[cfg(test)] + pub(crate) type HealBucketInfo = super::super::contract::bucket::BucketInfo; + #[cfg(test)] pub(crate) const STORAGE_CLASS_SUB_SYS: &str = super::super::STORAGE_CLASS_SUB_SYS;