gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-13 16:11:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): persist replacement intent on remote survivor (#7912)

Browse Source
This commit is contained in:
GatewayJ
2026-09-15 20:46:30 +08:00
committed by GitHub
parent 3774b75b93
commit c5a327df45
17 changed files with 944 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/reliant/grpc_lock_server.rs
+7
View File
@@ -409,6 +409,13 @@ impl NodeService for MinimalLockNodeService {
Err(Status::unimplemented("lock-only test server"))
}
async fn compare_and_update_file(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::CompareAndUpdateFileRequest>,
) -> Result<Response<rustfs_protos::proto_gen::node_service::CompareAndUpdateFileResponse>, Status> {
Err(Status::unimplemented("lock-only test server"))
}
async fn delete(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::DeleteRequest>,
crates/ecstore/src/cluster/rpc/client.rs
+1 -1
View File
@@ -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.
crates/ecstore/src/cluster/rpc/http_auth.rs
+11 -5
View File
@@ -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
crates/ecstore/src/cluster/rpc/remote_disk.rs
+88 -11
View File
@@ -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<ConditionalFileUpdate> {
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<Bytes>,
replacement: Option<Bytes>,
) -> Result<ConditionalFileUpdate> {
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<Bytes> {
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
crates/ecstore/src/cluster/rpc/runtime_sources.rs
+26 -4
View File
@@ -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);
crates/heal/src/heal/manager.rs
+3 -1
View File
@@ -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?;
}
crates/heal/src/heal/manager/auto_scan.rs
+72 -34
View File
@@ -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::<Vec<_>>();
// 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::<Vec<_>>();
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,
crates/heal/src/heal/manager/unclean_shutdown.rs
+67
View File
@@ -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<String>,
) -> 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<String>,
retry_succeeded: HashSet<String>,
@@ -83,6 +90,11 @@ impl HealManager {
let local_disk_map = local_disk_map_read().await;
local_disk_map.values().flatten().cloned().collect::<Vec<_>>()
};
let local_endpoints = local_disks
.iter()
.map(|disk| disk.endpoint().to_string())
.collect::<HashSet<_>>();
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
));
}
}
crates/heal/src/heal/storage.rs
+95 -51
View File
@@ -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<Vec<DiskStore>> {
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<Vec<DiskStore>> {
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<DiskStore> {
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<DiskStore> {
@@ -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<Vec<DiskStore>> {
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<ReplacementResumeDisk> {
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))
crates/io-metrics/src/internode_metrics.rs
+2
View File
@@ -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");
crates/protos/src/generated/proto_gen/node_service.rs
+102
View File
@@ -134,6 +134,29 @@ pub struct WriteAllResponse {
pub error: ::core::option::Option<Error>,
}
#[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<Error>,
}
#[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<Self> {
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<super::CompareAndUpdateFileRequest>,
) -> std::result::Result<tonic::Response<super::CompareAndUpdateFileResponse>, 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<super::DeleteRequest>,
@@ -3375,6 +3445,10 @@ pub mod node_service_server {
&self,
request: tonic::Request<super::WriteAllRequest>,
) -> std::result::Result<tonic::Response<super::WriteAllResponse>, tonic::Status>;
async fn compare_and_update_file(
&self,
request: tonic::Request<super::CompareAndUpdateFileRequest>,
) -> std::result::Result<tonic::Response<super::CompareAndUpdateFileResponse>, tonic::Status>;
async fn delete(
&self,
request: tonic::Request<super::DeleteRequest>,
@@ -4073,6 +4147,34 @@ pub mod node_service_server {
};
Box::pin(fut)
}
"/node_service.NodeService/CompareAndUpdateFile" => {
#[allow(non_camel_case_types)]
struct CompareAndUpdateFileSvc<T: NodeService>(pub Arc<T>);
impl<T: NodeService> tonic::server::UnaryService<super::CompareAndUpdateFileRequest> for CompareAndUpdateFileSvc<T> {
type Response = super::CompareAndUpdateFileResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
fn call(&mut self, request: tonic::Request<super::CompareAndUpdateFileRequest>) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as NodeService>::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<T: NodeService>(pub Arc<T>);
crates/protos/src/lib.rs
+43 -4
View File
@@ -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<Vec<u8>, 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<Vec<u8>, 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(),
crates/protos/src/node.proto
+22
View File
@@ -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
rustfs/src/server/http.rs
+4 -2
View File
@@ -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<Request<()>, 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(
rustfs/src/storage/rpc/node_service.rs
+297 -14
View File
@@ -1442,6 +1442,13 @@ impl Node for NodeService {
self.handle_write_all(request).await
}
async fn compare_and_update_file(
&self,
request: Request<CompareAndUpdateFileRequest>,
) -> Result<Response<CompareAndUpdateFileResponse>, Status> {
self.handle_compare_and_update_file(request).await
}
async fn delete(&self, request: Request<DeleteRequest>) -> Result<Response<DeleteResponse>, 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<Bytes>,
replacement: Option<Bytes>,
) -> Request<CompareAndUpdateFileRequest> {
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<u8>) {
use crate::storage::storage_api::ecstore_disk::{DiskAPI, ReadOptions};
let set = fixture
rustfs/src/storage/rpc/node_service/disk.rs
+74 -8
View File
@@ -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<CompareAndUpdateFileRequest>,
) -> Result<Response<CompareAndUpdateFileResponse>, 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<ReadAllRequest>) -> Result<Response<ReadAllResponse>, Status> {
debug!("read all");
rustfs/src/storage/storage_api.rs
+30 -11
View File
@@ -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<T> = super::super::Result<T>;
@@ -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<Vec<ObjectPartInfo>>;
async fn walk_dir<W: tokio::io::AsyncWrite + Unpin + Send>(&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<bytes::Bytes>,
replacement: Option<bytes::Bytes>,
) -> DiskResult<ConditionalFileUpdate>;
async fn read_all(&self, volume: &str, path: &str) -> DiskResult<bytes::Bytes>;
async fn append_file(&self, volume: &str, path: &str) -> DiskResult<FileWriter>;
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> DiskResult<FileWriter>;
@@ -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<bytes::Bytes>,
replacement: Option<bytes::Bytes>,
) -> DiskResult<ConditionalFileUpdate> {
ecstore_disk::DiskAPI::compare_and_update_file(self, volume, path, expected, replacement).await
}
async fn read_all(&self, volume: &str, path: &str) -> DiskResult<bytes::Bytes> {
ecstore_disk::DiskAPI::read_all(self, volume, path).await
}