From 8201a74f7f25517d197760997d0f8520fed886af Mon Sep 17 00:00:00 2001 From: cxymds Date: Sat, 8 Aug 2026 05:43:26 +0800 Subject: [PATCH] feat(rpc): advertise cross-pool fence capability (#5809) * feat(rpc): advertise cross-pool fence capability * refactor(rpc): narrow fence capability surface * fix(rpc): state fence compatibility removal condition --- crates/protos/src/lib.rs | 49 +++++++- docs/architecture/compat-cleanup-register.md | 1 + rustfs/src/storage/rpc/node_service.rs | 118 ++++++++++++++++++- 3 files changed, 160 insertions(+), 8 deletions(-) diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index d33594bc0..8a60d4963 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -172,6 +172,7 @@ pub const HEAL_CONTROL_PROTOCOL_VERSION: u32 = 3; pub const DYNAMIC_CONFIG_PROTOCOL_VERSION: u32 = 1; pub const HEAL_CONTROL_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-heal-control-capability-v3\0"; pub const REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-tier-remote-version-state-capability-v1\0"; +pub const CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-cross-pool-fence-capability-v1\0"; pub const TIER_MUTATION_RPC_MAX_PREPARE_PAYLOAD_SIZE: usize = 64 * 1024; pub const TIER_MUTATION_RPC_MAX_COMMIT_PAYLOAD_SIZE: usize = 1024; pub const TIER_MUTATION_RPC_MAX_MESSAGE_SIZE: usize = TIER_MUTATION_RPC_MAX_PREPARE_PAYLOAD_SIZE + 4096; @@ -210,6 +211,11 @@ pub fn is_remote_version_state_capability_probe(command: &[u8]) -> bool { && command.starts_with(REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX) } +pub fn is_cross_pool_fence_capability_probe(command: &[u8]) -> bool { + command.len() == CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX.len() + 16 + && command.starts_with(CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX) +} + pub fn encode_remote_version_state_capability( topology_member: &str, process_epoch: &[u8; 16], @@ -241,6 +247,18 @@ pub fn decode_remote_version_state_capability(result: &[u8]) -> Result<(&str, &[ Ok((topology_member, process_epoch)) } +pub fn encode_cross_pool_fence_capability( + supported_version: u32, + topology_member: &str, + process_epoch: &[u8; 16], +) -> Result, std::num::TryFromIntError> { + let identity = encode_remote_version_state_capability(topology_member, process_epoch)?; + let mut result = Vec::with_capacity(4 + identity.len()); + result.extend_from_slice(&supported_version.to_be_bytes()); + result.extend_from_slice(&identity); + Ok(result) +} + /// Builds the stable byte representation authenticated for a heal-control request. /// /// This deliberately does not reuse protobuf encoding: mixed-version peers may @@ -1700,11 +1718,12 @@ mod scanner_activity_tests { #[cfg(test)] mod heal_control_tests { use super::{ - HEAL_CONTROL_CAPABILITY_PROBE_PREFIX, HEAL_CONTROL_PROTOCOL_VERSION, REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX, - canonical_heal_control_capability_ack, canonical_heal_control_request_body, canonical_heal_control_response_body, - decode_remote_version_state_capability, encode_remote_version_state_capability, 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, is_remote_version_state_capability_probe, + CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX, HEAL_CONTROL_CAPABILITY_PROBE_PREFIX, HEAL_CONTROL_PROTOCOL_VERSION, + REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX, canonical_heal_control_capability_ack, canonical_heal_control_request_body, + canonical_heal_control_response_body, decode_remote_version_state_capability, encode_cross_pool_fence_capability, + encode_remote_version_state_capability, heal_control_capability_probe, heal_control_coordinator_epoch, + heal_control_execution_timeout, heal_control_execution_timeout_for, internode_rpc_timeout, + is_cross_pool_fence_capability_probe, is_heal_control_capability_probe, is_remote_version_state_capability_probe, normalize_internode_rpc_timeout, remote_version_state_capability_probe, }; use crate::heal_control; @@ -1785,6 +1804,26 @@ mod heal_control_tests { assert!(decode_remote_version_state_capability(&invalid_utf8).is_err()); } + #[test] + fn cross_pool_fence_capability_binds_version_member_and_epoch() { + assert_eq!(CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX, b"rustfs-cross-pool-fence-capability-v1\0"); + let mut probe = b"rustfs-cross-pool-fence-capability-v1\0".to_vec(); + probe.extend_from_slice(&[7; 16]); + assert!(is_cross_pool_fence_capability_probe(&probe)); + assert!(!is_cross_pool_fence_capability_probe(CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX)); + let mut wrong_prefix = probe.clone(); + wrong_prefix[0] ^= 1; + assert!(!is_cross_pool_fence_capability_probe(&wrong_prefix)); + + let encoded = encode_cross_pool_fence_capability(0x0102_0304, "node-a:9000", &[7; 16]) + .expect("small capability response should encode"); + let mut expected_response = vec![1, 2, 3, 4]; + expected_response.extend_from_slice(&11_u64.to_be_bytes()); + expected_response.extend_from_slice(b"node-a:9000"); + expected_response.extend_from_slice(&[7; 16]); + assert_eq!(encoded, expected_response); + } + #[test] fn canonical_response_binds_request_and_result() { let baseline = canonical_heal_control_response_body(2, "abcdef", b"query", b"result").unwrap(); diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index 4dd566e44..a94756900 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/docs/architecture/compat-cleanup-register.md @@ -12,6 +12,7 @@ for later deletion. ## Open Items +- `cross-pool-fence-v1` authenticated unsupported advertisement: predeployment servers recognize the versioned cross-pool fence capability probe but report support version 0, allowing a later all-peer probe to distinguish predeployment nodes without activating a second lock domain. Replace the unsupported advertisement only when composite lock acquisition, a cluster-wide activation fence, complete fleet proof, commit-time proof revalidation, and fail-closed revocation ship together. - `table-catalog-dotted-namespace` Iceberg REST namespace path compatibility: existing RustFS clients use dotted namespace paths, while the standard multi-level contract uses the URL-encoded unit separator `%1F`. New servers accept both forms so a rolling upgrade does not invalidate existing catalog configuration. Remove the dotted fallback after the minimum supported RustFS release advertises `%1F` and all supported clients have refreshed their catalog configuration. - `rustfs-5509` FileInfo positional MessagePack decoding: beta.11 serialized 28 fields, while beta.12 inserted transition-version fields in the middle and serialized an incompatible 30-field array. New releases write named maps and retain readers for both shipped array layouts so direct and rolling upgrades can read either release. Remove the positional-array readers after every supported direct-upgrade release writes named maps and no retained RPC payload can contain a pre-map FileInfo array. - `rustfs-5416` Helm distributed startup wait setting: charts that predate explicit local endpoint identity expose startupWaitTimeoutSeconds for their peer DNS/TCP init gate. The new chart keeps the value accepted but ignores it after moving startup convergence into RustFS. Remove the value and its documentation after the minimum supported direct-upgrade chart includes localEndpointHost.autoInject and no longer renders the peer gate. diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 9a572a707..a8a3156b0 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -153,6 +153,9 @@ fn remove_heal_control_replay( static HEAL_CONTROL_REPLAY_CACHE: OnceLock>>> = OnceLock::new(); static NODE_CAPABILITY_SERVER_EPOCH: LazyLock = LazyLock::new(Uuid::new_v4); +// RUSTFS_COMPAT_TODO(cross-pool-fence-v1): advertise unsupported during predeployment. Remove after composite acquisition, +// activation fencing, fleet proof, commit-time proof revalidation, and fail-closed revocation ship together. +const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 0; fn admit_heal_control_replay( replay_cache: &mut HashMap>, @@ -840,7 +843,9 @@ impl heal_control_service_server::HealControlService for HealControlRpcService { response_proof: Bytes::new(), })); } - if rustfs_protos::is_remote_version_state_capability_probe(&request.get_ref().command) { + let remote_version_state_probe = rustfs_protos::is_remote_version_state_capability_probe(&request.get_ref().command); + let cross_pool_fence_probe = rustfs_protos::is_cross_pool_fence_capability_probe(&request.get_ref().command); + if remote_version_state_probe || cross_pool_fence_probe { let topology_member = self .endpoint_pools() .await @@ -850,9 +855,17 @@ impl heal_control_service_server::HealControlService for HealControlRpcService { if topology_member.is_empty() { return Err(Status::failed_precondition("local topology member identity is unavailable")); } - let result = + let result = if remote_version_state_probe { rustfs_protos::encode_remote_version_state_capability(&topology_member, NODE_CAPABILITY_SERVER_EPOCH.as_bytes()) - .map_err(|_| Status::internal("remote version state capability length cannot be represented"))?; + .map_err(|_| Status::internal("remote version state capability length cannot be represented"))? + } else { + rustfs_protos::encode_cross_pool_fence_capability( + CROSS_POOL_FENCE_SUPPORTED_VERSION, + &topology_member, + NODE_CAPABILITY_SERVER_EPOCH.as_bytes(), + ) + .map_err(|_| Status::internal("cross-pool fence capability length cannot be represented"))? + }; let canonical_response = rustfs_protos::canonical_heal_control_response_body( request.get_ref().version, &request.get_ref().topology_fingerprint, @@ -3302,6 +3315,105 @@ mod tests { .expect_err("proof from one challenge must not be reusable"); } + #[tokio::test] + async fn cross_pool_fence_probe_authenticates_unsupported_rollout_state() { + let _ = rustfs_credentials::set_global_rpc_secret("cross-pool-fence-node-service-test-secret".to_string()); + let endpoints = heal_control_test_endpoints_with_coordinator("node-0", true); + assert!( + !super::heal::heal_control_coordinator(&endpoints) + .expect("test topology should have a coordinator") + .is_local + ); + let fingerprint = heal_topology_fingerprint(&endpoints).expect("test topology should hash"); + let (service, source) = super::make_heal_control_server_for_source(); + *source.write().await = Some(endpoints); + let mut probe_command = rustfs_protos::CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX.to_vec(); + probe_command.extend_from_slice(&[7; 16]); + + let unauthenticated = Request::new(HealControlRequest { + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + topology_fingerprint: fingerprint.clone(), + command: Bytes::from(probe_command.clone()), + }); + let auth_error = service + .heal_control(unauthenticated) + .await + .expect_err("capability probe without authentication must fail closed"); + assert_eq!(auth_error.code(), tonic::Code::PermissionDenied); + + let mut divergent = Request::new(HealControlRequest { + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + topology_fingerprint: "different-topology".to_string(), + command: Bytes::from(probe_command.clone()), + }); + let divergent_body = rustfs_protos::canonical_heal_control_request_body( + divergent.get_ref().version, + &divergent.get_ref().topology_fingerprint, + &divergent.get_ref().command, + ) + .expect("divergent probe should encode"); + set_tonic_canonical_body_digest(&mut divergent, &divergent_body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut divergent); + let topology_error = service + .heal_control(divergent) + .await + .expect_err("capability probe for a different topology must fail closed"); + assert_eq!(topology_error.code(), tonic::Code::FailedPrecondition); + + let mut request = Request::new(HealControlRequest { + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + topology_fingerprint: fingerprint.clone(), + command: Bytes::from(probe_command.clone()), + }); + let body = rustfs_protos::canonical_heal_control_request_body( + request.get_ref().version, + &request.get_ref().topology_fingerprint, + &request.get_ref().command, + ) + .expect("probe should encode"); + set_tonic_canonical_body_digest(&mut request, &body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut request); + let response = service + .heal_control(request) + .await + .expect("non-coordinator peer should answer a capability probe") + .into_inner(); + + assert!(response.success); + assert_eq!(response.error_info, None); + assert_eq!(&response.result[..4], &0_u32.to_be_bytes()); + let (topology_member, process_epoch) = rustfs_protos::decode_remote_version_state_capability(&response.result[4..]) + .expect("capability identity should decode"); + assert_eq!(topology_member, "node-a:9000"); + assert!( + !Uuid::from_slice(process_epoch) + .expect("server epoch should be a UUID") + .is_nil() + ); + + let canonical_response = rustfs_protos::canonical_heal_control_response_body( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + &fingerprint, + &probe_command, + &response.result, + ) + .expect("response should encode"); + crate::storage::storage_api::verify_tonic_rpc_response_proof(&canonical_response, &response.response_proof) + .expect("outer proof should bind the response to the request"); + + let mut different_probe = rustfs_protos::CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX.to_vec(); + different_probe.extend_from_slice(&[8; 16]); + let different_response = rustfs_protos::canonical_heal_control_response_body( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + &fingerprint, + &different_probe, + &response.result, + ) + .expect("different response should encode"); + crate::storage::storage_api::verify_tonic_rpc_response_proof(&different_response, &response.response_proof) + .expect_err("proof from one challenge must not be reusable"); + } + #[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());