chore: integrate main for error identity fix

crates/ecstore/src/disk/error.rs

@@ -627,7 +627,13 @@ impl From<tokio::task::JoinError> for DiskError {
 impl Clone for DiskError {
     fn clone(&self) -> Self {
         match self {
-            DiskError::Io(io_error) => DiskError::Io(std::io::Error::new(io_error.kind(), io_error.to_string())),
+            DiskError::Io(io_error) => DiskError::Io(
+                rustfs_rio::clone_internode_http_io_error(io_error)
+                    .and_then(std::io::Error::into_inner)
+                    // The helper derives a kind from the source; Clone must retain the original outer kind.
+                    .map(|source| std::io::Error::new(io_error.kind(), source))
+                    .unwrap_or_else(|| std::io::Error::new(io_error.kind(), io_error.to_string())),
+            ),
             DiskError::MaxVersionsExceeded => DiskError::MaxVersionsExceeded,
             DiskError::Unexpected => DiskError::Unexpected,
             DiskError::CorruptedFormat => DiskError::CorruptedFormat,
@@ -1265,6 +1271,49 @@ mod tests {
         assert!(!bad_request.is_retryable_internode_write_failure());
     }
 
+    #[test]
+    fn test_internode_http_clone_preserves_retryability_status_and_context() {
+        use http::StatusCode;
+        use rustfs_rio::InternodeHttpErrorKind::{ConnectionRefused, ConnectionReset, HttpStatus, Unknown};
+
+        for (kind, retryable) in [
+            (ConnectionRefused, true),
+            (ConnectionReset, true),
+            (HttpStatus(StatusCode::TOO_MANY_REQUESTS), true),
+            (HttpStatus(StatusCode::SERVICE_UNAVAILABLE), true),
+            (HttpStatus(StatusCode::CONFLICT), true),
+            (Unknown, false),
+            (HttpStatus(StatusCode::BAD_REQUEST), false),
+            (HttpStatus(StatusCode::INTERNAL_SERVER_ERROR), false),
+        ] {
+            let original = DiskError::from(rustfs_rio::new_test_internode_http_io_error(kind));
+            assert_eq!(original.internode_http_error_kind(), Some(kind));
+            assert_eq!(original.is_retryable_internode_write_failure(), retryable);
+
+            let cloned = original.clone();
+            assert_eq!(cloned, original, "clone must preserve the error bucket for {kind:?}");
+            assert_eq!(
+                cloned.is_retryable_internode_write_failure(),
+                retryable,
+                "clone changed retryability for {kind:?}"
+            );
+            assert_eq!(cloned.internode_http_error_kind(), Some(kind));
+            if let HttpStatus(status) = kind {
+                assert!(cloned.is_internode_http_status(status.as_u16()));
+            }
+            let DiskError::Io(io_error) = &cloned else {
+                panic!("unmarked internode error must remain Io: {cloned:?}");
+            };
+            let source = io_error
+                .get_ref()
+                .and_then(|source| source.downcast_ref::<InternodeHttpError>())
+                .expect("clone must retain the structured internode error");
+            assert_eq!(source.context().method(), "PUT");
+            assert_eq!(source.context().target(), "/rustfs/rpc/put_file_stream");
+            assert_eq!(source.context().operation(), Some(INTERNODE_OPERATION_PUT_FILE_STREAM));
+        }
+    }
+
     #[tokio::test]
     async fn read_stream_conflict_is_not_a_retryable_put_file_failure() {
         use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -1309,11 +1358,57 @@ mod tests {
                 !error.is_retryable_internode_write_failure(),
                 "read-operation 409 must not trigger put-file retry"
             );
+            let cloned = error.clone();
+            let reduced = crate::disk::error_reduce::reduce_write_quorum_errs(&[Some(error)], &[], 1)
+                .expect("the read conflict must remain the dominant error");
+            for preserved in [&cloned, &reduced] {
+                assert!(
+                    !preserved.is_retryable_internode_write_failure(),
+                    "cloning or reducing a read conflict must not turn it into a PUT retry"
+                );
+                assert!(preserved.is_internode_http_status(409));
+                let DiskError::Io(io_error) = preserved else {
+                    panic!("read conflict must remain Io: {preserved:?}");
+                };
+                let source = io_error
+                    .get_ref()
+                    .and_then(|source| source.downcast_ref::<InternodeHttpError>())
+                    .expect("read conflict must retain its request context");
+                assert_eq!(source.context().method(), "GET");
+                assert_eq!(source.context().target(), "/rustfs/rpc/read_file_stream");
+                assert_eq!(
+                    source.context().operation(),
+                    Some(rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_READ_FILE_STREAM)
+                );
+            }
         })
         .await
         .expect("isolated read-conflict test must finish within its budget");
     }
 
+    #[test]
+    fn test_internode_http_clone_preserves_outer_io_kind_and_message() {
+        let source = rustfs_rio::new_test_internode_http_io_error(InternodeHttpErrorKind::ConnectionReset)
+            .into_inner()
+            .expect("the internode helper must provide a typed source");
+        let original_io = io::Error::new(io::ErrorKind::InvalidData, source);
+        let message = original_io.to_string();
+        let original = DiskError::from(original_io);
+        assert_eq!(original.internode_http_error_kind(), Some(InternodeHttpErrorKind::ConnectionReset));
+        assert!(original.is_retryable_internode_write_failure());
+
+        let cloned = original.clone();
+        let reduced = crate::disk::error_reduce::reduce_write_quorum_errs(&[Some(original)], &[], 1)
+            .expect("the wrapped internode error must remain the dominant error");
+        for preserved in [&cloned, &reduced] {
+            let DiskError::Io(io_error) = preserved else {
+                panic!("the wrapped error must remain Io: {preserved:?}");
+            };
+            assert_eq!(io_error.kind(), io::ErrorKind::InvalidData);
+            assert_eq!(io_error.to_string(), message);
+        }
+    }
+
     #[test]
     fn test_internode_missing_errors_preserve_disk_error_types() {
         let file_missing = DiskError::from(rustfs_rio::new_test_remote_file_not_found_http_io_error());
@@ -1325,6 +1420,17 @@ mod tests {
         assert_eq!(file_missing, DiskError::FileNotFound);
         assert_eq!(volume_missing, DiskError::VolumeNotFound);
         assert!(matches!(unmarked_server_error, DiskError::Io(_)));
+        for missing in [file_missing, volume_missing] {
+            assert_eq!(missing.clone(), missing);
+            assert_eq!(
+                crate::disk::error_reduce::reduce_write_quorum_errs(
+                    &[Some(missing.clone()), Some(missing.clone()), None],
+                    &[],
+                    2
+                ),
+                Some(missing)
+            );
+        }
     }
 
     #[test]

crates/ecstore/src/disk/error_reduce.rs

@@ -226,6 +226,78 @@ mod tests {
         assert_eq!(res, Some(quorum_err));
     }
 
+    #[test]
+    fn test_write_quorum_reduction_preserves_internode_http_identity() {
+        use http::StatusCode;
+        use rustfs_rio::InternodeHttpErrorKind::{ConnectionRefused, HttpStatus, Unknown};
+
+        for (kind, retryable) in [
+            (ConnectionRefused, true),
+            (HttpStatus(StatusCode::SERVICE_UNAVAILABLE), true),
+            (HttpStatus(StatusCode::CONFLICT), true),
+            (Unknown, false),
+            (HttpStatus(StatusCode::BAD_REQUEST), false),
+        ] {
+            // Construct both producer errors independently: the reducer owns the first clone.
+            let first = Error::from(rustfs_rio::new_test_internode_http_io_error(kind));
+            let second = Error::from(rustfs_rio::new_test_internode_http_io_error(kind));
+            assert_eq!(first.internode_http_error_kind(), Some(kind));
+            assert_eq!(second.internode_http_error_kind(), Some(kind));
+            assert_eq!(first.is_retryable_internode_write_failure(), retryable);
+            let errors = [Some(first), Some(second), None];
+            let reduced = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, 2)
+                .expect("two equal producer errors must dominate one successful write");
+
+            assert_eq!(Some(&reduced), errors[0].as_ref());
+            assert_eq!(
+                reduced.is_retryable_internode_write_failure(),
+                retryable,
+                "quorum reduction changed retryability for {kind:?}"
+            );
+            assert_eq!(reduced.internode_http_error_kind(), Some(kind));
+            if let HttpStatus(status) = kind {
+                assert!(reduced.is_internode_http_status(status.as_u16()));
+            }
+            let Error::Io(io_error) = &reduced else {
+                panic!("the dominant error must remain Io: {reduced:?}");
+            };
+            let source = io_error
+                .get_ref()
+                .and_then(|source| source.downcast_ref::<rustfs_rio::InternodeHttpError>())
+                .expect("quorum reduction must retain the structured internode error");
+            assert_eq!(source.context().method(), "PUT");
+            assert_eq!(source.context().target(), "/rustfs/rpc/put_file_stream");
+            assert_eq!(
+                source.context().operation(),
+                Some(rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_PUT_FILE_STREAM)
+            );
+        }
+    }
+
+    #[test]
+    fn test_clone_and_write_quorum_do_not_promote_non_retryable_errors() {
+        use http::StatusCode;
+        use rustfs_rio::InternodeHttpErrorKind::{HttpStatus, Unknown};
+
+        for original in [
+            Error::from(rustfs_rio::new_test_internode_http_io_error(Unknown)),
+            Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::BAD_REQUEST))),
+            Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::FORBIDDEN))),
+            Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::NOT_FOUND))),
+            Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(
+                StatusCode::INTERNAL_SERVER_ERROR,
+            ))),
+            err_io("internode connection reset: PUT /rustfs/rpc/put_file_stream"),
+        ] {
+            assert!(!original.is_retryable_internode_write_failure());
+            let cloned = original.clone();
+            let reduced =
+                reduce_write_quorum_errs(&[Some(original)], &[], 1).expect("a non-retryable error must remain an error");
+            assert!(!cloned.is_retryable_internode_write_failure());
+            assert!(!reduced.is_retryable_internode_write_failure());
+        }
+    }
+
     #[test]
     fn test_count_errs() {
         let e1 = err_io("a");

crates/targets/src/store.rs

@@ -919,6 +919,22 @@ where
             }
         }
 
