fix: make ecstore trash cleanup idempotent (#3205)

* fix(http): reduce header timeout log noise

Log peer_addr for HTTP connection errors so idle or probe traffic can be traced back to the source.

Downgrade HeaderTimeout connection logs from warn to info because these timeouts are often caused by incomplete probe traffic rather than a service-side fault.

* fix(ecstore): suppress spurious warn logs for missing trash cleanup sources

Extract `reliable_rename_inner` with a `warn_on_failure` flag so that
`rename_all_ignore_missing_source` can skip the warn log when the source
is absent. This completes the idempotent cleanup fix — the previous
implementation still emitted a `reliable_rename failed` warning before
the NotFound was caught and silenced.

Also restore the `recursive` branching in `move_to_trash` so that
non-recursive deletions use `fs::rename` directly (preserving original
semantics), while recursive deletions use the idempotent helper.

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)?,

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<Path>,
+    dst_file_path: impl AsRef<Path>,
+    base_dir: impl AsRef<Path>,
+) -> 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<Path>,
     dst_file_path: impl AsRef<Path>,
     base_dir: impl AsRef<Path>,
+) -> 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<Path>,
+    dst_file_path: impl AsRef<Path>,
+    base_dir: impl AsRef<Path>,
+    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());
+    }
 }

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::<hyper::Error>() {
         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>() {
-        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);
     }
 }