mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-09 22:59:59 +00:00
fix(ecstore): harden issue3031 multipart validation path (#3106)
* fix(ecstore): harden issue3031 multipart validation path - clear stale multipart part destinations before rename fan-out - add repeated part overwrite regression coverage - reduce remote disk startup false-fault escalation to suspect-first - refine remote locker diagnostics and lower scanner leader-lock log noise - add a dedicated 4-node issue3031 docker validation script * refactor(admin): inline console version json macro - drop the unused serde_json::json import in admin console - call serde_json::json! inline in version_handler - keep the console version response behavior unchanged * fix(remote-disk): recover suspect health on probe success - record probe success during remote disk health checks so suspect drives recover - use async_with_vars for the remote disk health probe test - make the missing-listener test assert the state transition more robustly
This commit is contained in:
houseme
@@ -3187,6 +3187,76 @@ mod tests {
|
||||
assert!(!parts.user_defined.contains_key(RUSTFS_MULTIPART_OBJECT_KEY));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn repeated_upload_part_overwrites_previous_part_state() {
|
||||
let (_paths, ecstore) = setup_test_env().await;
|
||||
let bucket = format!("multipart-overwrite-{}", Uuid::new_v4().simple());
|
||||
let object = "overwrite/object.txt";
|
||||
create_test_bucket(&ecstore, &bucket).await;
|
||||
|
||||
let upload = ecstore
|
||||
.new_multipart_upload(&bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("multipart upload should be created");
|
||||
|
||||
let mut first = PutObjReader::from_vec(vec![1, 2, 3]);
|
||||
let first_part = ecstore
|
||||
.put_object_part(&bucket, object, &upload.upload_id, 1, &mut first, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("first multipart part should be uploaded");
|
||||
|
||||
let mut second = PutObjReader::from_vec(vec![4, 5, 6, 7]);
|
||||
let second_part = ecstore
|
||||
.put_object_part(&bucket, object, &upload.upload_id, 1, &mut second, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("second multipart part should overwrite the previous part");
|
||||
|
||||
assert_ne!(
|
||||
first_part.etag, second_part.etag,
|
||||
"the overwrite path should persist the latest part metadata rather than reusing stale state"
|
||||
);
|
||||
|
||||
let parts = ecstore
|
||||
.list_object_parts(
|
||||
&bucket,
|
||||
object,
|
||||
&upload.upload_id,
|
||||
None,
|
||||
crate::set_disk::MAX_PARTS_COUNT,
|
||||
&ObjectOptions::default(),
|
||||
)
|
||||
.await
|
||||
.expect("multipart parts should be readable after overwrite");
|
||||
|
||||
assert_eq!(parts.parts.len(), 1, "only the latest version of part 1 should remain visible");
|
||||
assert_eq!(parts.parts[0].part_num, 1);
|
||||
assert_eq!(parts.parts[0].etag, second_part.etag);
|
||||
assert_eq!(parts.parts[0].size, second_part.size);
|
||||
assert_eq!(parts.parts[0].actual_size, second_part.actual_size);
|
||||
|
||||
let completed = ecstore
|
||||
.complete_multipart_upload(
|
||||
&bucket,
|
||||
object,
|
||||
&upload.upload_id,
|
||||
vec![crate::store_api::CompletePart {
|
||||
part_num: 1,
|
||||
etag: second_part.etag.clone(),
|
||||
checksum_crc32: None,
|
||||
checksum_crc32c: None,
|
||||
checksum_sha1: None,
|
||||
checksum_sha256: None,
|
||||
checksum_crc64nvme: None,
|
||||
}],
|
||||
&ObjectOptions::default(),
|
||||
)
|
||||
.await
|
||||
.expect("complete multipart upload should succeed with the latest overwritten part");
|
||||
|
||||
assert_eq!(completed.size, second_part.size as i64);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn cleanup_removes_empty_multipart_sha_dirs() {
|
||||
|
||||
@@ -175,6 +175,10 @@ impl RemoteDisk {
|
||||
});
|
||||
}
|
||||
|
||||
fn mark_suspect_or_offline(&self, reason: &'static str) -> bool {
|
||||
self.health.mark_failure(&self.endpoint, reason)
|
||||
}
|
||||
|
||||
/// Enable health monitoring after disk creation.
|
||||
/// Used to defer health checks until after startup format loading completes,
|
||||
/// so that remote peers have time to come online.
|
||||
@@ -202,7 +206,10 @@ impl RemoteDisk {
|
||||
let mut interval = time::interval(get_drive_active_check_interval());
|
||||
|
||||
// Perform basic connectivity check
|
||||
if Self::perform_connectivity_check(&addr).await.is_err() && health.mark_offline(&endpoint, "connectivity_probe_failed") {
|
||||
let initial_probe_ok = Self::perform_connectivity_check(&addr).await.is_ok();
|
||||
if initial_probe_ok {
|
||||
health.record_operation_success(&endpoint, "connectivity_probe_success");
|
||||
} else if health.mark_failure(&endpoint, "connectivity_probe_failed") {
|
||||
warn!("Remote disk health check failed for {}: marking as faulty", addr);
|
||||
|
||||
// Start recovery monitoring
|
||||
@@ -245,7 +252,9 @@ impl RemoteDisk {
|
||||
}
|
||||
|
||||
// Perform basic connectivity check
|
||||
if Self::perform_connectivity_check(&addr).await.is_err() && health.mark_offline(&endpoint, "connectivity_probe_failed") {
|
||||
if Self::perform_connectivity_check(&addr).await.is_ok() {
|
||||
health.record_operation_success(&endpoint, "connectivity_probe_success");
|
||||
} else if health.mark_failure(&endpoint, "connectivity_probe_failed") {
|
||||
warn!("Remote disk health check failed for {}: marking as faulty", addr);
|
||||
|
||||
// Start recovery monitoring
|
||||
@@ -286,7 +295,7 @@ impl RemoteDisk {
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
health.mark_offline(&endpoint, "connectivity_probe_failed");
|
||||
health.mark_failure(&endpoint, "connectivity_probe_failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -440,7 +449,11 @@ impl RemoteDisk {
|
||||
}
|
||||
|
||||
async fn mark_faulty_and_evict(&self, reason: &'static str) {
|
||||
if self.health.mark_offline(&self.endpoint, reason) {
|
||||
let previous_state = self.runtime_state();
|
||||
let became_offline = self.mark_suspect_or_offline(reason);
|
||||
let state = self.runtime_state();
|
||||
|
||||
if state != previous_state {
|
||||
self.spawn_recovery_monitor_if_needed();
|
||||
counter!(
|
||||
"rustfs_drive_faulty_mark_total",
|
||||
@@ -448,10 +461,17 @@ impl RemoteDisk {
|
||||
"reason" => reason.to_string()
|
||||
)
|
||||
.increment(1);
|
||||
warn!(
|
||||
"Remote disk marked faulty after timeout: endpoint={}, addr={}, reason={}",
|
||||
self.endpoint, self.addr, reason
|
||||
);
|
||||
if became_offline || state == RuntimeDriveHealthState::Offline {
|
||||
warn!(
|
||||
"Remote disk marked faulty after timeout: endpoint={}, addr={}, reason={}",
|
||||
self.endpoint, self.addr, reason
|
||||
);
|
||||
} else {
|
||||
warn!(
|
||||
"Remote disk marked suspect after timeout: endpoint={}, addr={}, reason={}, state={:?}",
|
||||
self.endpoint, self.addr, reason, state
|
||||
);
|
||||
}
|
||||
counter!(
|
||||
"rustfs_drive_connection_evict_total",
|
||||
"endpoint" => self.endpoint.to_string(),
|
||||
@@ -1951,20 +1971,39 @@ mod tests {
|
||||
disk_idx: 0,
|
||||
};
|
||||
|
||||
let disk_option = DiskOption {
|
||||
cleanup: false,
|
||||
health_check: true,
|
||||
};
|
||||
temp_env::async_with_vars(
|
||||
[
|
||||
(rustfs_config::ENV_DRIVE_ACTIVE_CHECK_INTERVAL_SECS, Some("1")),
|
||||
(rustfs_config::ENV_DRIVE_ACTIVE_CHECK_TIMEOUT_SECS, Some("1")),
|
||||
],
|
||||
async {
|
||||
let disk_option = DiskOption {
|
||||
cleanup: false,
|
||||
health_check: true,
|
||||
};
|
||||
|
||||
let remote_disk = RemoteDisk::new(&endpoint, &disk_option, Arc::new(TcpHttpInternodeDataTransport))
|
||||
.await
|
||||
.unwrap();
|
||||
remote_disk.enable_health_check();
|
||||
let remote_disk = RemoteDisk::new(&endpoint, &disk_option, Arc::new(TcpHttpInternodeDataTransport))
|
||||
.await
|
||||
.unwrap();
|
||||
remote_disk.enable_health_check();
|
||||
|
||||
// wait for health check connect timeout
|
||||
tokio::time::sleep(Duration::from_secs(6)).await;
|
||||
|
||||
assert!(!remote_disk.is_online().await);
|
||||
// Wait out the initial success-grace window so the active probe loop
|
||||
// actually attempts a connectivity check. Under the new
|
||||
// suspect-first semantics we only need to prove that the drive
|
||||
// transitions away from a clean Online state at least once.
|
||||
tokio::time::sleep(SKIP_IF_SUCCESS_BEFORE + Duration::from_secs(2)).await;
|
||||
assert!(
|
||||
remote_disk.offline_duration_secs().is_some(),
|
||||
"missing listener should transition the drive through suspect/offline tracking"
|
||||
);
|
||||
assert_ne!(
|
||||
remote_disk.runtime_state(),
|
||||
RuntimeDriveHealthState::Online,
|
||||
"missing listener should not remain in a clean Online state after probing"
|
||||
);
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -2284,7 +2323,12 @@ mod tests {
|
||||
.expect_err("timeout should fail");
|
||||
|
||||
assert!(err.to_string().contains("timeout"));
|
||||
assert!(!remote_disk.is_online().await, "remote disk should be marked faulty after timeout");
|
||||
assert!(remote_disk.is_online().await, "first timeout should keep the remote disk online");
|
||||
assert_eq!(
|
||||
remote_disk.runtime_state(),
|
||||
RuntimeDriveHealthState::Suspect,
|
||||
"first timeout should move the remote disk into suspect state"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -2450,7 +2494,15 @@ mod tests {
|
||||
},
|
||||
std::io::ErrorKind::TimedOut
|
||||
);
|
||||
assert!(!remote_disk.is_online().await, "timeout-like errors should mark remote disk faulty");
|
||||
assert!(
|
||||
remote_disk.is_online().await,
|
||||
"first timeout-like error should keep the remote disk online"
|
||||
);
|
||||
assert_eq!(
|
||||
remote_disk.runtime_state(),
|
||||
RuntimeDriveHealthState::Suspect,
|
||||
"first timeout-like error should move the remote disk into suspect state"
|
||||
);
|
||||
assert!(
|
||||
!GLOBAL_CONN_MAP.read().await.contains_key(&addr),
|
||||
"timeout-like errors should evict cached connection"
|
||||
@@ -2503,7 +2555,15 @@ mod tests {
|
||||
},
|
||||
std::io::ErrorKind::ConnectionRefused
|
||||
);
|
||||
assert!(!remote_disk.is_online().await, "network-like errors should mark remote disk faulty");
|
||||
assert!(
|
||||
remote_disk.is_online().await,
|
||||
"first network-like error should keep the remote disk online"
|
||||
);
|
||||
assert_eq!(
|
||||
remote_disk.runtime_state(),
|
||||
RuntimeDriveHealthState::Suspect,
|
||||
"first network-like error should move the remote disk into suspect state"
|
||||
);
|
||||
assert!(
|
||||
!GLOBAL_CONN_MAP.read().await.contains_key(&addr),
|
||||
"network-like errors should evict cached connection"
|
||||
|
||||
@@ -25,7 +25,7 @@ use tokio::time::timeout;
|
||||
use tonic::Request;
|
||||
use tonic::service::interceptor::InterceptedService;
|
||||
use tonic::transport::Channel;
|
||||
use tracing::{info, warn};
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
/// Remote lock client implementation
|
||||
#[derive(Debug, Clone)]
|
||||
@@ -64,16 +64,44 @@ impl RemoteClient {
|
||||
.map_err(|err| LockError::internal(format!("can not get client, err: {err}")))
|
||||
}
|
||||
|
||||
async fn evict_connection(&self, op: &'static str, reason: &str) {
|
||||
warn!(
|
||||
addr = %self.addr,
|
||||
op,
|
||||
reason,
|
||||
"Evicting cached remote lock connection after RPC failure"
|
||||
);
|
||||
fn is_scanner_leader_lock(resource_summary: &str) -> bool {
|
||||
resource_summary == ".rustfs.sys/leader.lock@latest"
|
||||
}
|
||||
|
||||
async fn evict_connection(&self, op: &'static str, reason: &str, resource_summary: &str) {
|
||||
if Self::is_scanner_leader_lock(resource_summary) {
|
||||
debug!(
|
||||
addr = %self.addr,
|
||||
op,
|
||||
reason,
|
||||
resource_summary,
|
||||
"Evicting cached remote lock connection for scanner leader-lock RPC failure"
|
||||
);
|
||||
} else {
|
||||
warn!(
|
||||
addr = %self.addr,
|
||||
op,
|
||||
reason,
|
||||
resource_summary,
|
||||
"Evicting cached remote lock connection after RPC failure"
|
||||
);
|
||||
}
|
||||
evict_failed_connection(&self.addr).await;
|
||||
}
|
||||
|
||||
fn summarize_resources(requests: &[LockRequest]) -> String {
|
||||
const LIMIT: usize = 3;
|
||||
let mut resources = requests
|
||||
.iter()
|
||||
.take(LIMIT)
|
||||
.map(|request| request.resource.to_string())
|
||||
.collect::<Vec<_>>();
|
||||
if requests.len() > LIMIT {
|
||||
resources.push(format!("... (+{} more)", requests.len() - LIMIT));
|
||||
}
|
||||
resources.join(", ")
|
||||
}
|
||||
|
||||
fn rpc_timeout(timeout_duration: Duration) -> Duration {
|
||||
if timeout_duration.is_zero() {
|
||||
Duration::from_millis(1)
|
||||
@@ -86,6 +114,7 @@ impl RemoteClient {
|
||||
&self,
|
||||
op: &'static str,
|
||||
timeout_duration: Duration,
|
||||
resource_summary: &str,
|
||||
future: F,
|
||||
) -> std::result::Result<T, LockError>
|
||||
where
|
||||
@@ -96,12 +125,50 @@ impl RemoteClient {
|
||||
Ok(Ok(response)) => Ok(response),
|
||||
Ok(Err(err)) => {
|
||||
let reason = err.to_string();
|
||||
self.evict_connection(op, &reason).await;
|
||||
if Self::is_scanner_leader_lock(resource_summary) {
|
||||
debug!(
|
||||
addr = %self.addr,
|
||||
op,
|
||||
timeout_ms = lock_timeout.as_millis(),
|
||||
resource_summary,
|
||||
tonic_code = ?err.code(),
|
||||
tonic_message = err.message(),
|
||||
"Remote lock RPC returned tonic error for scanner leader lock"
|
||||
);
|
||||
} else {
|
||||
warn!(
|
||||
addr = %self.addr,
|
||||
op,
|
||||
timeout_ms = lock_timeout.as_millis(),
|
||||
resource_summary,
|
||||
tonic_code = ?err.code(),
|
||||
tonic_message = err.message(),
|
||||
"Remote lock RPC returned tonic error"
|
||||
);
|
||||
}
|
||||
self.evict_connection(op, &reason, resource_summary).await;
|
||||
Err(LockError::internal(format!("{op} RPC failed: {reason}")))
|
||||
}
|
||||
Err(_) => {
|
||||
let reason = format!("RPC timed out after {:?}", lock_timeout);
|
||||
self.evict_connection(op, &reason).await;
|
||||
if Self::is_scanner_leader_lock(resource_summary) {
|
||||
debug!(
|
||||
addr = %self.addr,
|
||||
op,
|
||||
timeout_ms = lock_timeout.as_millis(),
|
||||
resource_summary,
|
||||
"Remote lock RPC timed out for scanner leader lock"
|
||||
);
|
||||
} else {
|
||||
warn!(
|
||||
addr = %self.addr,
|
||||
op,
|
||||
timeout_ms = lock_timeout.as_millis(),
|
||||
resource_summary,
|
||||
"Remote lock RPC timed out"
|
||||
);
|
||||
}
|
||||
self.evict_connection(op, &reason, resource_summary).await;
|
||||
Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), lock_timeout))
|
||||
}
|
||||
}
|
||||
@@ -182,12 +249,16 @@ impl LockClient for RemoteClient {
|
||||
async fn acquire_lock(&self, request: &LockRequest) -> Result<LockResponse> {
|
||||
info!("remote acquire_exclusive for {}", request.resource);
|
||||
let mut client = self.get_client().await?;
|
||||
let resource_summary = request.resource.to_string();
|
||||
let req = Request::new(GenerallyLockRequest {
|
||||
args: serde_json::to_string(&request)
|
||||
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
|
||||
});
|
||||
|
||||
let resp = match self.execute_rpc("lock", request.acquire_timeout, client.lock(req)).await {
|
||||
let resp = match self
|
||||
.execute_rpc("lock", request.acquire_timeout, &resource_summary, client.lock(req))
|
||||
.await
|
||||
{
|
||||
Ok(resp) => resp.into_inner(),
|
||||
Err(err @ LockError::Timeout { .. }) => return Ok(Self::rpc_timeout_failure_response(request, &err)),
|
||||
Err(err) => return Ok(Self::rpc_failure_response(request, &err)),
|
||||
@@ -215,6 +286,7 @@ impl LockClient for RemoteClient {
|
||||
}
|
||||
|
||||
let mut client = self.get_client().await?;
|
||||
let resource_summary = Self::summarize_resources(requests);
|
||||
let req = Request::new(BatchGenerallyLockRequest {
|
||||
args: requests
|
||||
.iter()
|
||||
@@ -225,7 +297,7 @@ impl LockClient for RemoteClient {
|
||||
});
|
||||
|
||||
let resp = match self
|
||||
.execute_rpc("lock_batch", Self::batch_rpc_timeout(requests), client.lock_batch(req))
|
||||
.execute_rpc("lock_batch", Self::batch_rpc_timeout(requests), &resource_summary, client.lock_batch(req))
|
||||
.await
|
||||
{
|
||||
Ok(resp) => resp.into_inner(),
|
||||
|
||||
@@ -262,6 +262,11 @@ impl SetDisks {
|
||||
let dst_bucket = Arc::new(dst_bucket.to_string());
|
||||
let dst_object = Arc::new(dst_object.to_string());
|
||||
|
||||
// Match MinIO's multipart overwrite semantics: clear any stale destination
|
||||
// part payload and metadata before the new per-disk rename fan-out begins.
|
||||
self.cleanup_multipart_path(&[dst_object.to_string(), format!("{dst_object}.meta")])
|
||||
.await;
|
||||
|
||||
let mut errs = Vec::with_capacity(disks.len());
|
||||
|
||||
let futures = disks.iter().map(|disk| {
|
||||
|
||||
Reference in New Issue
Block a user