+        // The filename's item count is untrusted. Reject a payload that contains
+        // more items than advertised instead of returning success and allowing the
+        // caller to delete the entry with trailing events still in the file.
+        match deserializer.next() {
+            None => {}
+            Some(Ok(_)) => {
+                return Err(StoreError::Deserialization(format!(
+                    "Batch for key {key} contains more than {} items",
+                    key.item_count
+                )));
+            }
+            Some(Err(e)) => {
+                return Err(StoreError::Deserialization(format!("Failed to deserialize trailing batch item: {e}")));
+            }
+        }
+
         if items.is_empty() && key.item_count > 0 {
             return Err(StoreError::Deserialization("No items found".to_string()));
         }
@@ -1381,6 +1397,39 @@ mod tests {
         let _ = store.delete();
     }
 
+    #[test]
+    fn get_multiple_errors_on_batch_with_trailing_items_instead_of_partial_success() {
+        let dir = temp_store_dir("trailing-batch-items");
+        let store = QueueStore::<String>::new_with_compression(&dir, 8, ".test", false);
+        store.open().unwrap();
+
+        let items = vec!["aa".to_string(), "bb".to_string(), "cc".to_string()];
+        let original_key = store.put_multiple(items).unwrap();
+        assert_eq!(original_key.item_count, 3);
+
+        // Keep the three-item payload but make its filename claim that it contains
+        // only two items, simulating a corrupt or otherwise untrusted queue key.
+        let original_path = store.file_path(&original_key);
+        let advertised_key = Key {
+            item_count: 2,
+            ..original_key
+        };
+        let advertised_path = store.file_path(&advertised_key);
+        std::fs::rename(&original_path, &advertised_path).unwrap();
+
+        let err = store.get_multiple(&advertised_key).unwrap_err();
+        assert!(
+            matches!(err, StoreError::Deserialization(_)),
+            "expected Deserialization error, got {err:?}"
+        );
+
+        // Because get_multiple failed, the batch entry remains available for
+        // inspection or recovery instead of being silently discarded.
+        assert!(advertised_path.exists());
+
+        let _ = store.delete();
+    }
+
     #[test]
     fn concurrent_put_raw_respects_entry_limit() {
         let dir = temp_store_dir("concurrent-limit");