diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 02fe70f29..1498d3ada 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -1041,7 +1041,13 @@ fn should_count_decommission_version_complete(ignore: bool, cleanup_ignored: boo fn is_decommission_copy_cleanup_safe_error(err: &Error) -> bool { // DataMovementOverwriteErr only means source and destination pool resolved to // the same pool. Without a target equivalence check it is not cleanup-safe. - is_err_object_not_found(err) || is_err_version_not_found(err) + if is_err_object_not_found(err) || is_err_version_not_found(err) { + return true; + } + + // A not-found surfacing from inside a data-movement stage is the same + // condition once the wrapper is unwrapped (backlog#1827 T2). + crate::data_movement::data_movement_stage_source(err).is_some_and(is_decommission_copy_cleanup_safe_error) } fn is_decommission_target_capacity_error(err: &Error) -> bool { @@ -1049,6 +1055,13 @@ fn is_decommission_target_capacity_error(err: &Error) -> bool { return true; } + // A stage failure keeps the error it wrapped, so classify by type rather + // than by the rendered message (backlog#1827 T2). The substring fallback + // stays for errors that reached here through some other wrapper. + if let Some(source) = crate::data_movement::data_movement_stage_source(err) { + return is_decommission_target_capacity_error(source); + } + let message = err.to_string(); let disk_full = Error::DiskFull.to_string(); let storage_full = Error::StorageFull.to_string(); @@ -4427,6 +4440,36 @@ mod tests { assert!(is_decommission_target_capacity_error(&Error::StorageFull)); } + /// The decommission loop classifies errors that came back through a + /// data-movement stage wrapper. Before backlog#1827 T2 the wrapper flattened + /// everything into `Error::other(String)`, so these two classifiers had to + /// match on rendered text; now the wrapped error is recoverable by type. + #[test] + fn decommission_classifiers_see_through_a_stage_wrapper() { + let wrap = |inner: Error| { + crate::data_movement::data_movement_stage_error_for_test( + "decommission_object", + "put_object", + "bucket-a", + "object-a", + inner, + ) + }; + + // Capacity: the target pool filling up must still stop the loop. + assert!(is_decommission_target_capacity_error(&wrap(Error::DiskFull))); + assert!(is_decommission_target_capacity_error(&wrap(Error::StorageFull))); + assert!(!is_decommission_target_capacity_error(&wrap(Error::SlowDown))); + + // Cleanup safety: a not-found surfacing from inside a stage is the same + // condition as one surfacing directly, so the source entry stays + // eligible for cleanup. + let not_found = Error::ObjectNotFound("bucket-a".to_string(), "object-a".to_string()); + assert!(is_decommission_copy_cleanup_safe_error(¬_found)); + assert!(is_decommission_copy_cleanup_safe_error(&wrap(not_found))); + assert!(!is_decommission_copy_cleanup_safe_error(&wrap(Error::SlowDown))); + } + #[test] fn decommission_target_capacity_error_accepts_wrapped_capacity_errors() { let disk_full = Error::other(format!("decommission_object: put_object failed for bucket/object: {}", Error::DiskFull)); diff --git a/crates/ecstore/src/data_movement/mod.rs b/crates/ecstore/src/data_movement/mod.rs index 8f7d63580..1f4daa703 100644 --- a/crates/ecstore/src/data_movement/mod.rs +++ b/crates/ecstore/src/data_movement/mod.rs @@ -471,8 +471,60 @@ fn resolve_data_movement_abort_result( )) } -fn data_movement_stage_error(op_label: &str, stage: &str, bucket: &str, object: &str, err: impl std::fmt::Display) -> Error { - Error::other(format!("{op_label}: {stage} failed for {bucket}/{object}: {err}")) +/// A data-movement stage failure that keeps the error it wrapped. +/// +/// The rendered message is byte-identical to the `format!` this replaced, so +/// logs and any message-matching callers are unaffected. What changes is that +/// the original error stays reachable through `source()`, which is what lets +/// the decommission loop classify by type instead of by substring +/// (backlog#1827 T2). +#[derive(Debug)] +struct DataMovementStageError { + rendered: String, + source: Box, +} + +impl std::fmt::Display for DataMovementStageError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.rendered) + } +} + +impl std::error::Error for DataMovementStageError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.source.as_ref()) + } +} + +fn data_movement_stage_error(op_label: &str, stage: &str, bucket: &str, object: &str, err: E) -> Error +where + E: std::error::Error + Send + Sync + 'static, +{ + let rendered = format!("{op_label}: {stage} failed for {bucket}/{object}: {err}"); + Error::other(DataMovementStageError { + rendered, + source: Box::new(err), + }) +} + +#[cfg(test)] +pub(crate) fn data_movement_stage_error_for_test(op_label: &str, stage: &str, bucket: &str, object: &str, err: Error) -> Error { + data_movement_stage_error(op_label, stage, bucket, object, err) +} + +/// Recover the error a [`data_movement_stage_error`] wrapped, if this is one. +/// +/// `Error::other` boxes through `std::io::Error`, so the chain is +/// `StorageError::Io` -> `DataMovementStageError` -> the original error. +pub(crate) fn data_movement_stage_source(err: &Error) -> Option<&Error> { + let Error::Io(io_err) = err else { + return None; + }; + io_err + .get_ref()? + .downcast_ref::()? + .source + .downcast_ref::() } fn schedule_data_movement_multipart_abort_cleanup( @@ -1865,6 +1917,40 @@ mod tests { assert!(message.contains(Error::SlowDown.to_string().as_str())); } + #[test] + fn stage_error_renders_exactly_as_the_format_it_replaced() { + // The wrapper gained a source; its message must not have moved, or log + // scrapers and any message-matching caller would break (backlog#1827 T2). + // `Error::other` renders through `StorageError::Io`, which prefixes + // "Io error: " — that was true of the `format!` this replaced too, so + // the full string is what must stay stable. + let err = data_movement_stage_error("rebalance_object", "put_object", "bucket-a", "object-a", Error::SlowDown); + assert_eq!( + err.to_string(), + format!("Io error: rebalance_object: put_object failed for bucket-a/object-a: {}", Error::SlowDown) + ); + assert_eq!( + err.to_string(), + Error::other(format!("rebalance_object: put_object failed for bucket-a/object-a: {}", Error::SlowDown)).to_string() + ); + } + + #[test] + fn stage_error_keeps_the_wrapped_error_recoverable() { + for original in [Error::DiskFull, Error::StorageFull, Error::FileNotFound, Error::SlowDown] { + let wrapped = + data_movement_stage_error("decommission_object", "put_object", "bucket-a", "object-a", original.clone()); + let recovered = data_movement_stage_source(&wrapped).expect("the wrapped error must be recoverable"); + assert_eq!(recovered.to_string(), original.to_string()); + } + } + + #[test] + fn stage_source_ignores_errors_it_did_not_wrap() { + assert!(data_movement_stage_source(&Error::DiskFull).is_none()); + assert!(data_movement_stage_source(&Error::other("plain io error")).is_none()); + } + #[test] fn test_data_movement_part_stage_error_includes_stage_object_and_part() { let err =