diff --git a/crates/e2e_test/src/reliant/grpc_lock_server.rs b/crates/e2e_test/src/reliant/grpc_lock_server.rs index 93c872965..ccc466d23 100644 --- a/crates/e2e_test/src/reliant/grpc_lock_server.rs +++ b/crates/e2e_test/src/reliant/grpc_lock_server.rs @@ -409,6 +409,13 @@ impl NodeService for MinimalLockNodeService { Err(Status::unimplemented("lock-only test server")) } + async fn compare_and_update_file( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("lock-only test server")) + } + async fn delete( &self, _request: Request, diff --git a/crates/ecstore/src/cluster/rpc/client.rs b/crates/ecstore/src/cluster/rpc/client.rs index b6a3bda17..c10415b52 100644 --- a/crates/ecstore/src/cluster/rpc/client.rs +++ b/crates/ecstore/src/cluster/rpc/client.rs @@ -113,7 +113,7 @@ pub async fn tier_mutation_control_time_out_client( /// Build a `NodeServiceClient` bound to the [`ChannelClass`]-appropriate channel for `addr`. /// -/// Bulk `bytes`-carrying RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion) pass +/// Bulk `bytes`-carrying RPCs (ReadAll/WriteAll/CompareAndUpdateFile/ReadMultiple/BatchReadVersion) pass /// [`ChannelClass::Bulk`] so, when channel isolation is enabled, they are physically isolated /// from lock/health RPCs; everything else uses [`ChannelClass::Control`]. When isolation is /// disabled the two classes resolve to the same cached channel, i.e. legacy behavior. diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 35a5b41b5..a2a2e22ea 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -39,11 +39,12 @@ use http::{HeaderMap, HeaderValue, Method, Uri}; use rustfs_credentials::{DEFAULT_SECRET_KEY, RPC_SECRET_REQUIRED_MESSAGE}; use rustfs_credentials::{RPC_SECRET_REQUIRED_OPERATOR_MESSAGE, try_get_rpc_token}; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_OPERATION_GRPC_FORCE_UNLOCK, INTERNODE_OPERATION_GRPC_LOCK, - INTERNODE_OPERATION_GRPC_LOCK_BATCH, INTERNODE_OPERATION_GRPC_OTHER, INTERNODE_OPERATION_GRPC_READ_ALL, - INTERNODE_OPERATION_GRPC_READ_MULTIPLE, INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_OPERATION_GRPC_REFRESH, - INTERNODE_OPERATION_GRPC_UNLOCK, INTERNODE_OPERATION_GRPC_UNLOCK_BATCH, INTERNODE_OPERATION_GRPC_WRITE_ALL, - INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, + INTERNODE_OPERATION_GRPC_FORCE_UNLOCK, INTERNODE_OPERATION_GRPC_LOCK, INTERNODE_OPERATION_GRPC_LOCK_BATCH, + INTERNODE_OPERATION_GRPC_OTHER, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, + INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_OPERATION_GRPC_REFRESH, INTERNODE_OPERATION_GRPC_UNLOCK, + INTERNODE_OPERATION_GRPC_UNLOCK_BATCH, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, + global_internode_metrics, }; use rustfs_object_data_cache::{MemoryBasis, resolve_effective_memory}; use rustfs_utils::get_env_bool; @@ -1077,6 +1078,7 @@ fn tonic_rpc_metric_operation(path: &str) -> &'static str { Some("ReadVersion") => INTERNODE_OPERATION_GRPC_READ_VERSION, Some("BatchReadVersion") => INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, Some("WriteAll") => INTERNODE_OPERATION_GRPC_WRITE_ALL, + Some("CompareAndUpdateFile") => INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, Some("Lock") => INTERNODE_OPERATION_GRPC_LOCK, Some("UnLock") => INTERNODE_OPERATION_GRPC_UNLOCK, Some("LockBatch") => INTERNODE_OPERATION_GRPC_LOCK_BATCH, @@ -2713,6 +2715,10 @@ mod tests { tonic_rpc_metric_operation("/node_service.NodeService/WriteAll"), INTERNODE_OPERATION_GRPC_WRITE_ALL ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/CompareAndUpdateFile"), + INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE + ); assert_eq!( tonic_rpc_metric_operation("/node_service.NodeService/Lock"), INTERNODE_OPERATION_GRPC_LOCK diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 627d04a24..c1451332b 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -22,9 +22,10 @@ use crate::cluster::rpc::internode_data_transport::{ }; use crate::disk::error::{Error, Result}; use crate::disk::{ - BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, - DiskOption, FileInfoVersions, FileReader, FileWriter, PartTransactionAction, ReadMultipleReq, ReadMultipleResp, ReadOptions, - RenameDataResp, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, batch_read_version_one_by_one, + BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, ConditionalFileUpdate, DeleteOptions, DiskAPI, DiskInfo, + DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions, FileReader, FileWriter, PartTransactionAction, ReadMultipleReq, + ReadMultipleResp, ReadOptions, RenameDataResp, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, + batch_read_version_one_by_one, disk_store::{ DEFAULT_RUSTFS_DRIVE_ACTIVE_MONITORING, ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, SKIP_IF_SUCCESS_BEFORE, get_drive_active_check_interval, get_drive_active_check_timeout, get_drive_disk_info_timeout, get_drive_list_dir_timeout, @@ -54,13 +55,14 @@ use rustfs_protos::ChannelClass; use rustfs_protos::evict_failed_connection; use rustfs_protos::proto_gen::node_service::RenamePartRequest; use rustfs_protos::proto_gen::node_service::{ - BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, DeletePathsRequest, DeleteRequest, - DeleteVersionRequest, DeleteVersionsRequest, DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, - ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, - ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, - RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, - SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, - WriteMetadataRequest, node_service_client::NodeServiceClient, + BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, CompareAndUpdateFileOutcome, + CompareAndUpdateFileRequest, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, + DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest, MakeVolumeRequest, + MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest, ReadMultipleRequest, + ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, RenameFileRequest, + SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest, + SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest, + node_service_client::NodeServiceClient, }; use serde::{Serialize, de::DeserializeOwned}; use std::{ @@ -150,6 +152,18 @@ fn snapshot_lease_token_from_response(response: SnapshotLeaseResponse) -> Result SnapshotLeaseToken::from_slice(&response.token) } +fn conditional_file_update_from_wire(outcome: i32) -> Result { + match CompareAndUpdateFileOutcome::try_from(outcome) { + Ok(CompareAndUpdateFileOutcome::CompareAndUpdateFileUpdated) => Ok(ConditionalFileUpdate::Updated), + Ok(CompareAndUpdateFileOutcome::CompareAndUpdateFileMissing) => Ok(ConditionalFileUpdate::Missing), + Ok(CompareAndUpdateFileOutcome::CompareAndUpdateFileMismatch) => Ok(ConditionalFileUpdate::Mismatch), + Ok(CompareAndUpdateFileOutcome::CompareAndUpdateFileUnspecified) => { + Err(Error::other("unspecified compare-and-update-file outcome")) + } + Err(_) => Err(Error::other("invalid compare-and-update-file outcome")), + } +} + /// Bind a mutating disk RPC to its canonical body: the digest lands in the request metadata, and /// the signing interceptor folds it (plus a replay-protected nonce) into the v2 signature scope /// (backlog#1327). @@ -1836,7 +1850,7 @@ impl RemoteDisk { .map_err(|err| Error::RemoteClientUnavailable(err.to_string())) } - /// Client for large `bytes`-carrying RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion). + /// Client for large `bytes`-carrying RPCs (ReadAll/WriteAll/CompareAndUpdateFile/ReadMultiple/BatchReadVersion). /// Routes onto the isolated bulk channel pool so large transfers cannot head-of-line block /// lock/health RPCs (grpc-optimization P1). Falls back to the control channel when isolation /// is disabled. @@ -3816,6 +3830,58 @@ impl DiskAPI for RemoteDisk { .await } + async fn compare_and_update_file( + &self, + volume: &str, + path: &str, + expected: Option, + replacement: Option, + ) -> Result { + self.execute_with_timeout( + || async { + let data_len = expected + .as_ref() + .map_or(0, Bytes::len) + .saturating_add(replacement.as_ref().map_or(0, Bytes::len)); + let disk = self.disk_ref().await; + let mut client = self.get_bulk_client().await.inspect_err(|_| { + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_compare_and_update_file_error(); + })?; + let mut request = Request::new(CompareAndUpdateFileRequest { + disk, + volume: volume.to_string(), + path: path.to_string(), + expected, + replacement, + }); + let canonical_body = rustfs_protos::canonical_compare_and_update_file_request_body(request.get_ref()); + attach_mutation_body_digest(&mut request, canonical_body, "compare_and_update_file")?; + + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_compare_and_update_file_request(); + let response = match client.compare_and_update_file(request).await { + Ok(response) => response.into_inner(), + Err(err) => { + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_compare_and_update_file_error(); + return Err(err.into()); + } + }; + + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_compare_and_update_file_sent_bytes(data_len); + + if !response.success { + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_compare_and_update_file_error(); + return Err(response.error.unwrap_or_default().into()); + } + + conditional_file_update_from_wire(response.outcome).inspect_err(|_| { + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_compare_and_update_file_error(); + }) + }, + get_max_timeout_duration(), + ) + .await + } + #[tracing::instrument(level = "trace", skip_all)] async fn read_all(&self, volume: &str, path: &str) -> Result { trace!( @@ -3933,6 +3999,17 @@ mod tests { static INIT: Once = Once::new(); + #[test] + fn compare_and_update_wire_outcome_fails_closed() { + assert_eq!( + conditional_file_update_from_wire(CompareAndUpdateFileOutcome::CompareAndUpdateFileUpdated as i32) + .expect("updated outcome"), + ConditionalFileUpdate::Updated + ); + assert!(conditional_file_update_from_wire(CompareAndUpdateFileOutcome::CompareAndUpdateFileUnspecified as i32).is_err()); + assert!(conditional_file_update_from_wire(i32::MAX).is_err()); + } + #[test] fn request_compat_send_sites_keep_manifest_json_encoders() { // Rolling-upgrade contract (rustfs-protos compat manifest): every diff --git a/crates/ecstore/src/cluster/rpc/runtime_sources.rs b/crates/ecstore/src/cluster/rpc/runtime_sources.rs index ab9a34fe9..4ba4ef06d 100644 --- a/crates/ecstore/src/cluster/rpc/runtime_sources.rs +++ b/crates/ecstore/src/cluster/rpc/runtime_sources.rs @@ -14,10 +14,10 @@ use rustfs_io_metrics::internode_metrics::{ INTERNODE_MSGPACK_CODEC_JSON, INTERNODE_MSGPACK_CODEC_MSGPACK, INTERNODE_MSGPACK_DIRECTION_RESPONSE, - INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, - INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_OPERATION_PUT_FILE_STREAM, - INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_TRANSPORT_BACKEND_GRPC, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, - global_internode_metrics, + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_GRPC, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, }; use std::time::Duration; @@ -74,6 +74,28 @@ pub(crate) fn record_remote_disk_grpc_write_all_sent_bytes(bytes: usize) { ); } +pub(crate) fn record_remote_disk_grpc_compare_and_update_file_error() { + global_internode_metrics().record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); +} + +pub(crate) fn record_remote_disk_grpc_compare_and_update_file_request() { + global_internode_metrics().record_outgoing_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); +} + +pub(crate) fn record_remote_disk_grpc_compare_and_update_file_sent_bytes(bytes: usize) { + global_internode_metrics().record_sent_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, + INTERNODE_TRANSPORT_BACKEND_GRPC, + bytes, + ); +} + pub(crate) fn record_remote_disk_grpc_read_all_error() { global_internode_metrics() .record_error_for_operation_and_backend(INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC); diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 1fa9605f2..88cbe3389 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -1384,7 +1384,9 @@ impl HealManager { } async fn validate_replacement_recovery_records(disk: &crate::heal::DiskStore) -> Result<()> { - ResumeUtils::migrate_legacy_replacement_records(disk).await?; + if disk.endpoint().is_local { + ResumeUtils::migrate_legacy_replacement_records(disk).await?; + } for task_id in ResumeUtils::get_replacement_intent_tasks(disk).await? { ResumeManager::load_replacement_intent(disk.clone(), &task_id).await?; } diff --git a/crates/heal/src/heal/manager/auto_scan.rs b/crates/heal/src/heal/manager/auto_scan.rs index ec5cb84e0..fead6dc2f 100644 --- a/crates/heal/src/heal/manager/auto_scan.rs +++ b/crates/heal/src/heal/manager/auto_scan.rs @@ -83,37 +83,51 @@ impl HealManager { .lock() .expect("replacement recovery blocked set lock poisoned") .clone(); + let mut retry_recovery_disks = Vec::new(); if !blocked_sets.is_empty() { let mut retry_succeeded = HashSet::new(); let mut retry_failed = HashSet::new(); - for disk in &local_disks { - let endpoint = disk.endpoint(); - let Some(set_disk_id) = - crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx) - else { - continue; - }; - if !blocked_sets.contains(&set_disk_id) { - continue; - } - match Self::validate_replacement_recovery_records(disk).await { - Ok(()) => { - retry_succeeded.insert(set_disk_id); - } + for set_disk_id in blocked_sets { + let disks = match storage.replacement_intent_disks(&set_disk_id).await { + Ok(disks) => disks, Err(error) => { retry_failed.insert(set_disk_id.clone()); - conflicted_recovery_sets.insert(set_disk_id); + conflicted_recovery_sets.insert(set_disk_id.clone()); warn!( target: "rustfs::heal::manager", event = EVENT_HEAL_AUTO_SCAN_ENQUEUE, component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_DISK_SCANNER, - endpoint = %endpoint, + set_disk_id, + error = %error, + "Replacement recovery disk retry failed" + ); + continue; + } + }; + let mut valid = true; + for disk in &disks { + if let Err(error) = Self::validate_replacement_recovery_records(disk).await { + valid = false; + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_AUTO_SCAN_ENQUEUE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_DISK_SCANNER, + set_disk_id, + endpoint = %disk.endpoint(), error = %error, "Replacement recovery retry failed" ); } } + if valid { + retry_succeeded.insert(set_disk_id); + retry_recovery_disks.extend(disks); + } else { + retry_failed.insert(set_disk_id.clone()); + conflicted_recovery_sets.insert(set_disk_id); + } } let mut blocked = replacement_recovery_blocked_sets .lock() @@ -122,9 +136,9 @@ impl HealManager { } for disk in &local_disks { let endpoint = disk.endpoint(); - let runtime_state = disk.runtime_state(); - let set_disk_id = - crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx); + let runtime_state = disk.runtime_state(); + let set_disk_id = + crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx); if set_disk_id.as_ref().is_some_and(|set_disk_id| { replacement_recovery_blocked_sets .lock() @@ -212,6 +226,29 @@ impl HealManager { } } + let mut recovery_disks = local_disks.clone(); + recovery_disks.extend(retry_recovery_disks); + for set_disk_id in endpoints.keys() { + match storage.replacement_intent_disks(set_disk_id).await { + Ok(disks) => recovery_disks.extend(disks), + Err(error) => { + conflicted_recovery_sets.insert(set_disk_id.clone()); + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_AUTO_SCAN_ENQUEUE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_DISK_SCANNER, + set_disk_id, + error = %error, + "Replacement recovery disk discovery failed" + ); + } + } + } + recovery_disks.sort_by_key(|disk| disk.endpoint().to_string()); + recovery_disks + .dedup_by(|left, right| left.endpoint().to_string() == right.endpoint().to_string()); + // Once formatting succeeds a replacement is no longer // discoverable as UnformattedDisk. Re-admit exactly one // incomplete durable generation per set within its @@ -219,7 +256,7 @@ impl HealManager { // verified terminal cleanup. Multiple generations are a // durable conflict: leave every marker/state intact and // require reconciliation rather than choosing one. - for disk in &local_disks { + for disk in &recovery_disks { let endpoint = disk.endpoint(); let disk_set_disk_id = crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx); @@ -280,6 +317,21 @@ impl HealManager { } }; let state = resume_manager.get_state().await; + let targets = state + .replacement_targets + .iter() + .filter_map(|target| { + local_endpoints + .iter() + .find(|endpoint| endpoint.to_string() == *target) + .cloned() + }) + .collect::>(); + // A remote survivor may own the durable intent, but only the + // node mounting every replacement target may advance or replay it. + if targets.len() != state.replacement_targets.len() { + continue; + } if !durable_replacement_recovery_is_due(&state, &task_id) { if durable_replacement_reserves_targets(&state) { conflicted_recovery_sets.insert(state.set_disk_id.clone()); @@ -294,20 +346,6 @@ impl HealManager { } }; let task_id = state.task_id.clone(); - let targets = state - .replacement_targets - .iter() - .filter_map(|target| { - local_endpoints - .iter() - .find(|endpoint| endpoint.to_string() == *target) - .cloned() - }) - .collect::>(); - if targets.len() != state.replacement_targets.len() { - conflicted_recovery_sets.insert(state.set_disk_id.clone()); - continue; - } let Some(set_disk_id) = crate::heal::utils::format_set_disk_id_from_i32( targets[0].pool_idx, targets[0].set_idx, diff --git a/crates/heal/src/heal/manager/unclean_shutdown.rs b/crates/heal/src/heal/manager/unclean_shutdown.rs index ecf03dffa..4e6e035fa 100644 --- a/crates/heal/src/heal/manager/unclean_shutdown.rs +++ b/crates/heal/src/heal/manager/unclean_shutdown.rs @@ -54,6 +54,13 @@ pub(super) fn replacement_discovery_error_is_expected_for_deferred_endpoint( matches!(error, Error::Disk(DiskError::UnformattedDisk)) && deferred_replacement_endpoints.contains(endpoint) } +pub(super) fn replacement_targets_belong_to_local_node( + replacement_targets: &[String], + local_endpoints: &HashSet, +) -> bool { + !replacement_targets.is_empty() && replacement_targets.iter().all(|target| local_endpoints.contains(target)) +} + pub(super) fn unblock_replacement_recovery_sets_after_validation( blocked_sets: &mut HashSet, retry_succeeded: HashSet, @@ -83,6 +90,11 @@ impl HealManager { let local_disk_map = local_disk_map_read().await; local_disk_map.values().flatten().cloned().collect::>() }; + let local_endpoints = local_disks + .iter() + .map(|disk| disk.endpoint().to_string()) + .collect::>(); + let mut recovery_disks = Vec::new(); for disk in &local_disks { let endpoint = disk.endpoint(); match disk @@ -132,6 +144,11 @@ impl HealManager { set_disk_ids.insert(set_disk_id.clone()); } + if !matches!(disk.get_disk_id().await, Ok(Some(id)) if !id.is_nil()) { + continue; + } + recovery_disks.push(disk.clone()); + // Legacy flat records are inspected only while starting. The // periodic scanner lists the dedicated replacement directory. let migration = match ResumeUtils::migrate_approved_legacy_replacements(disk, self.storage.as_ref()).await { @@ -152,6 +169,31 @@ impl HealManager { "Legacy replacement recovery migration failed" ); } + } + + for set_disk_id in &set_disk_ids { + match self.storage.replacement_intent_disks(set_disk_id).await { + Ok(disks) => recovery_disks.extend(disks), + Err(error) => { + self.block_replacement_recovery_set(set_disk_id); + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_UNCLEAN_SHUTDOWN, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + set_disk_id, + error = %error, + "Replacement recovery disk discovery failed" + ); + } + } + } + recovery_disks.sort_by_key(|disk| disk.endpoint().to_string()); + recovery_disks.dedup_by(|left, right| left.endpoint().to_string() == right.endpoint().to_string()); + + for disk in &recovery_disks { + let endpoint = disk.endpoint(); + let disk_set_disk_id = crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx); let replacement_task_ids = match ResumeUtils::get_replacement_intent_tasks(disk).await { Ok(task_ids) => task_ids, Err(error) => { @@ -194,6 +236,9 @@ impl HealManager { if durable_replacement_reserves_targets(&state) { reserved_replacement_sets.insert(state.set_disk_id.clone()); } + if !replacement_targets_belong_to_local_node(&state.replacement_targets, &local_endpoints) { + continue; + } let active_replacement = !state.completed && matches!( state.replacement_phase, @@ -395,3 +440,25 @@ impl HealManager { } } } + +#[cfg(test)] +mod tests { + use super::replacement_targets_belong_to_local_node; + use std::collections::HashSet; + + #[test] + fn replacement_replay_requires_every_target_to_be_local() { + let local = HashSet::from(["http://node-b:9000/disk".to_string()]); + + assert!(replacement_targets_belong_to_local_node(&["http://node-b:9000/disk".to_string()], &local)); + assert!(!replacement_targets_belong_to_local_node(&[], &local)); + assert!(!replacement_targets_belong_to_local_node( + &["http://node-a:9000/disk".to_string()], + &local + )); + assert!(!replacement_targets_belong_to_local_node( + &["http://node-b:9000/disk".to_string(), "http://node-c:9000/disk".to_string(),], + &local + )); + } +} diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index afe625533..741314afc 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -30,7 +30,7 @@ use super::storage_api::storage::{ BucketInfo, BucketOperations, DiskSetSelector, EcstoreHealObjectStorageResult, HealOperations as _, ListOperations as _, ObjectIO as _, ObjectOperations as _, StorageAdminApi, }; -use super::{DiskStore, ECStore, HealDiskExt as _, StorageError, resume::ReplacementTargetIdentity}; +use super::{DiskError, DiskStore, ECStore, HealDiskExt as _, StorageError, resume::ReplacementTargetIdentity}; pub use super::{HealObjectInfo, HealObjectOptions, HealPutObjReader}; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] @@ -613,6 +613,13 @@ pub trait HealStorageAPI: Send + Sync { Err(Error::other("target-excluding resume disk selection is unsupported")) } + /// Return every formatted disk that may durably host or expose a + /// replacement intent. Implementations should order local disks first so + /// discovery remains deterministic and preserves the local preference. + async fn replacement_intent_disks(&self, _set_disk_id: &str) -> Result> { + Ok(Vec::new()) + } + /// Reopen the exact surviving disk that owns an existing replacement /// intent. Falling back to another disk would create a second copy of the /// same generation and split its progress. @@ -645,6 +652,16 @@ impl ECStoreHealStorage { Self { ecstore } } + async fn resume_disk_inventory(&self, set_disk_id: &str) -> Result> { + let (pool_idx, set_idx) = crate::heal::utils::parse_set_disk_id(set_disk_id)?; + let disks = StorageAdminApi::disk_set_inventory(self.ecstore.as_ref(), DiskSetSelector::new(pool_idx, set_idx)) + .await + .map_err(|e| Error::TaskExecutionFailed { + message: format!("Failed to get disks for pool {pool_idx} set {set_idx}: {e}"), + })?; + Ok(disks.into_iter().flatten().collect()) + } + async fn object_result_with_receipt( &self, bucket: &str, @@ -1699,7 +1716,14 @@ impl HealStorageAPI for ECStoreHealStorage { } async fn get_disk_for_resume(&self, set_disk_id: &str) -> Result { - self.get_disk_for_resume_excluding(set_disk_id, &[]).await + for disk in self.resume_disk_inventory(set_disk_id).await? { + if disk.endpoint().is_local && matches!(disk.get_disk_id().await, Ok(Some(id)) if !id.is_nil()) { + return Ok(disk); + } + } + Err(Error::TaskExecutionFailed { + message: format!("No available disk found for set_disk_id: {set_disk_id}"), + }) } async fn get_disk_for_resume_excluding(&self, set_disk_id: &str, excluded_targets: &[String]) -> Result { @@ -1714,40 +1738,30 @@ impl HealStorageAPI for ECStoreHealStorage { "Heal storage admin operation started" ); - // Parse set_disk_id to extract pool and set indices - let (pool_idx, set_idx) = crate::heal::utils::parse_set_disk_id(set_disk_id)?; - - // Get the first available disk from the set - let disks = StorageAdminApi::disk_set_inventory(self.ecstore.as_ref(), DiskSetSelector::new(pool_idx, set_idx)) - .await - .map_err(|e| Error::TaskExecutionFailed { - message: format!("Failed to get disks for pool {pool_idx} set {set_idx}: {e}"), - })?; - // The replacement target is unformatted before repair and must never // host the intent that authorizes its own formatting. - for disk_store in disks.into_iter().flatten() { - if !disk_store.endpoint().is_local { - continue; + let disks = self.resume_disk_inventory(set_disk_id).await?; + for prefer_local in [true, false] { + for disk_store in &disks { + if disk_store.endpoint().is_local != prefer_local + || excluded_targets.contains(&disk_store.endpoint().to_string()) + || !matches!(disk_store.get_disk_id().await, Ok(Some(id)) if !id.is_nil()) + { + continue; + } + debug!( + target: "rustfs::heal::storage", + event = EVENT_HEAL_STORAGE_ADMIN_OP, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "get_disk_for_resume", + set_disk_id, + result = "ok", + disk = ?disk_store, + "Heal storage resume disk resolved" + ); + return Ok(disk_store.clone()); } - if excluded_targets.contains(&disk_store.endpoint().to_string()) { - continue; - } - if !matches!(disk_store.get_disk_id().await, Ok(Some(id)) if !id.is_nil()) { - continue; - } - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_ADMIN_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "get_disk_for_resume", - set_disk_id, - result = "ok", - disk = ?disk_store, - "Heal storage resume disk resolved" - ); - return Ok(disk_store); } Err(Error::TaskExecutionFailed { @@ -1755,32 +1769,62 @@ impl HealStorageAPI for ECStoreHealStorage { }) } + async fn replacement_intent_disks(&self, set_disk_id: &str) -> Result> { + let disks = self.resume_disk_inventory(set_disk_id).await?; + let mut formatted = Vec::with_capacity(disks.len()); + for prefer_local in [true, false] { + for disk in &disks { + if disk.endpoint().is_local != prefer_local { + continue; + } + match disk.get_disk_id().await { + Ok(Some(id)) if !id.is_nil() => formatted.push(disk.clone()), + Ok(_) | Err(DiskError::UnformattedDisk) => {} + Err(error) => { + return Err(Error::TaskExecutionFailed { + message: format!( + "Failed to inspect replacement intent disk {} for set_disk_id {set_disk_id}: {error}", + disk.endpoint() + ), + }); + } + } + } + } + Ok(formatted) + } + async fn get_replacement_resume_disk( &self, set_disk_id: &str, task_id: &str, excluded_targets: &[String], ) -> Result { - let (pool_idx, set_idx) = crate::heal::utils::parse_set_disk_id(set_disk_id)?; - let disks = StorageAdminApi::disk_set_inventory(self.ecstore.as_ref(), DiskSetSelector::new(pool_idx, set_idx)) - .await - .map_err(|e| Error::TaskExecutionFailed { - message: format!("Failed to get disks for pool {pool_idx} set {set_idx}: {e}"), - })?; + let disks = self.resume_disk_inventory(set_disk_id).await?; let mut existing = None; - for disk_store in disks.into_iter().flatten() { - if !disk_store.endpoint().is_local || excluded_targets.contains(&disk_store.endpoint().to_string()) { - continue; + for prefer_local in [true, false] { + for disk_store in &disks { + if disk_store.endpoint().is_local != prefer_local + || excluded_targets.contains(&disk_store.endpoint().to_string()) + || !matches!(disk_store.get_disk_id().await, Ok(Some(id)) if !id.is_nil()) + { + continue; + } + let replacement_tasks = super::resume::ResumeUtils::get_replacement_intent_tasks(disk_store).await?; + let has_intent = replacement_tasks.iter().any(|candidate| candidate == task_id) + || (prefer_local && super::resume::ResumeManager::has_replacement_intent(disk_store, task_id).await); + if has_intent && existing.replace(disk_store.clone()).is_some() { + return Err(Error::TaskExecutionFailed { + message: format!("Replacement resume intent is duplicated for set_disk_id: {set_disk_id}"), + }); + } } - if !matches!(disk_store.get_disk_id().await, Ok(Some(id)) if !id.is_nil()) { - continue; - } - if super::resume::ResumeManager::has_replacement_intent(&disk_store, task_id).await - && existing.replace(disk_store).is_some() - { - return Err(Error::TaskExecutionFailed { - message: format!("Replacement resume intent is duplicated for set_disk_id: {set_disk_id}"), - }); + // Existing releases only write replacement intents locally, and + // this release also prefers a local survivor. Keep that path free + // of remote availability and latency; remote discovery is needed + // only when no local survivor owns the task. + if existing.is_some() { + break; } } Ok(existing.map_or(ReplacementResumeDisk::Fresh, ReplacementResumeDisk::Existing)) diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 24709b82a..2bc9b1ac1 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -27,6 +27,7 @@ pub const INTERNODE_OPERATION_WALK_DIR: &str = "walk_dir"; pub const INTERNODE_OPERATION_NS_SCANNER: &str = "ns_scanner"; pub const INTERNODE_OPERATION_GRPC_READ_ALL: &str = "grpc_read_all"; pub const INTERNODE_OPERATION_GRPC_WRITE_ALL: &str = "grpc_write_all"; +pub const INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE: &str = "grpc_compare_and_update_file"; pub const INTERNODE_OPERATION_GRPC_READ_MULTIPLE: &str = "grpc_read_multiple"; pub const INTERNODE_OPERATION_GRPC_READ_VERSION: &str = "grpc_read_version"; pub const INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION: &str = "grpc_batch_read_version"; @@ -1324,6 +1325,7 @@ mod tests { assert_eq!(INTERNODE_OPERATION_WALK_DIR, "walk_dir"); assert_eq!(INTERNODE_OPERATION_GRPC_READ_ALL, "grpc_read_all"); assert_eq!(INTERNODE_OPERATION_GRPC_WRITE_ALL, "grpc_write_all"); + assert_eq!(INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, "grpc_compare_and_update_file"); assert_eq!(INTERNODE_OPERATION_GRPC_READ_VERSION, "grpc_read_version"); assert_eq!(INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, "grpc_batch_read_version"); assert_eq!(INTERNODE_OPERATION_GRPC_LOCK, "grpc_lock"); diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index 7f7b9d2b1..f7372c037 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -134,6 +134,29 @@ pub struct WriteAllResponse { pub error: ::core::option::Option, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct CompareAndUpdateFileRequest { + /// indicate which one in the disks + #[prost(string, tag = "1")] + pub disk: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub volume: ::prost::alloc::string::String, + #[prost(string, tag = "3")] + pub path: ::prost::alloc::string::String, + #[prost(bytes = "bytes", optional, tag = "4")] + pub expected: ::core::option::Option<::prost::bytes::Bytes>, + #[prost(bytes = "bytes", optional, tag = "5")] + pub replacement: ::core::option::Option<::prost::bytes::Bytes>, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct CompareAndUpdateFileResponse { + #[prost(bool, tag = "1")] + pub success: bool, + #[prost(enumeration = "CompareAndUpdateFileOutcome", tag = "2")] + pub outcome: i32, + #[prost(message, optional, tag = "3")] + pub error: ::core::option::Option, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct DeleteRequest { /// indicate which one in the disks #[prost(string, tag = "1")] @@ -1702,6 +1725,38 @@ impl ControlPlaneErrorCode { } #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] #[repr(i32)] +pub enum CompareAndUpdateFileOutcome { + CompareAndUpdateFileUnspecified = 0, + CompareAndUpdateFileUpdated = 1, + CompareAndUpdateFileMissing = 2, + CompareAndUpdateFileMismatch = 3, +} +impl CompareAndUpdateFileOutcome { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::CompareAndUpdateFileUnspecified => "COMPARE_AND_UPDATE_FILE_UNSPECIFIED", + Self::CompareAndUpdateFileUpdated => "COMPARE_AND_UPDATE_FILE_UPDATED", + Self::CompareAndUpdateFileMissing => "COMPARE_AND_UPDATE_FILE_MISSING", + Self::CompareAndUpdateFileMismatch => "COMPARE_AND_UPDATE_FILE_MISMATCH", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "COMPARE_AND_UPDATE_FILE_UNSPECIFIED" => Some(Self::CompareAndUpdateFileUnspecified), + "COMPARE_AND_UPDATE_FILE_UPDATED" => Some(Self::CompareAndUpdateFileUpdated), + "COMPARE_AND_UPDATE_FILE_MISSING" => Some(Self::CompareAndUpdateFileMissing), + "COMPARE_AND_UPDATE_FILE_MISMATCH" => Some(Self::CompareAndUpdateFileMismatch), + _ => None, + } + } +} +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] +#[repr(i32)] pub enum TierMutationPeerState { Unspecified = 0, Prepared = 1, @@ -1976,6 +2031,21 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "WriteAll")); self.inner.unary(req, path, codec).await } + pub async fn compare_and_update_file( + &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/CompareAndUpdateFile"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "CompareAndUpdateFile")); + self.inner.unary(req, path, codec).await + } pub async fn delete( &mut self, request: impl tonic::IntoRequest, @@ -3375,6 +3445,10 @@ pub mod node_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn compare_and_update_file( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn delete( &self, request: tonic::Request, @@ -4073,6 +4147,34 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/CompareAndUpdateFile" => { + #[allow(non_camel_case_types)] + struct CompareAndUpdateFileSvc(pub Arc); + impl tonic::server::UnaryService for CompareAndUpdateFileSvc { + type Response = super::CompareAndUpdateFileResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::compare_and_update_file(&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 = CompareAndUpdateFileSvc(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/Delete" => { #[allow(non_camel_case_types)] struct DeleteSvc(pub Arc); diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index fc806c006..0c930df9a 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -1087,6 +1087,20 @@ pub fn canonical_write_all_request_body( Ok(body.finish()) } +pub fn canonical_compare_and_update_file_request_body( + request: &proto_gen::node_service::CompareAndUpdateFileRequest, +) -> Result, std::num::TryFromIntError> { + let mut body = CanonicalBodyBuilder::new(b"rustfs-compare-and-update-file-request-v1\0"); + body.push_str(&request.disk)?; + body.push_str(&request.volume)?; + body.push_str(&request.path)?; + body.push_bool(request.expected.is_some()); + body.push_bytes(request.expected.as_deref().unwrap_or_default())?; + body.push_bool(request.replacement.is_some()); + body.push_bytes(request.replacement.as_deref().unwrap_or_default())?; + Ok(body.finish()) +} + pub fn canonical_delete_request_body( request: &proto_gen::node_service::DeleteRequest, ) -> Result, std::num::TryFromIntError> { @@ -1242,10 +1256,10 @@ pub fn canonical_make_volumes_request_body( #[cfg(test)] mod disk_mutation_canonical_tests { use super::proto_gen::node_service::{ - DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, MakeVolumeRequest, - MakeVolumesRequest, PreparePartTransactionRequest, RenameDataRequest, RenameFileRequest, RenamePartRequest, - SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest, - UpdateMetadataRequest, WriteAllRequest, WriteMetadataRequest, + CompareAndUpdateFileRequest, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, + DeleteVolumeRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, RenameDataRequest, + RenameFileRequest, RenamePartRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, + SnapshotLeaseRenewRequest, SnapshotLeaseRequest, UpdateMetadataRequest, WriteAllRequest, WriteMetadataRequest, }; use super::*; @@ -1443,6 +1457,31 @@ mod disk_mutation_canonical_tests { } assert_all_distinct(&bodies); + let compare_and_update_file = CompareAndUpdateFileRequest { + disk: "d".into(), + volume: "v".into(), + path: "p".into(), + expected: Some(vec![0xAA, 0xBB].into()), + replacement: Some(vec![0xCC, 0xDD].into()), + }; + let mut bodies = vec![canonical_compare_and_update_file_request_body(&compare_and_update_file).unwrap()]; + for mutate in [ + |r: &mut CompareAndUpdateFileRequest| r.disk = "d2".into(), + |r: &mut CompareAndUpdateFileRequest| r.volume = "v2".into(), + |r: &mut CompareAndUpdateFileRequest| r.path = "p2".into(), + |r: &mut CompareAndUpdateFileRequest| r.expected = None, + |r: &mut CompareAndUpdateFileRequest| r.expected = Some(Vec::new().into()), + |r: &mut CompareAndUpdateFileRequest| r.expected = Some(vec![0xAA, 0xBC].into()), + |r: &mut CompareAndUpdateFileRequest| r.replacement = None, + |r: &mut CompareAndUpdateFileRequest| r.replacement = Some(Vec::new().into()), + |r: &mut CompareAndUpdateFileRequest| r.replacement = Some(vec![0xCC, 0xDE].into()), + ] { + let mut request = compare_and_update_file.clone(); + mutate(&mut request); + bodies.push(canonical_compare_and_update_file_request_body(&request).unwrap()); + } + assert_all_distinct(&bodies); + let delete = DeleteRequest { bucket_incarnation_id: Default::default(), disk: "d".into(), diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index 8608c3636..302bbe103 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -121,6 +121,27 @@ message WriteAllResponse { optional Error error = 2; } +message CompareAndUpdateFileRequest { + string disk = 1; // indicate which one in the disks + string volume = 2; + string path = 3; + optional bytes expected = 4; + optional bytes replacement = 5; +} + +enum CompareAndUpdateFileOutcome { + COMPARE_AND_UPDATE_FILE_UNSPECIFIED = 0; + COMPARE_AND_UPDATE_FILE_UPDATED = 1; + COMPARE_AND_UPDATE_FILE_MISSING = 2; + COMPARE_AND_UPDATE_FILE_MISMATCH = 3; +} + +message CompareAndUpdateFileResponse { + bool success = 1; + CompareAndUpdateFileOutcome outcome = 2; + optional Error error = 3; +} + message DeleteRequest { string disk = 1; // indicate which one in the disks string volume = 2; @@ -1190,6 +1211,7 @@ service NodeService { rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {}; // auth-policy: read-only rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {}; // auth-policy: body-bound + rpc CompareAndUpdateFile(CompareAndUpdateFileRequest) returns (CompareAndUpdateFileResponse) {}; // auth-policy: body-bound rpc Delete(DeleteRequest) returns (DeleteResponse) {}; // auth-policy: body-bound rpc DeleteAtIncarnation(DeleteRequest) returns (DeleteResponse) {}; // auth-policy: body-bound rpc AcquireSnapshotLease(SnapshotLeaseRequest) returns (SnapshotLeaseResponse) {}; // auth-policy: body-bound diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index afe9c3a0b..2f115b939 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -59,8 +59,9 @@ use opentelemetry::global; use opentelemetry::trace::TraceContextExt; use rustfs_common::GlobalReadiness; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_OPERATION_GRPC_OTHER, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, - INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, + INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, INTERNODE_OPERATION_GRPC_OTHER, INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_OPERATION_GRPC_READ_MULTIPLE, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, + global_internode_metrics, }; use rustfs_keystone::KeystoneAuthLayer; #[cfg(feature = "swift")] @@ -2327,6 +2328,7 @@ fn check_auth(req: Request<()>) -> std::result::Result, Status> { "ReadAll" => INTERNODE_OPERATION_GRPC_READ_ALL, "ReadMultiple" => INTERNODE_OPERATION_GRPC_READ_MULTIPLE, "WriteAll" => INTERNODE_OPERATION_GRPC_WRITE_ALL, + "CompareAndUpdateFile" => INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, _ => INTERNODE_OPERATION_GRPC_OTHER, }; global_internode_metrics().record_rpc_auth_failure_for_operation_and_backend( diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 6e0bfedd7..c5c5e025c 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -1442,6 +1442,13 @@ impl Node for NodeService { self.handle_write_all(request).await } + async fn compare_and_update_file( + &self, + request: Request, + ) -> Result, Status> { + self.handle_compare_and_update_file(request).await + } + async fn delete(&self, request: Request) -> Result, Status> { if !request.get_ref().bucket_incarnation_id.is_empty() { return Err(Status::invalid_argument("incarnation-bound mutation requires its dedicated RPC")); @@ -2975,7 +2982,11 @@ mod tests { ecstore_layout::{EndpointServerPools, Endpoints, PoolEndpoints}, }; use bytes::Bytes; - use rustfs_heal::heal::{manager::HealManager, storage::HealStorageAPI}; + use rustfs_heal::heal::{ + manager::HealManager, + resume::{ReplacementTargetIdentity, ResumeManager, ResumeUtils}, + storage::{ECStoreHealStorage, HealStorageAPI, ReplacementResumeDisk}, + }; use rustfs_iam::{ store::{ Store as _, @@ -2989,18 +3000,19 @@ mod tests { use rustfs_protos::proto_gen::node_service::scanner_control_service_server::ScannerControlService as _; use rustfs_protos::proto_gen::node_service::{ BackgroundHealStatusRequest, BatchGenerallyLockRequest, CancelDecommissionRequest, CheckPartsRequest, - ClearDecommissionRequest, ControlPlaneErrorCode, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest, - DeletePolicyRequest, DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest, - DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest, - GetAllBucketStatsRequest, GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest, - GetMetacacheListingRequest, GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, - GetProcInfoRequest, GetSeLinuxInfoRequest, GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest, - HealBucketRequest, HealControlRequest, HealControlResponse, ListBucketRequest, ListDirRequest, ListVolumesRequest, - LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, - LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, - MakeVolumeRequest, MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, - ReadMultipleRequest, ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, - RenameDataRequest, RenameFileRequest, RenamePartRequest, ScannerActivityRequest, ScannerDirtyUsageSnapshotRequest, + ClearDecommissionRequest, CompareAndUpdateFileRequest, ControlPlaneErrorCode, DeleteBucketMetadataRequest, + DeleteBucketRequest, DeletePathsRequest, DeletePolicyRequest, DeleteRequest, DeleteServiceAccountRequest, + DeleteUserRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, + DownloadProfileDataRequest, GenerallyLockRequest, GetAllBucketStatsRequest, GetBucketInfoRequest, + GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest, GetMetacacheListingRequest, GetMetricsRequest, + GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest, + GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest, HealBucketRequest, HealControlRequest, + HealControlResponse, ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest, LoadGroupRequest, + LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest, + LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, MakeVolumeRequest, + MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest, + ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest, + RenameFileRequest, RenamePartRequest, ScannerActivityRequest, ScannerDirtyUsageSnapshotRequest, ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest, ServerInfoRequest, SettlePartTransactionRequest, SignalServiceRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest, StartDecommissionRequest, StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationAbortRequest, @@ -3023,7 +3035,7 @@ mod tests { use tonic::{Request, Response, Status}; use uuid::Uuid; - const DISK_MUTATION_RPC_METHODS: [&str; 23] = [ + const DISK_MUTATION_RPC_METHODS: [&str; 24] = [ "renamedata", "renamedataatincarnation", "writemetadataatincarnation", @@ -3035,6 +3047,7 @@ mod tests { "writemetadata", "updatemetadata", "writeall", + "compareandupdatefile", "delete", "deletepaths", "renamefile", @@ -4485,6 +4498,17 @@ mod tests { }, rustfs_protos::canonical_write_all_request_body ); + assert_gated!( + compare_and_update_file, + CompareAndUpdateFileRequest { + disk: disk.clone(), + volume: "v".into(), + path: "p".into(), + expected: Some(vec![0x01].into()), + replacement: Some(vec![0x02].into()), + }, + rustfs_protos::canonical_compare_and_update_file_request_body + ); assert_gated!( delete, DeleteRequest { @@ -5947,6 +5971,265 @@ mod tests { .expect("bounded real fixture initialization") } + fn signed_compare_and_update_file_request( + disk: String, + volume: &str, + path: &str, + expected: Option, + replacement: Option, + ) -> Request { + let mut request = Request::new(CompareAndUpdateFileRequest { + disk, + volume: volume.to_string(), + path: path.to_string(), + expected, + replacement, + }); + let body = rustfs_protos::canonical_compare_and_update_file_request_body(request.get_ref()) + .expect("compare-and-update request body"); + set_tonic_canonical_body_digest(&mut request, &body).expect("compare-and-update digest"); + mark_v2_authenticated(&mut request); + request + } + + #[tokio::test] + async fn compare_and_update_file_handler_preserves_atomic_outcomes() { + use rustfs_protos::proto_gen::node_service::CompareAndUpdateFileOutcome; + + let fixture = target_rpc_fixture().await; + let set = fixture + .env + .ecstore + .all_set_disks() + .into_iter() + .next() + .expect("target erasure set"); + let disk = set.disks.read().await.iter().find_map(Clone::clone).expect("local target"); + let service = make_server_for_context(Some(fixture.context.clone())); + let volume = crate::storage::storage_api::ecstore_disk::RUSTFS_META_BUCKET; + if let Err(error) = disk.make_volume(volume).await { + assert_eq!(error, DiskError::VolumeExists, "create metadata volume"); + } + let path = format!("compare-and-update-{}", Uuid::new_v4()); + let old = Bytes::from_static(b"old-intent"); + let new = Bytes::from_static(b"new-intent"); + + let created = service + .compare_and_update_file(signed_compare_and_update_file_request( + disk.endpoint().to_string(), + volume, + &path, + None, + Some(old.clone()), + )) + .await + .expect("create compare-and-update response") + .into_inner(); + assert!(created.success, "create compare-and-update failed: {:?}", created.error); + assert_eq!( + CompareAndUpdateFileOutcome::try_from(created.outcome).expect("created outcome"), + CompareAndUpdateFileOutcome::CompareAndUpdateFileUpdated + ); + + let mismatched = service + .compare_and_update_file(signed_compare_and_update_file_request( + disk.endpoint().to_string(), + volume, + &path, + None, + Some(new.clone()), + )) + .await + .expect("mismatched compare-and-update response") + .into_inner(); + assert!(mismatched.success); + assert_eq!( + CompareAndUpdateFileOutcome::try_from(mismatched.outcome).expect("mismatched outcome"), + CompareAndUpdateFileOutcome::CompareAndUpdateFileMismatch + ); + assert_eq!(disk.read_all(volume, &path).await.expect("unchanged intent"), old); + + let updated = service + .compare_and_update_file(signed_compare_and_update_file_request( + disk.endpoint().to_string(), + volume, + &path, + Some(old), + Some(new.clone()), + )) + .await + .expect("update compare-and-update response") + .into_inner(); + assert!(updated.success); + assert_eq!( + CompareAndUpdateFileOutcome::try_from(updated.outcome).expect("updated outcome"), + CompareAndUpdateFileOutcome::CompareAndUpdateFileUpdated + ); + assert_eq!(disk.read_all(volume, &path).await.expect("updated intent"), new.clone()); + + let deleted = service + .compare_and_update_file(signed_compare_and_update_file_request( + disk.endpoint().to_string(), + volume, + &path, + Some(new.clone()), + None, + )) + .await + .expect("delete compare-and-update response") + .into_inner(); + assert!(deleted.success); + assert_eq!( + CompareAndUpdateFileOutcome::try_from(deleted.outcome).expect("deleted outcome"), + CompareAndUpdateFileOutcome::CompareAndUpdateFileUpdated + ); + + let missing = service + .compare_and_update_file(signed_compare_and_update_file_request( + disk.endpoint().to_string(), + volume, + &path, + Some(new), + None, + )) + .await + .expect("missing compare-and-update response") + .into_inner(); + assert!(missing.success); + assert_eq!( + CompareAndUpdateFileOutcome::try_from(missing.outcome).expect("missing outcome"), + CompareAndUpdateFileOutcome::CompareAndUpdateFileMissing + ); + assert!(matches!(disk.read_all(volume, &path).await, Err(DiskError::FileNotFound))); + } + + #[cfg(not(windows))] + #[tokio::test] + async fn replacement_resume_selection_falls_back_to_a_remote_survivor() { + use crate::storage::storage_api::{ + ecstore_disk::{DiskOption, new_disk}, + init_local_disks_with_instance_ctx, + }; + + let fixture = target_rpc_fixture().await; + let _ = rustfs_credentials::set_global_rpc_secret(Uuid::new_v4().to_string()); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind replacement intent target"); + let addr = listener.local_addr().expect("replacement intent target address"); + let set = fixture.env.ecstore.all_set_disks().into_iter().next().expect("erasure set"); + let original_disks = set.disks.read().await.clone(); + let mut endpoints = fixture.env.endpoint_pools.as_ref()[0].endpoints.as_ref().clone(); + let mut survivor_endpoint = Endpoint::try_from(format!("http://{addr}{}", fixture.env.disk_paths[0].display()).as_str()) + .expect("remote survivor endpoint"); + survivor_endpoint.set_pool_index(0); + survivor_endpoint.set_set_index(0); + survivor_endpoint.set_disk_index(0); + survivor_endpoint.is_local = true; + endpoints[0] = survivor_endpoint.clone(); + let mut pool = fixture.env.endpoint_pools.as_ref()[0].clone(); + pool.endpoints = Endpoints::from(endpoints.clone()); + init_local_disks_with_instance_ctx(&fixture.instance, EndpointServerPools::from(vec![pool])) + .await + .expect("register replacement intent target disks"); + let service = make_server_for_context(Some(fixture.context.clone())); + let survivor = service + .find_disk(&survivor_endpoint.to_string()) + .await + .expect("registered survivor disk"); + if let Err(error) = survivor + .make_volume(crate::storage::storage_api::ecstore_disk::RUSTFS_META_BUCKET) + .await + { + assert_eq!(error, DiskError::VolumeExists, "create survivor metadata volume"); + } + let replacement = service + .find_disk(&endpoints[1].to_string()) + .await + .expect("registered replacement disk"); + let (shutdown, stopped) = tokio::sync::oneshot::channel(); + let server = tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(NodeServiceServer::new(service)) + .serve_with_incoming_shutdown(TcpListenerStream::new(listener), async { + let _ = stopped.await; + }) + .await + .expect("replacement intent target server"); + }); + + survivor_endpoint.is_local = false; + let remote_survivor = new_disk( + &survivor_endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("remote survivor client"); + let survivor_id = survivor + .get_disk_id() + .await + .expect("survivor disk id") + .filter(|id| !id.is_nil()) + .expect("formatted survivor disk id"); + remote_survivor + .set_disk_id(Some(survivor_id)) + .await + .expect("seed remote survivor disk id"); + *set.disks.write().await = vec![Some(remote_survivor.clone()), Some(replacement.clone()), None, None]; + + let storage = ECStoreHealStorage::new(fixture.env.ecstore.clone()); + let replacement_endpoint = replacement.endpoint().to_string(); + let selected = storage + .get_disk_for_resume_excluding("pool_0_set_0", std::slice::from_ref(&replacement_endpoint)) + .await + .expect("remote survivor should host the replacement intent"); + assert!(!selected.endpoint().is_local); + assert_eq!(selected.endpoint().to_string(), remote_survivor.endpoint().to_string()); + + let task_id = ResumeUtils::generate_task_id(); + ResumeManager::new_replacement_intent( + selected, + task_id.clone(), + "pool_0_set_0".to_string(), + vec!["bucket-a".to_string()], + vec![replacement_endpoint.clone()], + vec![ReplacementTargetIdentity { + endpoint: replacement_endpoint.clone(), + canonical_path: fixture.env.disk_paths[1].display().to_string(), + physical_device_ids: vec!["replacement-device".to_string()], + filesystem_identity: "replacement-filesystem".to_string(), + }], + ) + .await + .expect("remote survivor intent should persist atomically"); + assert!(ResumeManager::has_replacement_intent(&survivor, &task_id).await); + let reopened = storage + .get_replacement_resume_disk("pool_0_set_0", &task_id, std::slice::from_ref(&replacement_endpoint)) + .await + .expect("remote survivor intent should be discoverable after restart"); + let ReplacementResumeDisk::Existing(reopened) = reopened else { + panic!("persisted remote replacement intent must be reopened"); + }; + assert_eq!(reopened.endpoint().to_string(), remote_survivor.endpoint().to_string()); + let reopened_state = ResumeManager::load_replacement_intent(reopened, &task_id) + .await + .expect("remote replacement intent should replay") + .get_state() + .await; + assert_eq!(reopened_state.replacement_targets, vec![replacement_endpoint]); + + *set.disks.write().await = original_disks; + remote_survivor.close().await.expect("close remote survivor"); + let _ = shutdown.send(()); + super::timeout(Duration::from_secs(10), server) + .await + .expect("replacement intent server shuts down") + .expect("replacement intent server task"); + } + async fn stage_target_rpc(fixture: &TargetRpcFixture) -> (super::DiskStore, rustfs_filemeta::FileInfo, Vec) { use crate::storage::storage_api::ecstore_disk::{DiskAPI, ReadOptions}; let set = fixture diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index 11d86cc90..8095fff4a 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -14,8 +14,9 @@ use super::{LocalMutationTarget, NodeService}; use crate::storage::storage_api::rpc_consumer::node_service::{ - BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, FileInfoVersions, ReadMultipleReq, - ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, validate_batch_read_version_item_count, + BatchReadVersionReq, BatchReadVersionResp, ConditionalFileUpdate, DeleteOptions, DiskError, DiskInfoOptions, + FileInfoVersions, ReadMultipleReq, ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, + validate_batch_read_version_item_count, }; use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; use crate::storage::storage_api::{PartTransactionAction, RenameDataResp, SnapshotLeaseToken, verify_tonic_mutation_body_digest}; @@ -23,12 +24,13 @@ use bytes::Bytes; use rustfs_filemeta::FileInfo; use rustfs_io_metrics::internode_metrics::{ INTERNODE_MSGPACK_CODEC_JSON, INTERNODE_MSGPACK_CODEC_MSGPACK, INTERNODE_MSGPACK_DIRECTION_REQUEST, - INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_VERSION, - INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_STAGE_BATCH_READ_VERSION_DISK_READ, - INTERNODE_STAGE_BATCH_READ_VERSION_REQUEST_DECODE, INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_JSON_ENCODE, - INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_MSGPACK_ENCODE, INTERNODE_STAGE_READ_VERSION_DISK_READ, - INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE, INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE, - INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE, INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_STAGE_BATCH_READ_VERSION_DISK_READ, INTERNODE_STAGE_BATCH_READ_VERSION_REQUEST_DECODE, + INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_JSON_ENCODE, INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_MSGPACK_ENCODE, + INTERNODE_STAGE_READ_VERSION_DISK_READ, INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE, + INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE, INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE, + INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, }; use rustfs_protos::proto_gen::node_service::*; use serde::de::DeserializeOwned; @@ -1884,6 +1886,70 @@ impl NodeService { } } + pub(super) async fn handle_compare_and_update_file( + &self, + request: Request, + ) -> Result, Status> { + verify_disk_mutation_digest( + &request, + rustfs_protos::canonical_compare_and_update_file_request_body(request.get_ref()), + "compare_and_update_file", + )?; + let request = request.into_inner(); + let data_len = request + .expected + .as_ref() + .map_or(0, Bytes::len) + .saturating_add(request.replacement.as_ref().map_or(0, Bytes::len)); + let metrics = runtime_sources::current_internode_metrics(); + metrics.record_incoming_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + metrics.record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, + INTERNODE_TRANSPORT_BACKEND_GRPC, + data_len, + ); + if let Some(disk) = self.find_disk(&request.disk).await { + match disk + .compare_and_update_file(&request.volume, &request.path, request.expected, request.replacement) + .await + { + Ok(outcome) => Ok(Response::new(CompareAndUpdateFileResponse { + success: true, + outcome: match outcome { + ConditionalFileUpdate::Updated => CompareAndUpdateFileOutcome::CompareAndUpdateFileUpdated as i32, + ConditionalFileUpdate::Missing => CompareAndUpdateFileOutcome::CompareAndUpdateFileMissing as i32, + ConditionalFileUpdate::Mismatch => CompareAndUpdateFileOutcome::CompareAndUpdateFileMismatch as i32, + }, + error: None, + })), + Err(err) => { + metrics.record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + Ok(Response::new(CompareAndUpdateFileResponse { + success: false, + outcome: CompareAndUpdateFileOutcome::CompareAndUpdateFileUnspecified as i32, + error: Some(err.into()), + })) + } + } + } else { + metrics.record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_COMPARE_AND_UPDATE_FILE, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + Ok(Response::new(CompareAndUpdateFileResponse { + success: false, + outcome: CompareAndUpdateFileOutcome::CompareAndUpdateFileUnspecified as i32, + error: Some(DiskError::other("cannot find disk".to_string()).into()), + })) + } + } + pub(super) async fn handle_read_all(&self, request: Request) -> Result, Status> { debug!("read all"); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 322438ac9..1cf6e98a8 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -260,13 +260,13 @@ pub(crate) mod rpc_consumer { SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, }; pub(crate) use super::super::{ - BatchReadVersionReq, BatchReadVersionResp, CollectMetricsOpts, DeleteOptions, DiskError, DiskInfoOptions, DiskStore, - ECStore, Error, FileInfoVersions, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, MetricType, PEER_RESTDRY_RUN, - PEER_RESTSIGNAL, PEER_RESTSUB_SYS, ReadMultipleReq, ReadMultipleResp, ReadOptions, SCANNER_PUBLICATION_LEASE_TTL_MS, - SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt, StoragePeerS3ClientExt, - TierDailyStatsWire, UpdateMetadataOpts, all_local_disk_path, collect_local_metrics, find_local_disk_by_ref, - get_global_transition_state, get_local_server_property, reload_bucket_metadata, reload_transition_tier_config, - remove_bucket_metadata, validate_batch_read_version_item_count, + BatchReadVersionReq, BatchReadVersionResp, CollectMetricsOpts, ConditionalFileUpdate, DeleteOptions, DiskError, + DiskInfoOptions, DiskStore, ECStore, Error, FileInfoVersions, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, MetricType, + PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, ReadMultipleReq, ReadMultipleResp, ReadOptions, + SCANNER_PUBLICATION_LEASE_TTL_MS, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt, + StoragePeerS3ClientExt, TierDailyStatsWire, UpdateMetadataOpts, all_local_disk_path, collect_local_metrics, + find_local_disk_by_ref, get_global_transition_state, get_local_server_property, reload_bucket_metadata, + reload_transition_tier_config, remove_bucket_metadata, validate_batch_read_version_item_count, }; pub(crate) type StorageResult = super::super::Result; @@ -471,10 +471,11 @@ pub(crate) mod ecstore_data_usage { #[allow(unused_imports)] pub(crate) mod ecstore_disk { pub(crate) use rustfs_ecstore::api::disk::{ - BUCKET_META_PREFIX, BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, - DiskInfoOptions, DiskStore, FileInfoVersions, FileReader, FileWriter, OldCurrentSize, PartTransactionAction, - RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, SnapshotLeaseToken, - UpdateMetadataOpts, VolumeInfo, WalkDirOptions, get_object_disk_read_timeout, validate_batch_read_version_item_count, + BUCKET_META_PREFIX, BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, ConditionalFileUpdate, DeleteOptions, + DiskAPI, DiskInfo, DiskInfoOptions, DiskStore, FileInfoVersions, FileReader, FileWriter, OldCurrentSize, + PartTransactionAction, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, + SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, get_object_disk_read_timeout, + validate_batch_read_version_item_count, }; #[cfg(test)] pub(crate) use rustfs_ecstore::api::disk::{DiskOption, new_disk}; @@ -721,6 +722,7 @@ pub(crate) type QuotaError = ecstore_bucket::quota::QuotaError; pub(crate) type RawFileInfo = rustfs_filemeta::RawFileInfo; pub(crate) type BatchReadVersionReq = ecstore_disk::BatchReadVersionReq; pub(crate) type BatchReadVersionResp = ecstore_disk::BatchReadVersionResp; +pub(crate) type ConditionalFileUpdate = ecstore_disk::ConditionalFileUpdate; pub(crate) type ReadMultipleReq = ecstore_disk::ReadMultipleReq; pub(crate) type ReadMultipleResp = ecstore_disk::ReadMultipleResp; pub(crate) type ReadOptions = ecstore_disk::ReadOptions; @@ -1433,6 +1435,13 @@ pub(crate) trait StorageDiskRpcExt { async fn read_parts(&self, bucket: &str, paths: &[String]) -> DiskResult>; async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> DiskResult<()>; async fn write_all(&self, volume: &str, path: &str, data: bytes::Bytes) -> DiskResult<()>; + async fn compare_and_update_file( + &self, + volume: &str, + path: &str, + expected: Option, + replacement: Option, + ) -> DiskResult; async fn read_all(&self, volume: &str, path: &str) -> DiskResult; async fn append_file(&self, volume: &str, path: &str) -> DiskResult; async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> DiskResult; @@ -1619,6 +1628,16 @@ where ecstore_disk::DiskAPI::write_all(self, volume, path, data).await } + async fn compare_and_update_file( + &self, + volume: &str, + path: &str, + expected: Option, + replacement: Option, + ) -> DiskResult { + ecstore_disk::DiskAPI::compare_and_update_file(self, volume, path, expected, replacement).await + } + async fn read_all(&self, volume: &str, path: &str) -> DiskResult { ecstore_disk::DiskAPI::read_all(self, volume, path).await }