diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 89ddc5df1..92de5493c 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -26,7 +26,7 @@ use crate::disk::{ format::FormatV3, fs::{O_APPEND, O_CREATE, O_RDONLY, O_TRUNC, O_WRONLY, access, lstat, lstat_std, remove, remove_all_std, remove_std, rename}, os, - os::{check_path_length, is_empty_dir, is_root_disk, rename_all}, + os::{check_path_length, is_empty_dir, is_root_disk, rename_all, rename_all_ignore_missing_source}, }; use crate::erasure_coding::bitrot_verify; use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold}; @@ -929,19 +929,20 @@ impl LocalDisk { // } let err = if recursive { - rename_all(delete_path, trash_path, self.get_bucket_path(RUSTFS_META_TMP_DELETED_BUCKET)?) + rename_all_ignore_missing_source(delete_path, trash_path, self.get_bucket_path(RUSTFS_META_TMP_DELETED_BUCKET)?) .await .err() } else { - rename(&delete_path, &trash_path) - .await - .map_err(|e| to_file_error(e).into()) - .err() + match rename(&delete_path, &trash_path).await { + Ok(()) => None, + Err(err) if err.kind() == ErrorKind::NotFound => None, + Err(err) => Some(to_file_error(err).into()), + } }; if immediate_purge || delete_path.to_string_lossy().ends_with(SLASH_SEPARATOR) { let trash_path2 = self.get_object_path(RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?; - let _ = rename_all( + let _ = rename_all_ignore_missing_source( encode_dir_object(delete_path.to_string_lossy().as_ref()), trash_path2, self.get_bucket_path(RUSTFS_META_TMP_DELETED_BUCKET)?, diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index b70cdb449..4fffcffa1 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -140,10 +140,32 @@ pub async fn rename_all( Ok(()) } +#[tracing::instrument(level = "debug", skip_all)] +pub async fn rename_all_ignore_missing_source( + src_file_path: impl AsRef, + dst_file_path: impl AsRef, + base_dir: impl AsRef, +) -> Result<()> { + match reliable_rename_inner(src_file_path, dst_file_path.as_ref(), base_dir, false).await { + Ok(()) => Ok(()), + Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()), + Err(err) => Err(to_file_error(err).into()), + } +} + async fn reliable_rename( src_file_path: impl AsRef, dst_file_path: impl AsRef, base_dir: impl AsRef, +) -> io::Result<()> { + reliable_rename_inner(src_file_path, dst_file_path, base_dir, true).await +} + +async fn reliable_rename_inner( + src_file_path: impl AsRef, + dst_file_path: impl AsRef, + base_dir: impl AsRef, + warn_on_failure: bool, ) -> io::Result<()> { if let Some(parent) = dst_file_path.as_ref().parent() && !file_exists(parent) @@ -158,13 +180,15 @@ async fn reliable_rename( i += 1; continue; } - warn!( - "reliable_rename failed. src_file_path: {:?}, dst_file_path: {:?}, base_dir: {:?}, err: {:?}", - src_file_path.as_ref(), - dst_file_path.as_ref(), - base_dir.as_ref(), - e - ); + if warn_on_failure { + warn!( + "reliable_rename failed. src_file_path: {:?}, dst_file_path: {:?}, base_dir: {:?}, err: {:?}", + src_file_path.as_ref(), + dst_file_path.as_ref(), + base_dir.as_ref(), + e + ); + } return Err(e); } @@ -263,4 +287,17 @@ mod tests { assert!(matches!(err, DiskError::FileNotFound)); assert!(!dst.exists()); } + + #[tokio::test] + async fn rename_all_ignore_missing_source_returns_ok() { + let temp_dir = tempdir().expect("create temp dir"); + let src = temp_dir.path().join("missing"); + let dst = temp_dir.path().join("dst"); + + rename_all_ignore_missing_source(&src, &dst, temp_dir.path()) + .await + .expect("missing cleanup source must be ignored"); + + assert!(!dst.exists()); + } } diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 0e92451b4..1ce05a4f6 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -884,7 +884,7 @@ fn process_connection( let stream = TokioIo::new(tls_socket); let conn = http_server.serve_connection(stream, hybrid_service); if let Err(err) = graceful.watch(conn).await { - handle_connection_error(&*err); + handle_connection_error(Some(peer_addr.as_str()), &*err); } } Err(err) => { @@ -921,10 +921,11 @@ fn process_connection( debug!("TLS handshake success"); } else { debug!("Http handshake start"); + let peer_addr = socket.peer_addr().ok().map(|addr| addr.to_string()); let stream = TokioIo::new(socket); let conn = http_server.serve_connection(stream, hybrid_service); if let Err(err) = graceful.watch(conn).await { - handle_connection_error(&*err); + handle_connection_error(peer_addr.as_deref(), &*err); } debug!("Http handshake success"); }; @@ -932,34 +933,46 @@ fn process_connection( } /// Handles connection errors by logging them with appropriate severity -fn handle_connection_error(err: &(dyn std::error::Error + 'static)) { +fn handle_connection_error(peer_addr: Option<&str>, err: &(dyn std::error::Error + 'static)) { let s = err.to_string(); if s.contains("connection reset") || s.contains("broken pipe") { - warn!("The connection was reset by the peer or broken pipe: {}", s); + warn!( + peer_addr = %peer_addr.unwrap_or("unknown"), + "The connection was reset by the peer or broken pipe: {}", s + ); // Ignore common non-fatal errors return; } if let Some(hyper_err) = err.downcast_ref::() { if hyper_err.is_incomplete_message() { - warn!("The HTTP connection is closed prematurely and the message is not completed:{}", hyper_err); + warn!( + peer_addr = %peer_addr.unwrap_or("unknown"), + "The HTTP connection is closed prematurely and the message is not completed:{}", hyper_err + ); } else if hyper_err.is_closed() { - warn!("The HTTP connection is closed:{}", hyper_err); + warn!(peer_addr = %peer_addr.unwrap_or("unknown"), "The HTTP connection is closed:{}", hyper_err); } else if hyper_err.is_parse() { - error!("HTTP message parsing failed:{}", hyper_err); + error!(peer_addr = %peer_addr.unwrap_or("unknown"), "HTTP message parsing failed:{}", hyper_err); } else if hyper_err.is_user() { - error!("HTTP user-custom error:{}", hyper_err); + error!(peer_addr = %peer_addr.unwrap_or("unknown"), "HTTP user-custom error:{}", hyper_err); } else if hyper_err.is_canceled() { - warn!("The HTTP connection is canceled:{}", hyper_err); + warn!( + peer_addr = %peer_addr.unwrap_or("unknown"), + "The HTTP connection is canceled:{}", hyper_err + ); } else if format!("{:?}", hyper_err).contains("HeaderTimeout") { - warn!("The HTTP connection timed out (HeaderTimeout): {}", hyper_err); + info!( + peer_addr = %peer_addr.unwrap_or("unknown"), + "The HTTP connection timed out while reading request headers (HeaderTimeout): {}", hyper_err + ); } else { - error!("Unknown hyper error:{:?}", hyper_err); + error!(peer_addr = %peer_addr.unwrap_or("unknown"), "Unknown hyper error:{:?}", hyper_err); } } else if let Some(io_err) = err.downcast_ref::() { - error!("Unknown connection IO error:{}", io_err); + error!(peer_addr = %peer_addr.unwrap_or("unknown"), "Unknown connection IO error:{}", io_err); } else { - error!("Unknown connection error type:{:?}", err); + error!(peer_addr = %peer_addr.unwrap_or("unknown"), "Unknown connection error type:{:?}", err); } }