diff --git a/crates/e2e_test/src/reliant/grpc_lock_server.rs b/crates/e2e_test/src/reliant/grpc_lock_server.rs index d73a44787..4f534bea6 100644 --- a/crates/e2e_test/src/reliant/grpc_lock_server.rs +++ b/crates/e2e_test/src/reliant/grpc_lock_server.rs @@ -856,6 +856,13 @@ impl NodeService for MinimalLockNodeService { Err(Status::unimplemented("lock-only test server")) } + async fn scanner_dirty_usage_snapshot( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("lock-only test server")) + } + async fn background_heal_status( &self, _request: Request, diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 6ad57fb21..aa04369a6 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -525,8 +525,8 @@ pub mod rpc { pub use crate::cluster::rpc::{ AuthenticatedChannel, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client, S3PeerSys, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, - ScannerBucketListing, ScannerPeerActivity, ScannerPublicationLease, TONIC_RPC_PREFIX, TonicInterceptor, - build_put_file_auth_trailer, check_and_record_signed_rpc_nonce, decode_heal_bucket_rpc_options, + ScannerBucketListing, ScannerPeerActivity, ScannerPeerDirtyUsageSnapshot, ScannerPublicationLease, TONIC_RPC_PREFIX, + TonicInterceptor, build_put_file_auth_trailer, check_and_record_signed_rpc_nonce, decode_heal_bucket_rpc_options, encode_heal_bucket_rpc_options, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers, gen_tonic_signature_interceptor, node_service_time_out_client, node_service_time_out_client_no_auth, normalize_tonic_rpc_audience, set_tonic_canonical_body_digest, sign_ns_scanner_capability, diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index 36886b36e..d0987a063 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/crates/ecstore/src/cluster/rpc/mod.rs @@ -48,7 +48,7 @@ pub use internode_data_transport::build_internode_data_transport_from_env; pub(crate) use peer_rest_client::TierConfigReloadOutcome; pub use peer_rest_client::{ KMS_SIGNAL_SUBSYSTEM, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, SERVICE_SIGNAL_REFRESH_CONFIG, - SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerPeerActivity, ScannerPublicationLease, + SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerPeerActivity, ScannerPeerDirtyUsageSnapshot, ScannerPublicationLease, }; pub(crate) use peer_s3_client::heal_bucket_local_on_disks; pub use peer_s3_client::{ diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 03648d8d6..5316199e3 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -20,7 +20,8 @@ use crate::cluster::rpc::{set_tonic_canonical_body_digest, set_tonic_mutation_bo use crate::error::{Error, Result}; use crate::storage_api_contracts::internode::{ SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION, - SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, + SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES, + SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE, }; use crate::{ bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats, TierDailyStatsWire}, @@ -47,18 +48,19 @@ use rustfs_protos::proto_gen::node_service::{ HealControlRequest, LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ReplacementRecoveryStatusRequest, - ScannerActivityRequest, ScannerActivityResponse, ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest, - ScannerPublicationLeaseResponse, ServerInfoRequest, SignalServiceRequest, SignalServiceResponse, StartDecommissionRequest, - StartProfilingRequest, StopRebalanceRequest, TierDailyStatsRequest, TierMutationAbortRequest, TierMutationCommitRequest, - TierMutationControlResponse, TierMutationFailureClass, TierMutationPeerState, TierMutationPrepareRequest, - node_service_client::NodeServiceClient, tier_mutation_control_service_client::TierMutationControlServiceClient, + ScannerActivityRequest, ScannerActivityResponse, ScannerDirtyUsageSnapshotRequest, ScannerDirtyUsageSnapshotResponse, + ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest, ScannerPublicationLeaseResponse, ServerInfoRequest, + SignalServiceRequest, SignalServiceResponse, StartDecommissionRequest, StartProfilingRequest, StopRebalanceRequest, + TierDailyStatsRequest, TierMutationAbortRequest, TierMutationCommitRequest, TierMutationControlResponse, + TierMutationFailureClass, TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient, + tier_mutation_control_service_client::TierMutationControlServiceClient, }; pub use rustfs_protos::{PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS}; use rustfs_protos::{TierMutationRpcPhase, evict_failed_connection}; use rustfs_utils::XHost; use serde::{Deserialize, Serialize as _}; use std::{ - collections::HashMap, + collections::{BTreeMap, HashMap}, io::Cursor, sync::{ Arc, Weak, @@ -185,18 +187,31 @@ pub struct ScannerPeerActivity { pub publication_blocked: Option, } +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ScannerPeerDirtyUsageSnapshot { + pub instance_id: String, + pub generation: u64, + pub pending_bucket_count: u64, + pub protocol_version: u32, + pub complete: bool, + pub buckets: BTreeMap, +} + +fn scanner_instance_id_is_valid(instance_id: &str) -> bool { + instance_id.len() == 32 + && instance_id + .as_bytes() + .iter() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(byte)) +} + fn decode_scanner_activity_with_verifier( response: ScannerActivityResponse, challenge: &[u8; 16], verify_proof: impl FnOnce(&[u8], &[u8]) -> Result<()>, ) -> Result { let instance_id = &response.instance_id; - if instance_id.len() != 32 - || !instance_id - .as_bytes() - .iter() - .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(byte)) - { + if !scanner_instance_id_is_valid(instance_id) { return Err(Error::other("peer returned an invalid scanner activity instance ID")); } let ( @@ -318,6 +333,82 @@ fn decode_scanner_activity(response: ScannerActivityResponse, challenge: &[u8; 1 }) } +fn decode_scanner_dirty_usage_snapshot_with_verifier( + response: ScannerDirtyUsageSnapshotResponse, + challenge: &[u8; 16], + verify_proof: impl FnOnce(&[u8], &[u8]) -> Result<()>, +) -> Result { + let canonical = rustfs_protos::canonical_scanner_dirty_usage_snapshot_response_body(challenge, &response) + .map_err(|_| Error::other("peer scanner dirty usage snapshot is too large to authenticate"))?; + verify_proof(&canonical, &response.response_proof)?; + + if response.protocol_version != SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION { + return Err(Error::other("peer returned unsupported scanner dirty usage snapshot protocol")); + } + if !scanner_instance_id_is_valid(&response.instance_id) { + return Err(Error::other("peer returned an invalid scanner dirty usage snapshot instance ID")); + } + if response.generation == u64::MAX { + return Err(Error::other("peer scanner dirty usage snapshot exhausted its generation")); + } + if response.pending_bucket_count > 0 && response.generation == 0 { + return Err(Error::other("peer scanner dirty usage snapshot has pending buckets without a generation")); + } + if response.buckets.len() > SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES { + return Err(Error::other("peer scanner dirty usage snapshot exceeds the entry limit")); + } + let bucket_count = u64::try_from(response.buckets.len()) + .map_err(|_| Error::other("peer scanner dirty usage snapshot entry count cannot be represented"))?; + let max_entries = u64::try_from(SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES).unwrap_or(u64::MAX); + if response.complete { + if response.pending_bucket_count != bucket_count { + return Err(Error::other( + "complete peer scanner dirty usage snapshot has an inconsistent bucket count", + )); + } + } else if !response.buckets.is_empty() || response.pending_bucket_count <= max_entries { + return Err(Error::other( + "incomplete peer scanner dirty usage snapshot must represent an entry-limit overflow", + )); + } + for pair in response.buckets.windows(2) { + if pair[0].bucket >= pair[1].bucket { + return Err(Error::other("peer scanner dirty usage snapshot buckets are not strictly ordered")); + } + } + for bucket in &response.buckets { + if bucket.bucket.is_empty() { + return Err(Error::other("peer scanner dirty usage snapshot contains an empty bucket name")); + } + if bucket.generation == 0 || bucket.generation > response.generation { + return Err(Error::other("peer scanner dirty usage snapshot contains an invalid bucket generation")); + } + } + + Ok(ScannerPeerDirtyUsageSnapshot { + instance_id: response.instance_id, + generation: response.generation, + pending_bucket_count: response.pending_bucket_count, + protocol_version: response.protocol_version, + complete: response.complete, + buckets: response + .buckets + .into_iter() + .map(|bucket| (bucket.bucket, bucket.generation)) + .collect(), + }) +} + +fn decode_scanner_dirty_usage_snapshot( + response: ScannerDirtyUsageSnapshotResponse, + challenge: &[u8; 16], +) -> Result { + decode_scanner_dirty_usage_snapshot_with_verifier(response, challenge, |canonical, proof| { + verify_tonic_rpc_response_proof(canonical, proof) + .map_err(|_| Error::other("peer returned an invalid scanner dirty usage snapshot response proof")) + }) +} + fn scanner_activity_protocol_unsupported(err: &Error) -> bool { matches!( err, @@ -1935,6 +2026,30 @@ impl PeerRestClient { } } + pub async fn scanner_dirty_usage_snapshot(&self) -> Result { + self.finalize_result( + async { + let challenge = Uuid::new_v4(); + let mut client = self + .get_client() + .await? + .max_decoding_message_size(SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE) + .max_encoding_message_size(SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE); + let mut request = Request::new(ScannerDirtyUsageSnapshotRequest { + challenge: challenge.as_bytes().to_vec().into(), + protocol_version: SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, + }); + let canonical = rustfs_protos::canonical_scanner_dirty_usage_snapshot_request_body(request.get_ref()) + .map_err(|_| Error::other("scanner dirty usage snapshot request is too large to authenticate"))?; + set_tonic_canonical_body_digest(&mut request, &canonical)?; + let response = client.scanner_dirty_usage_snapshot(request).await?.into_inner(); + decode_scanner_dirty_usage_snapshot(response, challenge.as_bytes()) + } + .await, + ) + .await + } + pub async fn acknowledge_scanner_dirty_usage(&self, instance_id: String, generation: u64) -> Result { let result = self .scanner_activity_request_with_protocol(instance_id.clone(), generation, SCANNER_ACTIVITY_PROTOCOL_VERSION) @@ -2640,6 +2755,141 @@ mod tests { }) } + fn decode_test_scanner_dirty_usage_snapshot( + response: ScannerDirtyUsageSnapshotResponse, + ) -> Result { + decode_scanner_dirty_usage_snapshot_with_verifier(response, &[9; 16], |_canonical, proof| { + (proof == b"proof") + .then_some(()) + .ok_or_else(|| Error::other("peer returned an invalid scanner dirty usage snapshot response proof")) + }) + } + + fn test_scanner_dirty_usage_snapshot_response() -> ScannerDirtyUsageSnapshotResponse { + ScannerDirtyUsageSnapshotResponse { + instance_id: "0123456789abcdef0123456789abcdef".to_string(), + generation: 7, + pending_bucket_count: 2, + protocol_version: SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, + complete: true, + buckets: vec![ + rustfs_protos::proto_gen::node_service::ScannerDirtyUsageBucket { + bucket: "archive".to_string(), + generation: 3, + }, + rustfs_protos::proto_gen::node_service::ScannerDirtyUsageBucket { + bucket: "photos".to_string(), + generation: 7, + }, + ], + response_proof: b"proof".to_vec().into(), + } + } + + #[test] + fn scanner_dirty_usage_snapshot_requires_a_complete_authenticated_ordered_view() { + let decoded = decode_test_scanner_dirty_usage_snapshot(test_scanner_dirty_usage_snapshot_response()) + .expect("a complete authenticated dirty usage snapshot should decode"); + assert_eq!(decoded.instance_id, "0123456789abcdef0123456789abcdef"); + assert_eq!(decoded.generation, 7); + assert_eq!(decoded.pending_bucket_count, 2); + assert_eq!(decoded.protocol_version, SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION); + assert!(decoded.complete); + assert_eq!(decoded.buckets.get("archive"), Some(&3)); + assert_eq!(decoded.buckets.get("photos"), Some(&7)); + + let overflow_count = + u64::try_from(SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES + 1).expect("the test snapshot entry limit should fit in u64"); + let overflow = decode_test_scanner_dirty_usage_snapshot(ScannerDirtyUsageSnapshotResponse { + pending_bucket_count: overflow_count, + complete: false, + buckets: Vec::new(), + ..test_scanner_dirty_usage_snapshot_response() + }) + .expect("an explicit all-or-nothing overflow snapshot should decode"); + assert!(!overflow.complete); + assert!(overflow.buckets.is_empty()); + } + + #[test] + fn scanner_dirty_usage_snapshot_rejects_inconsistent_or_partial_peer_data() { + let mut cases = Vec::new(); + + let mut invalid_instance = test_scanner_dirty_usage_snapshot_response(); + invalid_instance.instance_id = "ABCDEF0123456789ABCDEF0123456789".to_string(); + cases.push((invalid_instance, "instance ID")); + + let mut unsupported = test_scanner_dirty_usage_snapshot_response(); + unsupported.protocol_version = SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION + 1; + cases.push((unsupported, "unsupported")); + + let mut exhausted = test_scanner_dirty_usage_snapshot_response(); + exhausted.generation = u64::MAX; + cases.push((exhausted, "exhausted")); + + let mut inconsistent_count = test_scanner_dirty_usage_snapshot_response(); + inconsistent_count.pending_bucket_count = 3; + cases.push((inconsistent_count, "bucket count")); + + let mut unordered = test_scanner_dirty_usage_snapshot_response(); + unordered.buckets.reverse(); + cases.push((unordered, "strictly ordered")); + + let mut future_bucket = test_scanner_dirty_usage_snapshot_response(); + future_bucket.buckets[0].generation = 8; + cases.push((future_bucket, "bucket generation")); + + let mut zero_generation = test_scanner_dirty_usage_snapshot_response(); + zero_generation.buckets[0].generation = 0; + cases.push((zero_generation, "bucket generation")); + + let mut empty_bucket = test_scanner_dirty_usage_snapshot_response(); + empty_bucket.buckets[0].bucket.clear(); + cases.push((empty_bucket, "empty bucket name")); + + let mut partial = test_scanner_dirty_usage_snapshot_response(); + partial.complete = false; + cases.push((partial, "entry-limit overflow")); + + let too_many_buckets = ScannerDirtyUsageSnapshotResponse { + generation: 1, + pending_bucket_count: u64::try_from(SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES + 1) + .expect("the test snapshot entry limit should fit in u64"), + buckets: (0..=SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES) + .map(|index| rustfs_protos::proto_gen::node_service::ScannerDirtyUsageBucket { + bucket: format!("bucket-{index:04}"), + generation: 1, + }) + .collect(), + ..test_scanner_dirty_usage_snapshot_response() + }; + cases.push((too_many_buckets, "exceeds the entry limit")); + + let overflow_count = + u64::try_from(SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES + 1).expect("the test snapshot entry limit should fit in u64"); + let invalid_overflow = ScannerDirtyUsageSnapshotResponse { + generation: 0, + pending_bucket_count: overflow_count, + complete: false, + buckets: Vec::new(), + ..test_scanner_dirty_usage_snapshot_response() + }; + cases.push((invalid_overflow, "without a generation")); + + for (response, expected) in cases { + let err = + decode_test_scanner_dirty_usage_snapshot(response).expect_err("malformed dirty usage snapshots must fail closed"); + assert!(err.to_string().contains(expected), "expected {expected:?} in {err}"); + } + + let mut invalid_proof = test_scanner_dirty_usage_snapshot_response(); + invalid_proof.protocol_version = SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION + 1; + invalid_proof.response_proof = b"invalid".to_vec().into(); + let err = decode_test_scanner_dirty_usage_snapshot(invalid_proof) + .expect_err("an invalid response proof must fail before peer fields are trusted"); + assert!(err.to_string().contains("response proof")); + } + #[test] fn build_clients_from_slots_preserves_missing_remote_topology_slots() { let slots = vec![ diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index a7e399031..a0bb3fac0 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -13,7 +13,9 @@ // limitations under the License. use crate::bucket::lifecycle::tier_last_day_stats::DailyAllTierStats; -use crate::cluster::rpc::{PeerRestClient, ScannerPeerActivity, ScannerPublicationLease, TierConfigReloadOutcome}; +use crate::cluster::rpc::{ + PeerRestClient, ScannerPeerActivity, ScannerPeerDirtyUsageSnapshot, ScannerPublicationLease, TierConfigReloadOutcome, +}; use crate::diagnostics::admin_server_info::get_commit_id; use crate::disk::DiskAPI; use crate::error::{Error, Result}; @@ -2080,6 +2082,32 @@ impl NotificationSys { Ok(generations) } + pub async fn scanner_dirty_usage_snapshots(&self) -> Result> { + if self.peer_clients.is_empty() { + return Err(Error::other("scanner dirty usage snapshot probe has no remote peers")); + } + if self.all_peer_clients.len() != self.peer_clients.len() + 1 { + return Err(Error::other("scanner dirty usage snapshot peer topology is incomplete")); + } + + let mut futures = Vec::with_capacity(self.peer_clients.len()); + for client in self.peer_clients.iter().cloned() { + futures.push(async move { + let client = client.ok_or_else(|| Error::other("scanner dirty usage snapshot peer is unreachable"))?; + let host = client.grid_host.clone(); + scanner_dirty_usage_snapshot_with_retry(&client, &host) + .await + .map(|snapshot| (host, snapshot)) + }); + } + + let mut snapshots = Vec::with_capacity(futures.len()); + for result in join_all(futures).await { + snapshots.push(result?); + } + Ok(snapshots) + } + pub async fn acknowledge_scanner_dirty_usage(&self, acknowledgements: Vec<(String, String, u64)>) -> Result { let mut by_host = HashMap::with_capacity(acknowledgements.len()); for (host, instance_id, generation) in acknowledgements { @@ -2591,6 +2619,54 @@ async fn scanner_activity_with_retry(client: &PeerRestClient, host: &str) -> Res } } +async fn scanner_dirty_usage_snapshot_with_retry(client: &PeerRestClient, host: &str) -> Result { + let first = timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, client.scanner_dirty_usage_snapshot()).await; + let should_retry = match &first { + Ok(Ok(_)) => false, + Ok(Err(err)) => scanner_activity_should_retry(Some(err), false), + Err(_) => scanner_activity_should_retry(None, true), + }; + + match first { + Ok(Ok(snapshot)) => return Ok(snapshot), + Ok(Err(err)) if !should_retry => return Err(err), + Ok(Err(err)) => { + debug!( + event = EVENT_NOTIFICATION_CAPABILITY_PROBE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + result = "retrying", + capability = "scanner_dirty_usage_snapshot", + peer = host, + error = %err, + "notification capability probe retrying" + ); + client.prepare_retry().await; + } + Err(_) => { + debug!( + event = EVENT_NOTIFICATION_CAPABILITY_PROBE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + result = "retrying", + capability = "scanner_dirty_usage_snapshot", + peer = host, + timeout = ?SCANNER_ACTIVITY_PROBE_TIMEOUT, + "notification capability probe retrying" + ); + client.prepare_retry().await; + } + } + + match timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, client.scanner_dirty_usage_snapshot()).await { + Ok(result) => result, + Err(_) => { + client.evict_connection().await; + Err(Error::Timeout) + } + } +} + #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")] async fn call_peer_with_timeout( timeout_dur: Duration, @@ -3688,6 +3764,52 @@ mod tests { assert!(err.to_string().contains("peer topology is incomplete")); } + #[tokio::test] + async fn scanner_dirty_usage_snapshot_probe_rejects_unusable_peer_topologies() { + let unreachable = NotificationSys { + peer_clients: vec![None], + all_peer_clients: vec![None, None], + peer_topology_hosts: vec!["node-a:9000".to_string()], + peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), + }; + let err = unreachable + .scanner_dirty_usage_snapshots() + .await + .expect_err("an unreachable peer must invalidate the distributed dirty usage snapshot"); + assert!(err.to_string().contains("peer is unreachable")); + + let empty = NotificationSys { + peer_clients: Vec::new(), + all_peer_clients: Vec::new(), + peer_topology_hosts: Vec::new(), + peer_admin_caches: Vec::new(), + tier_config_reload_workers: Default::default(), + }; + let err = empty + .scanner_dirty_usage_snapshots() + .await + .expect_err("an empty peer set must not produce a distributed dirty usage snapshot"); + assert!(err.to_string().contains("no remote peers")); + + let client = PeerRestClient::new( + "127.0.0.1:9000".to_string().try_into().expect("peer host should parse"), + "http://127.0.0.1:9000".to_string(), + ); + let incomplete = NotificationSys { + peer_clients: vec![Some(client)], + all_peer_clients: vec![None], + peer_topology_hosts: vec!["127.0.0.1:9000".to_string()], + peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), + }; + let err = incomplete + .scanner_dirty_usage_snapshots() + .await + .expect_err("an incomplete topology must not produce a distributed dirty usage snapshot"); + assert!(err.to_string().contains("peer topology is incomplete")); + } + #[tokio::test] async fn server_info_no_client_slot_uses_topology_host_without_counting_rpc_failure() { let sys = NotificationSys { diff --git a/crates/ecstore/src/storage_api_contracts/mod.rs b/crates/ecstore/src/storage_api_contracts/mod.rs index cd48a5266..414080eb7 100644 --- a/crates/ecstore/src/storage_api_contracts/mod.rs +++ b/crates/ecstore/src/storage_api_contracts/mod.rs @@ -32,8 +32,9 @@ pub(crate) mod internode { PUT_FILE_AUTH_TRAILER_MAGIC, PUT_FILE_AUTH_V1, PUT_FILE_CAPABILITY_CHALLENGE_QUERY, PUT_FILE_CAPABILITY_QUERY, PUT_FILE_CAPABILITY_VERSION, PUT_FILE_NONCE_QUERY, PUT_FILE_SERVER_EPOCH_QUERY, PutFileCapabilityResponse, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION, - SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_STREAM_COMPLETION_QUERY, - WALK_DIR_STREAM_COMPLETION_V1, + SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES, + SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE, + WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1, }; } diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index 037e9e597..f015305f0 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -1251,6 +1251,38 @@ pub struct ScannerActivityResponse { #[prost(bool, optional, tag = "11")] pub publication_blocked: ::core::option::Option, } +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct ScannerDirtyUsageBucket { + #[prost(string, tag = "1")] + pub bucket: ::prost::alloc::string::String, + #[prost(uint64, tag = "2")] + pub generation: u64, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct ScannerDirtyUsageSnapshotRequest { + #[prost(bytes = "bytes", tag = "1")] + pub challenge: ::prost::bytes::Bytes, + #[prost(uint32, tag = "2")] + pub protocol_version: u32, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct ScannerDirtyUsageSnapshotResponse { + #[prost(string, tag = "1")] + pub instance_id: ::prost::alloc::string::String, + #[prost(uint64, tag = "2")] + pub generation: u64, + #[prost(uint64, tag = "3")] + pub pending_bucket_count: u64, + #[prost(uint32, tag = "4")] + pub protocol_version: u32, + /// Incomplete snapshots are all-or-nothing and carry no bucket entries. + #[prost(bool, tag = "5")] + pub complete: bool, + #[prost(message, repeated, tag = "6")] + pub buckets: ::prost::alloc::vec::Vec, + #[prost(bytes = "bytes", tag = "7")] + pub response_proof: ::prost::bytes::Bytes, +} /// A short-lived storage-owned read admission used only around a final /// scanner metadata publication. It is intentionally separate from the /// ScannerActivity observation wire so v6/v7 rolling compatibility remains @@ -2899,6 +2931,21 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "ScannerActivity")); self.inner.unary(req, path, codec).await } + pub async fn scanner_dirty_usage_snapshot( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ScannerDirtyUsageSnapshot"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "ScannerDirtyUsageSnapshot")); + self.inner.unary(req, path, codec).await + } pub async fn acquire_scanner_publication_lease( &mut self, request: impl tonic::IntoRequest, @@ -3457,6 +3504,10 @@ pub mod node_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn scanner_dirty_usage_snapshot( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn acquire_scanner_publication_lease( &self, request: tonic::Request, @@ -5749,6 +5800,34 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/ScannerDirtyUsageSnapshot" => { + #[allow(non_camel_case_types)] + struct ScannerDirtyUsageSnapshotSvc(pub Arc); + impl tonic::server::UnaryService for ScannerDirtyUsageSnapshotSvc { + type Response = super::ScannerDirtyUsageSnapshotResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::scanner_dirty_usage_snapshot(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = ScannerDirtyUsageSnapshotSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/node_service.NodeService/AcquireScannerPublicationLease" => { #[allow(non_camel_case_types)] struct AcquireScannerPublicationLeaseSvc(pub Arc); diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 691f9a6f3..fd8704382 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -541,6 +541,34 @@ pub fn canonical_scanner_activity_v7_response_body( Ok(body) } +pub fn canonical_scanner_dirty_usage_snapshot_request_body( + request: &proto_gen::node_service::ScannerDirtyUsageSnapshotRequest, +) -> Result, std::num::TryFromIntError> { + let mut body = CanonicalBodyBuilder::new(b"rustfs-scanner-dirty-usage-snapshot-request-v1\0"); + body.push_u32(request.protocol_version); + body.push_bytes(request.challenge.as_ref())?; + Ok(body.finish()) +} + +pub fn canonical_scanner_dirty_usage_snapshot_response_body( + challenge: &[u8], + response: &proto_gen::node_service::ScannerDirtyUsageSnapshotResponse, +) -> Result, std::num::TryFromIntError> { + let mut body = CanonicalBodyBuilder::new(b"rustfs-scanner-dirty-usage-snapshot-response-v1\0"); + body.push_bytes(challenge)?; + body.push_str(&response.instance_id)?; + body.push_u64(response.generation); + body.push_u64(response.pending_bucket_count); + body.push_u32(response.protocol_version); + body.push_bool(response.complete); + body.push_count(response.buckets.len())?; + for bucket in &response.buckets { + body.push_str(&bucket.bucket)?; + body.push_u64(bucket.generation); + } + Ok(body.finish()) +} + /// Builds the body authenticated by the short-lived remote scanner publication /// lease request. This is a separate domain from ScannerActivity so v6/v7 /// observation proofs remain byte-for-byte compatible. @@ -1751,13 +1779,112 @@ mod scanner_activity_tests { use super::{ canonical_scanner_activity_request_body, canonical_scanner_activity_response_body, canonical_scanner_activity_v4_response_body, canonical_scanner_activity_v7_response_body, + canonical_scanner_dirty_usage_snapshot_request_body, canonical_scanner_dirty_usage_snapshot_response_body, canonical_scanner_publication_lease_release_request_body, canonical_scanner_publication_lease_request_body, canonical_scanner_publication_lease_response_body, proto_gen::node_service::{ - ScannerActivityRequest, ScannerActivityResponse, ScannerPublicationLeaseRequest, ScannerPublicationLeaseResponse, + ScannerActivityRequest, ScannerActivityResponse, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshotRequest, + ScannerDirtyUsageSnapshotResponse, ScannerPublicationLeaseRequest, ScannerPublicationLeaseResponse, }, }; + #[test] + fn canonical_scanner_dirty_usage_snapshot_request_binds_every_field() { + let request = ScannerDirtyUsageSnapshotRequest { + challenge: vec![1; 16].into(), + protocol_version: 1, + }; + let baseline = canonical_scanner_dirty_usage_snapshot_request_body(&request) + .expect("scanner dirty usage snapshot request should encode"); + + for variant in [ + ScannerDirtyUsageSnapshotRequest { + challenge: vec![2; 16].into(), + ..request.clone() + }, + ScannerDirtyUsageSnapshotRequest { + protocol_version: 2, + ..request + }, + ] { + assert_ne!( + baseline, + canonical_scanner_dirty_usage_snapshot_request_body(&variant) + .expect("scanner dirty usage snapshot request variant should encode") + ); + } + } + + #[test] + fn canonical_scanner_dirty_usage_snapshot_response_binds_every_field() { + let response = ScannerDirtyUsageSnapshotResponse { + instance_id: "0123456789abcdef0123456789abcdef".to_string(), + generation: 7, + pending_bucket_count: 2, + protocol_version: 1, + complete: true, + buckets: vec![ + ScannerDirtyUsageBucket { + bucket: "archive".to_string(), + generation: 3, + }, + ScannerDirtyUsageBucket { + bucket: "photos".to_string(), + generation: 7, + }, + ], + response_proof: vec![9; 32].into(), + }; + let baseline = canonical_scanner_dirty_usage_snapshot_response_body(&[1; 16], &response) + .expect("scanner dirty usage snapshot response should encode"); + let mut variants = Vec::new(); + let mut instance = response.clone(); + instance.instance_id = "fedcba9876543210fedcba9876543210".to_string(); + variants.push(instance); + let mut generation = response.clone(); + generation.generation = 8; + variants.push(generation); + let mut count = response.clone(); + count.pending_bucket_count = 3; + variants.push(count); + let mut protocol = response.clone(); + protocol.protocol_version = 2; + variants.push(protocol); + let mut complete = response.clone(); + complete.complete = false; + variants.push(complete); + let mut bucket_name = response.clone(); + bucket_name.buckets[0].bucket = "backups".to_string(); + variants.push(bucket_name); + let mut bucket_generation = response.clone(); + bucket_generation.buckets[0].generation = 4; + variants.push(bucket_generation); + let mut bucket_order = response.clone(); + bucket_order.buckets.reverse(); + variants.push(bucket_order); + + for variant in variants { + assert_ne!( + baseline, + canonical_scanner_dirty_usage_snapshot_response_body(&[1; 16], &variant) + .expect("scanner dirty usage snapshot response variant should encode") + ); + } + assert_ne!( + baseline, + canonical_scanner_dirty_usage_snapshot_response_body(&[2; 16], &response) + .expect("scanner dirty usage snapshot response challenge variant should encode") + ); + + let mut proof_only = response; + proof_only.response_proof = vec![8; 32].into(); + assert_eq!( + baseline, + canonical_scanner_dirty_usage_snapshot_response_body(&[1; 16], &proof_only) + .expect("response proof must not authenticate itself") + ); + } + #[test] fn canonical_scanner_activity_request_binds_every_field() { let request = ScannerActivityRequest { diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index 477b7a120..d3796991c 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -882,6 +882,27 @@ message ScannerActivityResponse { optional bool publication_blocked = 11; } +message ScannerDirtyUsageBucket { + string bucket = 1; + uint64 generation = 2; +} + +message ScannerDirtyUsageSnapshotRequest { + bytes challenge = 1; + uint32 protocol_version = 2; +} + +message ScannerDirtyUsageSnapshotResponse { + string instance_id = 1; + uint64 generation = 2; + uint64 pending_bucket_count = 3; + uint32 protocol_version = 4; + // Incomplete snapshots are all-or-nothing and carry no bucket entries. + bool complete = 5; + repeated ScannerDirtyUsageBucket buckets = 6; + bytes response_proof = 7; +} + // A short-lived storage-owned read admission used only around a final // scanner metadata publication. It is intentionally separate from the // ScannerActivity observation wire so v6/v7 rolling compatibility remains @@ -1206,6 +1227,7 @@ service NodeService { // rpc CommitBinary() returns () {}; rpc SignalService(SignalServiceRequest) returns (SignalServiceResponse) {}; // auth-policy: body-bound rpc ScannerActivity(ScannerActivityRequest) returns (ScannerActivityResponse) {}; // auth-policy: body-bound + rpc ScannerDirtyUsageSnapshot(ScannerDirtyUsageSnapshotRequest) returns (ScannerDirtyUsageSnapshotResponse) {}; // auth-policy: body-bound rpc AcquireScannerPublicationLease(ScannerPublicationLeaseRequest) returns (ScannerPublicationLeaseResponse) {}; // auth-policy: body-bound rpc ReleaseScannerPublicationLease(ScannerPublicationLeaseReleaseRequest) returns (ScannerPublicationLeaseReleaseResponse) {}; // auth-policy: body-bound rpc BackgroundHealStatus(BackgroundHealStatusRequest) returns (BackgroundHealStatusResponse) {}; // auth-policy: read-only diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 6d0b84edb..f85a2e0b6 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -89,14 +89,17 @@ pub use scanner::{ scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_topology_digest, }; pub use scanner_io::{ - ScannerDirtyUsageAckError, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, clear_dirty_usage_bucket, - record_dirty_usage_bucket, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_state, - scanner_maintenance_generation, + ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, + acknowledge_dirty_usage_generation, clear_dirty_usage_bucket, record_dirty_usage_bucket, record_scanner_maintenance_change, + scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation, }; pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER}; use std::sync::atomic::{AtomicU64, Ordering}; pub use storage_api::ScannerReplicationConfig as ReplicationConfig; -pub use storage_api::scan::{SCANNER_ACTIVITY_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION}; +pub use storage_api::scan::{ + SCANNER_ACTIVITY_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES, + SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE, +}; pub use workload_admission::set_scanner_workload_admission_snapshot_provider; static SCANNER_ACTIVE_WORK_UNITS: AtomicU64 = AtomicU64::new(0); diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 79af335cf..860ebb2cd 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -748,9 +748,9 @@ pub(crate) use cache::{ current_cache_root_or_prepare_with_generation, }; pub use dirty_usage::{ - ScannerDirtyUsageAckError, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, clear_dirty_usage_bucket, - record_dirty_usage_bucket, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_state, - scanner_maintenance_generation, + ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, + acknowledge_dirty_usage_generation, clear_dirty_usage_bucket, record_dirty_usage_bucket, record_scanner_maintenance_change, + scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation, }; #[cfg(test)] pub(crate) use dirty_usage::{clear_dirty_usage_buckets_for_tests, dirty_usage_buckets_for_tests}; diff --git a/crates/scanner/src/scanner_io/dirty_usage.rs b/crates/scanner/src/scanner_io/dirty_usage.rs index 85bd2d806..a5978e263 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -27,6 +27,25 @@ pub struct ScannerDirtyUsageState { pub pending: bool, } +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ScannerDirtyUsageBucket { + pub bucket: String, + pub generation: u64, +} + +/// A point-in-time view of the local dirty bucket generations. +/// +/// `complete == false` is an all-or-nothing overflow signal: `buckets` is +/// empty and callers must fall back to the global dirty generation rather than +/// treating a bounded prefix as authoritative. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ScannerDirtyUsageSnapshot { + pub generation: u64, + pub pending_bucket_count: u64, + pub complete: bool, + pub buckets: Vec, +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)] pub enum ScannerDirtyUsageAckError { #[error("scanner process instance changed before dirty usage acknowledgement")] @@ -98,6 +117,35 @@ pub fn scanner_dirty_usage_state() -> ScannerDirtyUsageState { } } +pub fn scanner_dirty_usage_snapshot(max_entries: usize) -> ScannerDirtyUsageSnapshot { + let (generation, pending_bucket_count, complete, mut buckets) = { + let dirty_buckets = dirty_usage_buckets(); + let generation = DIRTY_USAGE_BUCKET_GENERATION.load(Ordering::Acquire); + let pending_bucket_count = usize_to_u64_saturated(dirty_buckets.len()); + let complete = dirty_buckets.len() <= max_entries; + let buckets = if complete { + dirty_buckets + .iter() + .map(|(bucket, generation)| ScannerDirtyUsageBucket { + bucket: bucket.clone(), + generation: *generation, + }) + .collect::>() + } else { + Vec::new() + }; + (generation, pending_bucket_count, complete, buckets) + }; + buckets.sort_unstable_by(|left, right| left.bucket.cmp(&right.bucket)); + + ScannerDirtyUsageSnapshot { + generation, + pending_bucket_count, + complete, + buckets, + } +} + pub fn acknowledge_dirty_usage_generation( instance_id: &str, generation: u64, diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 2b086aa5e..47f1e86a0 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -500,6 +500,51 @@ fn dirty_usage_generation_acknowledgement_preserves_newer_mutations() { clear_dirty_usage_buckets_for_tests(); } +#[test] +#[serial] +fn dirty_usage_snapshot_is_sorted_and_reports_its_cutoff() { + clear_dirty_usage_buckets_for_tests(); + let empty = scanner_dirty_usage_snapshot(0); + assert_eq!(empty.pending_bucket_count, 0); + assert!(empty.complete); + assert!(empty.buckets.is_empty()); + + record_dirty_usage_bucket("videos"); + record_dirty_usage_bucket("photos"); + let expected_generation = scanner_dirty_usage_state().generation; + + let snapshot = scanner_dirty_usage_snapshot(2); + + assert_eq!(snapshot.generation, expected_generation); + assert_eq!(snapshot.pending_bucket_count, 2); + assert!(snapshot.complete); + assert_eq!( + snapshot + .buckets + .iter() + .map(|bucket| bucket.bucket.as_str()) + .collect::>(), + vec!["photos", "videos"] + ); + assert!(snapshot.buckets.iter().all(|bucket| bucket.generation <= snapshot.generation)); + clear_dirty_usage_buckets_for_tests(); +} + +#[test] +#[serial] +fn dirty_usage_snapshot_marks_truncated_results_incomplete() { + clear_dirty_usage_buckets_for_tests(); + record_dirty_usage_bucket("archive"); + record_dirty_usage_bucket("photos"); + + let snapshot = scanner_dirty_usage_snapshot(1); + + assert_eq!(snapshot.pending_bucket_count, 2); + assert!(!snapshot.complete); + assert!(snapshot.buckets.is_empty(), "incomplete snapshots must not expose a partial bucket list"); + clear_dirty_usage_buckets_for_tests(); +} + #[test] #[serial] fn dirty_usage_generation_acknowledgement_rejects_stale_process_and_future_generation() { diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index 3c410deed..a0aa874b0 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -304,7 +304,10 @@ pub(crate) mod scan { }; #[cfg(test)] pub(crate) use super::storage_contracts::{DeleteBucketOptions, MakeBucketOptions, ObjectIO}; - pub use super::storage_contracts::{SCANNER_ACTIVITY_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION}; + pub use super::storage_contracts::{ + SCANNER_ACTIVITY_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES, + SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE, + }; } pub(crate) mod scanner_io { diff --git a/crates/storage-api/src/lib.rs b/crates/storage-api/src/lib.rs index b16a47152..2726939f0 100644 --- a/crates/storage-api/src/lib.rs +++ b/crates/storage-api/src/lib.rs @@ -56,6 +56,9 @@ pub const SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION: u32 = 5; /// state is not authenticated by that version. pub const SCANNER_ACTIVITY_V6_PROTOCOL_VERSION: u32 = 6; pub const SCANNER_ACTIVITY_PROTOCOL_VERSION: u32 = 7; +pub const SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION: u32 = 1; +pub const SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES: usize = 4096; +pub const SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE: usize = 512 * 1024; #[derive(Debug, serde::Deserialize, serde::Serialize)] #[serde(deny_unknown_fields)] diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 44578265c..0dd18424e 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -279,6 +279,27 @@ fn scanner_activity_response( } } +fn scanner_dirty_usage_snapshot_response( + snapshot: rustfs_scanner::ScannerDirtyUsageSnapshot, +) -> ScannerDirtyUsageSnapshotResponse { + ScannerDirtyUsageSnapshotResponse { + instance_id: rustfs_scanner::scanner_activity_epoch().to_string(), + generation: snapshot.generation, + pending_bucket_count: snapshot.pending_bucket_count, + protocol_version: rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, + complete: snapshot.complete, + buckets: snapshot + .buckets + .into_iter() + .map(|bucket| ScannerDirtyUsageBucket { + bucket: bucket.bucket, + generation: bucket.generation, + }) + .collect(), + response_proof: Bytes::new(), + } +} + fn scanner_activity_response_v7( namespace_generation: u64, topology_digest: [u8; 32], @@ -2085,6 +2106,42 @@ impl Node for NodeService { Ok(Response::new(response)) } + async fn scanner_dirty_usage_snapshot( + &self, + request: Request, + ) -> Result, Status> { + let canonical = rustfs_protos::canonical_scanner_dirty_usage_snapshot_request_body(request.get_ref()) + .map_err(|_| Status::invalid_argument("scanner dirty usage snapshot request is too large to authenticate"))?; + verify_tonic_canonical_body_digest(&request, &canonical) + .map_err(|err| Status::permission_denied(format!("scanner dirty usage snapshot authentication failed: {err}")))?; + if request.get_ref().protocol_version != rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION { + return Err(Status::failed_precondition(format!( + "unsupported scanner dirty usage snapshot request protocol {}", + request.get_ref().protocol_version + ))); + } + if request.get_ref().challenge.len() != 16 { + return Err(Status::invalid_argument("scanner dirty usage snapshot challenge must be 16 bytes")); + } + let challenge: [u8; 16] = request + .into_inner() + .challenge + .as_ref() + .try_into() + .map_err(|_| Status::invalid_argument("scanner dirty usage snapshot challenge must be 16 bytes"))?; + let snapshot = rustfs_scanner::scanner_dirty_usage_snapshot(rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES); + if snapshot.generation == u64::MAX { + return Err(Status::resource_exhausted("scanner dirty usage generation is exhausted")); + } + let mut response = scanner_dirty_usage_snapshot_response(snapshot); + let canonical = rustfs_protos::canonical_scanner_dirty_usage_snapshot_response_body(&challenge, &response) + .map_err(|_| Status::internal("scanner dirty usage snapshot response is too large to authenticate"))?; + response.response_proof = sign_tonic_rpc_response_proof(&canonical) + .map_err(|_| Status::unavailable("scanner dirty usage snapshot response authentication is unavailable"))? + .into(); + Ok(Response::new(response)) + } + async fn acquire_scanner_publication_lease( &self, request: Request, @@ -2579,12 +2636,12 @@ mod tests { LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, MakeVolumeRequest, MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest, ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest, - RenameFileRequest, RenamePartRequest, ScannerActivityRequest, ScannerPublicationLeaseReleaseRequest, - ScannerPublicationLeaseRequest, ServerInfoRequest, SettlePartTransactionRequest, SignalServiceRequest, - SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest, StartDecommissionRequest, - StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationAbortRequest, TierMutationFailureClass, - TierMutationPeerState, TierMutationPrepareRequest, UpdateMetacacheListingRequest, UpdateMetadataRequest, - VerifyFileRequest, WriteAllRequest, WriteMetadataRequest, WriteRequest, + RenameFileRequest, RenamePartRequest, ScannerActivityRequest, ScannerDirtyUsageSnapshotRequest, + ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest, ServerInfoRequest, SettlePartTransactionRequest, + SignalServiceRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest, + StartDecommissionRequest, StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationAbortRequest, + TierMutationFailureClass, TierMutationPeerState, TierMutationPrepareRequest, UpdateMetacacheListingRequest, + UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest, WriteRequest, heal_control_service_client::HealControlServiceClient, heal_control_service_server::{HealControlService as _, HealControlServiceServer}, node_service_client::NodeServiceClient, @@ -5883,6 +5940,13 @@ mod tests { acknowledge_dirty_usage_generation: 0, } ); + assert_tampered!( + scanner_dirty_usage_snapshot, + ScannerDirtyUsageSnapshotRequest { + challenge: vec![7; 16].into(), + protocol_version: rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, + } + ); assert_tampered!( acquire_scanner_publication_lease, ScannerPublicationLeaseRequest { @@ -6056,6 +6120,71 @@ mod tests { assert_eq!(unavailable.code(), tonic::Code::Unavailable); } + #[tokio::test] + async fn test_scanner_dirty_usage_snapshot_requires_body_bound_auth_and_signs_a_consistent_view() { + let _ = rustfs_credentials::set_global_rpc_secret("scanner-dirty-usage-snapshot-test-secret".to_string()); + let service = create_test_node_service(); + let unsigned = service + .scanner_dirty_usage_snapshot(Request::new(ScannerDirtyUsageSnapshotRequest { + challenge: vec![7; 16].into(), + protocol_version: rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, + })) + .await + .expect_err("unsigned scanner dirty usage snapshot requests must fail"); + assert_eq!(unsigned.code(), tonic::Code::PermissionDenied); + + let mut unsupported = Request::new(ScannerDirtyUsageSnapshotRequest { + challenge: vec![7; 16].into(), + protocol_version: rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION + 1, + }); + let unsupported_body = rustfs_protos::canonical_scanner_dirty_usage_snapshot_request_body(unsupported.get_ref()) + .expect("scanner dirty usage snapshot request should encode"); + set_tonic_canonical_body_digest(&mut unsupported, &unsupported_body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut unsupported); + let unsupported = service + .scanner_dirty_usage_snapshot(unsupported) + .await + .expect_err("unsupported scanner dirty usage snapshot protocols must fail closed"); + assert_eq!(unsupported.code(), tonic::Code::FailedPrecondition); + + let mut malformed = Request::new(ScannerDirtyUsageSnapshotRequest { + challenge: vec![7; 15].into(), + protocol_version: rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, + }); + let malformed_body = rustfs_protos::canonical_scanner_dirty_usage_snapshot_request_body(malformed.get_ref()) + .expect("scanner dirty usage snapshot request should encode"); + set_tonic_canonical_body_digest(&mut malformed, &malformed_body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut malformed); + let malformed = service + .scanner_dirty_usage_snapshot(malformed) + .await + .expect_err("malformed scanner dirty usage snapshot challenges must fail closed"); + assert_eq!(malformed.code(), tonic::Code::InvalidArgument); + + let challenge = [7; 16]; + let mut signed = Request::new(ScannerDirtyUsageSnapshotRequest { + challenge: challenge.to_vec().into(), + protocol_version: rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, + }); + let signed_body = rustfs_protos::canonical_scanner_dirty_usage_snapshot_request_body(signed.get_ref()) + .expect("scanner dirty usage snapshot request should encode"); + set_tonic_canonical_body_digest(&mut signed, &signed_body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut signed); + let response = service + .scanner_dirty_usage_snapshot(signed) + .await + .expect("an authenticated scanner dirty usage snapshot request should succeed") + .into_inner(); + assert_eq!(response.instance_id, rustfs_scanner::scanner_activity_epoch()); + assert_eq!(response.protocol_version, rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION); + let bucket_count = u64::try_from(response.buckets.len()).expect("snapshot bucket count should fit in u64"); + assert_eq!(response.complete, response.pending_bucket_count == bucket_count); + let canonical = rustfs_protos::canonical_scanner_dirty_usage_snapshot_response_body(&challenge, &response) + .expect("scanner dirty usage snapshot response should encode"); + crate::storage::storage_api::verify_tonic_rpc_response_proof(&canonical, &response.response_proof) + .expect("scanner dirty usage snapshot response proof should verify"); + } + #[test] fn test_scanner_activity_response_uses_process_epoch_and_generations() { let response = scanner_activity_response_v7(