From a9670ebb4debea0d2db9b7ffc71b6d0aef750146 Mon Sep 17 00:00:00 2001 From: cxymds Date: Thu, 8 Oct 2026 20:03:33 +0800 Subject: [PATCH] fix(ci): preserve heal quorum retries and synchronize background tests (#8422) --- crates/heal/src/error.rs | 47 ++++++++++++-- .../src/app/lifecycle_transition_api_test.rs | 61 +++++++++---------- rustfs/src/connect/diagnostics/schedule.rs | 59 ++++++++++++++++++ rustfs/tests/connect_diagnostic_schedule.rs | 28 --------- 4 files changed, 132 insertions(+), 63 deletions(-) diff --git a/crates/heal/src/error.rs b/crates/heal/src/error.rs index 56a6f23f5..d1db7292a 100644 --- a/crates/heal/src/error.rs +++ b/crates/heal/src/error.rs @@ -141,7 +141,7 @@ impl Error { return true; } err.is_quorum_error() - || matches!(err, EcstoreError::Io(error) if is_recoverable_internode_error(error)) + || matches!(err, EcstoreError::Io(error) if is_recoverable_io_error(error)) || matches!( err, EcstoreError::DiskNotFound @@ -171,11 +171,11 @@ impl Error { | DiskError::FaultyRemoteDisk | DiskError::FaultyDisk | DiskError::RemoteClientUnavailable(_) - ) || matches!(err, DiskError::Io(error) if is_recoverable_internode_error(error)) + ) || matches!(err, DiskError::Io(error) if is_recoverable_io_error(error)) || is_recoverable_heal_error_message(&err.to_string()) } Error::TaskExecutionFailed { message } | Error::Other(message) => is_recoverable_heal_error_message(message), - Error::Io(err) => is_recoverable_internode_error(err) || is_recoverable_heal_error_message(&err.to_string()), + Error::Io(err) => is_recoverable_io_error(err) || is_recoverable_heal_error_message(&err.to_string()), _ => false, } } @@ -222,7 +222,15 @@ impl Error { } } -fn is_recoverable_internode_error(error: &std::io::Error) -> bool { +fn is_recoverable_io_error(error: &std::io::Error) -> bool { + // ObjectIO preserves store errors as IO sources; their quorum identity must survive this boundary. + if error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .is_some_and(EcstoreError::is_quorum_error) + { + return true; + } // A peer restart can cancel an RPC after a partial repair. Replay it within // the existing heal retry budget; task cancellation remains terminal. if DiskError::io_error_is_rpc_cancelled(error) { @@ -301,6 +309,37 @@ mod tests { use super::Error; use crate::heal::{DiskError, EcstoreError}; + #[test] + fn namespace_lock_quorum_is_recoverable_across_io_error_conversions() { + for (required, achieved) in [(2, 1), (1, 0)] { + let quorum = || EcstoreError::NamespaceLockQuorumUnavailable { + mode: "write", + bucket: "bucket".to_owned(), + object: "object".to_owned(), + required, + achieved, + }; + for error in [ + Error::Storage(quorum()), + Error::Io(std::io::Error::from(quorum())), + Error::Storage(EcstoreError::from(std::io::Error::from(quorum()))), + Error::Disk(DiskError::from(std::io::Error::from(quorum()))), + ] { + assert!( + error.is_recoverable_object_heal(), + "typed lock quorum failure must remain retryable: {error:?}" + ); + } + } + + let message = "Namespace lock quorum unavailable for write lock on bucket/object: required 2, achieved 1"; + assert!(!Error::Io(std::io::Error::other(message)).is_recoverable_object_heal()); + let denied = EcstoreError::Lock(rustfs_lock::LockError::PermissionDenied { + reason: message.to_owned(), + }); + assert!(!Error::Storage(EcstoreError::from(std::io::Error::from(denied))).is_recoverable_object_heal()); + } + #[test] fn cancelled_rpc_is_recoverable_across_storage_error_conversions() { let status = tonic::Status::cancelled("operation was canceled"); diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 4af5540b1..a35a12446 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -407,11 +407,15 @@ async fn wait_for_restore_completion( // runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the // `env::set_var` / `env::remove_var` window. #[allow(unsafe_code)] -async fn with_forced_immediate_enqueue_timeout(test_fn: F) +async fn with_forced_immediate_enqueue_timeout(backend: &MockWarmBackend, test_fn: F) where F: FnOnce() -> Fut, Fut: std::future::Future, { + let transition_state = get_global_transition_state(); + let missed_before = transition_state.missed_immediate_tasks(); + let compensation_before = transition_state.compensation_scheduled_tasks(); + let put_barrier = backend.arm_put_barrier().await; let original = env::var_os(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT); unsafe { env::set_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, "1"); @@ -428,6 +432,20 @@ where if let Err(err) = result { std::panic::resume_unwind(err); } + + assert!( + transition_state.missed_immediate_tasks() > missed_before, + "the operation should exercise the forced immediate enqueue failure" + ); + assert!( + transition_state.compensation_scheduled_tasks() > compensation_before, + "the operation should schedule automatic compensation backfill" + ); + + // Separate automatic discovery and queue latency from the metadata commit deadline. + put_barrier.wait_until_paused().await; + assert_eq!(backend.object_count().await, 1, "compensation should store the body before committing"); + put_barrier.release(); } async fn with_get_codec_streaming_remote_probe_env(test_fn: F) @@ -1641,7 +1659,7 @@ async fn immediate_transition_timeout_eventually_completes_via_compensation() { .await .expect("Failed to set lifecycle configuration"); - with_forced_immediate_enqueue_timeout(|| async { + with_forced_immediate_enqueue_timeout(&backend, || async { let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await; }) .await; @@ -1650,6 +1668,7 @@ async fn immediate_transition_timeout_eventually_completes_via_compensation() { .await .expect("object should eventually transition after compensation backfill"); + assert_eq!(backend.put_count().await, 1, "compensation should upload the version once"); assert_eq!(info.transitioned_object.status, "complete"); assert_eq!(info.transitioned_object.tier, tier_name); assert!(backend.contains(&info.transitioned_object.name).await); @@ -1689,37 +1708,13 @@ async fn compensation_driven_copy_still_completes_transition() { .build() .unwrap(); - let transition_state = get_global_transition_state(); - let missed_before = transition_state.missed_immediate_tasks(); - let compensation_before = transition_state.compensation_scheduled_tasks(); - let put_barrier = backend.arm_put_barrier().await; - - with_forced_immediate_enqueue_timeout(|| async { + with_forced_immediate_enqueue_timeout(&backend, || async { Box::pin(usecase.execute_copy_object(build_request(copy_input, Method::PUT))) .await .expect("Failed to copy object through usecase"); }) .await; - assert!( - transition_state.missed_immediate_tasks() > missed_before, - "copy should exercise the forced immediate enqueue failure" - ); - assert!( - transition_state.compensation_scheduled_tasks() > compensation_before, - "copy should schedule automatic compensation backfill" - ); - - // Observe automatic compensation reaching the tier before timing its final - // metadata commit; discovery and queue latency are separate from that commit. - put_barrier.wait_until_paused().await; - assert_eq!( - backend.object_count().await, - 1, - "compensation should store the copied body before committing" - ); - put_barrier.release(); - let info = wait_for_transition(&ecstore, dst_bucket.as_str(), dst_object, TRANSITION_WAIT_TIMEOUT) .await .expect("copied object should eventually transition after compensation backfill"); @@ -1792,7 +1787,7 @@ async fn compensation_driven_complete_multipart_upload_still_transitions_inner() .build() .unwrap(); - with_forced_immediate_enqueue_timeout(|| async { + with_forced_immediate_enqueue_timeout(&backend, || async { Box::pin(usecase.execute_complete_multipart_upload(build_request(complete_input, Method::POST))) .await .expect("Failed to complete multipart upload through usecase"); @@ -1803,6 +1798,7 @@ async fn compensation_driven_complete_multipart_upload_still_transitions_inner() .await .expect("multipart object should eventually transition after compensation backfill"); + assert_eq!(backend.put_count().await, 1, "compensation should upload the version once"); assert_eq!(info.transitioned_object.status, "complete"); assert_eq!(info.transitioned_object.tier, tier_name); assert!(backend.contains(&info.transitioned_object.name).await); @@ -1829,7 +1825,7 @@ async fn compensation_driven_transition_still_cleans_remote_tier_on_delete() { .await .expect("Failed to set lifecycle configuration"); - with_forced_immediate_enqueue_timeout(|| async { + with_forced_immediate_enqueue_timeout(&backend, || async { let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await; }) .await; @@ -1837,6 +1833,7 @@ async fn compensation_driven_transition_still_cleans_remote_tier_on_delete() { let transitioned = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT) .await .expect("object should eventually transition after compensation backfill"); + assert_eq!(backend.put_count().await, 1, "compensation should upload the version once"); let remote_object = transitioned.transitioned_object.name.clone(); assert!(backend.contains(&remote_object).await); @@ -1890,7 +1887,7 @@ async fn compensation_driven_versioned_delete_still_creates_delete_marker() { .await .expect("Failed to set lifecycle configuration"); - with_forced_immediate_enqueue_timeout(|| async { + with_forced_immediate_enqueue_timeout(&backend, || async { let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await; }) .await; @@ -1898,6 +1895,7 @@ async fn compensation_driven_versioned_delete_still_creates_delete_marker() { let transitioned = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT) .await .expect("object should eventually transition after compensation backfill"); + assert_eq!(backend.put_count().await, 1, "compensation should upload the version once"); let remote_object = transitioned.transitioned_object.name.clone(); assert!(backend.contains(&remote_object).await); @@ -1945,7 +1943,7 @@ async fn compensation_driven_delete_marker_still_honors_lifecycle_cleanup() { .await .expect("Failed to set transition lifecycle configuration"); - with_forced_immediate_enqueue_timeout(|| async { + with_forced_immediate_enqueue_timeout(&backend, || async { let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await; }) .await; @@ -1953,6 +1951,7 @@ async fn compensation_driven_delete_marker_still_honors_lifecycle_cleanup() { let transitioned = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT) .await .expect("object should eventually transition after compensation backfill"); + assert_eq!(backend.put_count().await, 1, "compensation should upload the version once"); let remote_object = transitioned.transitioned_object.name.clone(); assert!(backend.contains(&remote_object).await); diff --git a/rustfs/src/connect/diagnostics/schedule.rs b/rustfs/src/connect/diagnostics/schedule.rs index 5dfdbf5f1..7227c5737 100644 --- a/rustfs/src/connect/diagnostics/schedule.rs +++ b/rustfs/src/connect/diagnostics/schedule.rs @@ -701,6 +701,65 @@ mod tests { }) } + #[tokio::test] + async fn consent_revoke_cancels_retry_and_persists_receipt() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::sync::Notify; + + let temp = tempfile::tempdir().expect("tempdir"); + let store = StateStore::new(temp.path()); + let (policy_tx, policy_rx) = watch::channel(policy(4, true)); + let shutdown = CancellationToken::new(); + let attempts = Arc::new(AtomicUsize::new(0)); + let release = Arc::new(Notify::new()); + let failed = Arc::new(Notify::new()); + let runner: Runner = Arc::new({ + let attempts = Arc::clone(&attempts); + let release = Arc::clone(&release); + let failed = Arc::clone(&failed); + move |_| { + let attempts = Arc::clone(&attempts); + let release = Arc::clone(&release); + let failed = Arc::clone(&failed); + Box::pin(async move { + attempts.fetch_add(1, Ordering::SeqCst); + release.notified().await; + failed.notify_one(); + Err(DiagnosticScheduleError::InventoryUnavailable) + }) + } + }); + let mut runtime = spawn_schedule(store.clone(), policy_rx, shutdown.clone(), runner); + let mut receipts = runtime.receipts(); + wait_running(&mut runtime.status).await; + + // Pause only after real state IO completes, then revoke within the retry backoff. + tokio::time::pause(); + release.notify_one(); + tokio::time::timeout(Duration::from_secs(10), failed.notified()) + .await + .expect("first attempt should fail before revocation"); + let mut revoked = policy(5, false); + revoked.reason_code = Some(ReasonCode::ConsentInactive); + policy_tx.send(revoked).expect("revoke consent"); + tokio::time::resume(); + + tokio::time::timeout(Duration::from_secs(10), receipts.changed()) + .await + .expect("cancel receipt timeout") + .expect("cancel receipt should be published"); + let cancelled = receipts.borrow().clone().expect("cancel receipt"); + assert_eq!(cancelled.outcome, ReceiptOutcome::Cancelled); + assert_eq!(cancelled.policy_revision, 4); + assert_eq!(cancelled.attempt_count, 1); + assert_eq!(attempts.load(Ordering::SeqCst), 1, "revocation must prevent another attempt"); + shutdown.cancel(); + runtime.shutdown().await; + let persisted = store.read().await.expect("durable cancelled state"); + assert_eq!(persisted.last_receipt, Some(cancelled)); + assert!(persisted.active_interval_started_at.is_none()); + } + #[tokio::test] async fn bootstrap_stopped_policy_preserves_a_due_interval_until_fresh_policy_arrives() { let temp = tempfile::tempdir().expect("tempdir"); diff --git a/rustfs/tests/connect_diagnostic_schedule.rs b/rustfs/tests/connect_diagnostic_schedule.rs index 0fbf251b2..4e302c966 100644 --- a/rustfs/tests/connect_diagnostic_schedule.rs +++ b/rustfs/tests/connect_diagnostic_schedule.rs @@ -112,31 +112,3 @@ async fn missing_inventory_has_bounded_retries_and_durable_failure_receipt() { shutdown.cancel(); runtime.shutdown().await; } - -#[tokio::test] -async fn consent_revoke_cancels_retry_and_persists_receipt() { - let temp = tempfile::tempdir().expect("tempdir"); - let (policy_tx, policy_rx) = watch::channel(policy(4, "RUNNING", None)); - let shutdown = CancellationToken::new(); - let runtime = spawn_environment_schedule(temp.path(), policy_rx, shutdown.clone()).expect("scheduler"); - let mut status = runtime.status(); - tokio::time::timeout(Duration::from_secs(2), async { - while !matches!(&*status.borrow(), DiagnosticScheduleStatus::Running { .. }) { - status.changed().await.expect("schedule remains active"); - } - }) - .await - .expect("first attempt"); - - policy_tx - .send(policy(5, "STOPPED", Some("CONSENT_INACTIVE"))) - .expect("revoke policy"); - let cancelled = receipt(&mut status).await; - assert_eq!(cancelled.outcome, ReceiptOutcome::Cancelled); - let state: serde_json::Value = - serde_json::from_slice(&std::fs::read(temp.path().join("diagnostics/schedule.json")).expect("durable state")) - .expect("state JSON"); - assert_eq!(state["lastReceipt"]["outcome"], "CANCELLED"); - shutdown.cancel(); - runtime.shutdown().await; -}