gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-12 23:52:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ci): preserve heal quorum retries and synchronize background tests (#8422)

Browse Source
This commit is contained in:
cxymds
2026-10-08 20:03:33 +08:00
committed by GitHub
parent 99ae115731
commit a9670ebb4d
4 changed files with 132 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/error.rs
+43 -4
View File
@@ -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::<EcstoreError>())
.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");
rustfs/src/app/lifecycle_transition_api_test.rs
+30 -31
View File
@@ -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<F, Fut>(test_fn: F)
async fn with_forced_immediate_enqueue_timeout<F, Fut>(backend: &MockWarmBackend, test_fn: F)
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = ()>,
{
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<F, Fut>(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);
rustfs/src/connect/diagnostics/schedule.rs
+59
View File
@@ -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");
rustfs/tests/connect_diagnostic_schedule.rs
-28
View File
@@ -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;
}