From c02967baf6b27eaface884fea16718ad51c71ea0 Mon Sep 17 00:00:00 2001 From: overtrue Date: Tue, 8 Sep 2026 15:45:12 +0800 Subject: [PATCH] fix: address confirmed release validation regressions --- .config/e2e-full-selection.txt | 4 +- .github/workflows/audit.yml | 4 +- crates/e2e_test/src/chaos.rs | 54 +++ crates/e2e_test/src/distributed/heal_test.rs | 7 +- .../src/heal_erasure_disk_rebuild_test.rs | 7 +- .../src/inline_fast_path_cluster_test.rs | 2 +- crates/ecstore/src/erasure/coding/encode.rs | 142 +++++-- crates/heal/src/heal/manager/tests.rs | 6 +- crates/heal/src/heal/mrf_queue.rs | 369 ++++++++++++++++-- crates/heal/src/heal/task/heal_object.rs | 5 - crates/heal/src/heal/task/tests.rs | 122 +++++- crates/heal/tests/mrf_pipeline_test.rs | 51 +-- crates/kms/src/backends/local.rs | 154 +++++++- docs/testing/ci-gates.md | 7 +- scripts/error-other-format-baseline.txt | 2 +- 15 files changed, 827 insertions(+), 109 deletions(-) diff --git a/.config/e2e-full-selection.txt b/.config/e2e-full-selection.txt index 349e17819..d519fff8e 100644 --- a/.config/e2e-full-selection.txt +++ b/.config/e2e-full-selection.txt @@ -1,2 +1,2 @@ -sha256-darwin=f0c78fdb93471575d9a64c5c46eae6c806bdd0bc10a6e33d7fb574aabd8db5a3 -sha256-linux=03ed7016cab672de9320e31375a0358eceacb4408b0e79cf063614fa7c878b87 +sha256-darwin=71d04825c143b85334802dbd2d67df6f08006f18dc2835df85bc21ea85eca02a +sha256-linux=baee6b0c8b38a5be139d5583bcd4fd645a27659a70b2ca8f81646d760c299395 diff --git a/.github/workflows/audit.yml b/.github/workflows/audit.yml index 78f694aad..36d1d6b8f 100644 --- a/.github/workflows/audit.yml +++ b/.github/workflows/audit.yml @@ -16,7 +16,7 @@ name: Security Audit on: push: - branches: [ main ] + branches: [ main, release ] paths: - '**/Cargo.toml' - '**/Cargo.lock' @@ -32,7 +32,7 @@ on: - 'scripts/security/check_workflow_pins.sh' pull_request: types: [ opened, synchronize, reopened, closed ] - branches: [ main ] + branches: [ main, release ] paths: - '**/Cargo.toml' - '**/Cargo.lock' diff --git a/crates/e2e_test/src/chaos.rs b/crates/e2e_test/src/chaos.rs index f0b2971e9..cd4fdc4d9 100644 --- a/crates/e2e_test/src/chaos.rs +++ b/crates/e2e_test/src/chaos.rs @@ -44,6 +44,7 @@ use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, BTreeSet}; use std::error::Error; use std::path::{Path, PathBuf}; +use tokio::time::{Duration, Instant, sleep}; use tracing::info; use uuid::Uuid; use walkdir::WalkDir; @@ -374,6 +375,33 @@ pub(crate) fn census_object_version_on_disk( }) } +/// Wait for the background PUT tail to commit every physical part on one disk. +/// Invalid metadata remains an immediate error instead of a retryable absence. +pub(crate) async fn wait_for_complete_physical_shard_on_disk( + disk: &Path, + bucket: &str, + key: &str, + version_id: Option<&str>, + timeout: Duration, +) -> ChaosResult { + let deadline = Instant::now() + timeout; + loop { + let census = census_object_version_on_disk(disk, bucket, key, version_id)?; + if census.is_complete() && !census.expected_part_numbers.is_empty() { + return Ok(census); + } + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(format!( + "physical shard for {bucket}/{key}@{version_id:?} on {} did not become complete within {timeout:?}: {census:?}", + disk.display() + ) + .into()); + } + sleep(remaining.min(Duration::from_millis(50))).await; + } +} + /// `POST` a signed (SigV4, service `s3`) admin request without relying on the /// external `awscurl` binary. Mirrors the admin heal calls used by the heal /// regression suite. @@ -451,4 +479,30 @@ mod tests { assert!(expected.matches_manifest(&expected)); assert!(!changed.matches_manifest(&expected)); } + + #[tokio::test] + async fn physical_shard_readiness_fails_closed_with_last_census() { + let disk = tempfile::tempdir().expect("temporary disk"); + let error = wait_for_complete_physical_shard_on_disk(disk.path(), "bucket", "missing", None, Duration::ZERO) + .await + .expect_err("missing physical shards must fail the baseline gate"); + assert!(error.to_string().contains("has_xl_meta: false")); + assert!(error.to_string().contains("bucket/missing")); + } + + #[tokio::test] + async fn physical_shard_readiness_does_not_retry_invalid_metadata() { + let disk = tempfile::tempdir().expect("temporary disk"); + let object = disk.path().join("bucket").join("corrupt"); + std::fs::create_dir_all(&object).expect("object directory"); + std::fs::write(object.join("xl.meta"), b"invalid metadata").expect("corrupt metadata fixture"); + let error = tokio::time::timeout( + Duration::from_secs(1), + wait_for_complete_physical_shard_on_disk(disk.path(), "bucket", "corrupt", None, Duration::from_secs(30)), + ) + .await + .expect("corrupt metadata must fail immediately") + .expect_err("invalid metadata must not be accepted as a complete baseline"); + assert!(!error.to_string().contains("did not become complete")); + } } diff --git a/crates/e2e_test/src/distributed/heal_test.rs b/crates/e2e_test/src/distributed/heal_test.rs index 01d3f099a..29b79a7d6 100644 --- a/crates/e2e_test/src/distributed/heal_test.rs +++ b/crates/e2e_test/src/distributed/heal_test.rs @@ -13,7 +13,9 @@ // limitations under the License. use super::harness::{DistCluster, DistLayout, TestResult, assert_inventory, payload_for, put_object, unique_bucket, wait_until}; -use crate::chaos::{VersionShardCensus, census_object_version_on_disk, signed_admin_post}; +use crate::chaos::{ + VersionShardCensus, census_object_version_on_disk, signed_admin_post, wait_for_complete_physical_shard_on_disk, +}; use crate::common::init_logging; use aws_sdk_s3::Client; use aws_sdk_s3::primitives::ByteStream; @@ -110,7 +112,8 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res let replaced_drive = PathBuf::from(&dist.cluster.nodes[replaced_node].data_dirs[replaced_drive_index]); for item in &mut expected { - item.baseline = census_object_version_on_disk(&replaced_drive, &bucket, &item.key, None)?; + item.baseline = + wait_for_complete_physical_shard_on_disk(&replaced_drive, &bucket, &item.key, None, Duration::from_secs(10)).await?; assert_ec84_geometry(&item.baseline, &item.key)?; } diff --git a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs index 0b6e30bd6..1a0eccfd8 100644 --- a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs +++ b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs @@ -16,7 +16,10 @@ #[cfg(test)] mod tests { - use crate::chaos::{VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post}; + use crate::chaos::{ + VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post, + wait_for_complete_physical_shard_on_disk, + }; use crate::common::{ ClusterTopology, FAST_DATA_USAGE_SCANNER_ENV, RustFSTestClusterEnvironment, RustFSTestEnvironment, admin_request, init_logging, rustfs_binary_path, @@ -1210,6 +1213,8 @@ mod tests { attempt_count += 1; continue; } + let shard_census = + wait_for_complete_physical_shard_on_disk(&replaced_disk, bucket, &key, None, Duration::from_secs(10)).await?; assert!( shard_census.is_complete(), "node 1 should hold a complete baseline shard for {key}: {shard_census:?}" diff --git a/crates/e2e_test/src/inline_fast_path_cluster_test.rs b/crates/e2e_test/src/inline_fast_path_cluster_test.rs index cd013e848..b568364d7 100644 --- a/crates/e2e_test/src/inline_fast_path_cluster_test.rs +++ b/crates/e2e_test/src/inline_fast_path_cluster_test.rs @@ -2234,7 +2234,6 @@ async fn four_node_manual_transition_distributed_admission_conflict_reports_stat let bucket = format!("distributed-admission-{}", Uuid::new_v4().simple()); let prefix = "transition/distributed-admission/"; hot_client.create_bucket().bucket(&bucket).send().await?; - put_lifecycle_with_transition_retry(&hot_client, &bucket, &tier_name).await?; for index in 0u8..64 { let key = format!("{prefix}object-{index:02}.bin"); hot_client @@ -2245,6 +2244,7 @@ async fn four_node_manual_transition_distributed_admission_conflict_reports_stat .send() .await?; } + put_lifecycle_with_transition_retry(&hot_client, &bucket, &tier_name).await?; let (node0, node1) = tokio::join!( start_manual_transition_job_on_node(&hot, 0, &bucket, prefix, &tier_name, false, 64), diff --git a/crates/ecstore/src/erasure/coding/encode.rs b/crates/ecstore/src/erasure/coding/encode.rs index 7872499e4..261548c25 100644 --- a/crates/ecstore/src/erasure/coding/encode.rs +++ b/crates/ecstore/src/erasure/coding/encode.rs @@ -431,7 +431,7 @@ impl<'a> MultiWriter<'a> { errs = ?self.errs, "Erasure encode write quorum unavailable: {summary_text}" ); - Err(std::io::Error::other(format!("Failed to write data: {summary_text}"))) + Err(write_err.into()) } async fn shutdown_writer(writer_opt: &mut Option, err: &mut Option) { @@ -503,7 +503,7 @@ impl<'a> MultiWriter<'a> { errs = ?self.errs, "Erasure encode shutdown quorum unavailable: {summary_text}" ); - Err(std::io::Error::other(format!("Failed to shutdown writers: {summary_text}"))) + Err(write_err.into()) } } @@ -1002,6 +1002,7 @@ impl Erasure { mod tests { use super::*; use crate::erasure::coding::{BitrotWriterWrapper, CustomWriter}; + use crate::error::StorageError; use rustfs_rio::HardLimitReader; use rustfs_utils::HashAlgorithm; use std::future::Future; @@ -1451,7 +1452,14 @@ mod tests { Ok(_) => panic!("writer quorum failure should fail the encode pipeline"), Err(err) => err, }; - assert!(err.to_string().contains("Failed to write data")); + let err = StorageError::from(err); + assert!(matches!( + &err, + StorageError::Io(source) + if source.kind() == std::io::ErrorKind::Other + && source.to_string() == "injected write failure after producer blocks" + )); + assert!(!err.is_quorum_error()); tokio::time::timeout(Duration::from_secs(1), reader_dropped) .await .expect("writer failure should abort the blocked producer") @@ -1644,7 +1652,7 @@ mod tests { #[tokio::test] async fn multi_writer_short_write_fails_before_shutdown() { - let mut writers = vec![Some(bitrot_writer(ShortWriteWriter, 16))]; + let mut writers = vec![Some(bitrot_writer(ShortWriteWriter, 32))]; let err = { let mut writer = MultiWriter::new(&mut writers, 1); writer @@ -1653,63 +1661,93 @@ mod tests { .expect_err("short writes must fail the shard writer") }; - assert!(err.to_string().contains("Failed to write data")); + let err = StorageError::from(err); + assert!(matches!(&err, StorageError::Io(source) if source.kind() == std::io::ErrorKind::WriteZero)); + assert!(!err.is_quorum_error()); assert!(writers[0].is_none(), "short-write shard must be removed before commit"); } #[tokio::test] async fn multi_writer_reports_fallback_summary_when_only_offline_writers_remain() { let mut writers = vec![None, None]; - let err = { + let (err, summary) = { let mut writer = MultiWriter::new(&mut writers, 1); - writer + let err = writer .write(vec![Bytes::from_static(b"offline-a"), Bytes::from_static(b"offline-b")]) .await - .expect_err("offline writers cannot satisfy write quorum") + .expect_err("offline writers cannot satisfy write quorum"); + let summary = build_write_quorum_failure_summary(&writer.errs, OBJECT_OP_IGNORED_ERRS, writer.write_quorum); + (err, format_write_quorum_failure(&summary)) }; - let err = err.to_string(); - assert!(err.contains("Failed to write data")); - assert!(err.contains("offline-disks=2/2")); - assert!(err.contains("required=1")); + assert_eq!( + err.get_ref().and_then(|source| source.downcast_ref::()), + Some(&Error::ErasureWriteQuorum), + ); + let err = StorageError::from(err); + assert_eq!(err, StorageError::ErasureWriteQuorum); + assert!(err.is_quorum_error()); + assert!(summary.contains("offline-disks=2/2")); + assert!(summary.contains("required=1")); - let shutdown_err = { + let (shutdown_err, summary) = { let mut writer = MultiWriter::new(&mut writers, 1); - writer + let err = writer .shutdown() .await - .expect_err("offline writers cannot satisfy shutdown quorum") + .expect_err("offline writers cannot satisfy shutdown quorum"); + let summary = build_write_quorum_failure_summary(&writer.errs, OBJECT_OP_IGNORED_ERRS, writer.write_quorum); + (err, format_write_quorum_failure(&summary)) }; - let shutdown_err = shutdown_err.to_string(); - assert!(shutdown_err.contains("Failed to shutdown writers")); - assert!(shutdown_err.contains("offline-disks=2/2")); - assert!(shutdown_err.contains("required=1")); + assert_eq!( + shutdown_err.get_ref().and_then(|source| source.downcast_ref::()), + Some(&Error::ErasureWriteQuorum), + ); + let shutdown_err = StorageError::from(shutdown_err); + assert_eq!(shutdown_err, StorageError::ErasureWriteQuorum); + assert!(shutdown_err.is_quorum_error()); + assert!(summary.contains("offline-disks=2/2")); + assert!(summary.contains("required=1")); } #[tokio::test] async fn multi_writer_reports_quorum_failure_when_quorum_exceeds_writer_count() { let committed = Arc::new(Mutex::new(Vec::new())); - let mut writers = vec![Some(bitrot_writer(DeferredCommitWriter::new(committed), 16))]; + let mut writers = vec![Some(bitrot_writer(DeferredCommitWriter::new(committed), 32))]; let mut writer = MultiWriter::new(&mut writers, 2); let err = writer .write(vec![Bytes::from_static(b"quorum impossible")]) .await .expect_err("write quorum above writer count must fail"); - let err = err.to_string(); - assert!(err.contains("Failed to write data")); - assert!(err.contains("required=2")); - assert!(err.contains("erasure write quorum")); + assert_eq!( + err.get_ref().and_then(|source| source.downcast_ref::()), + Some(&Error::ErasureWriteQuorum), + ); + let err = StorageError::from(err); + assert_eq!(err, StorageError::ErasureWriteQuorum); + assert!(err.is_quorum_error()); + let summary = build_write_quorum_failure_summary(&writer.errs, OBJECT_OP_IGNORED_ERRS, writer.write_quorum); + let summary = format_write_quorum_failure(&summary); + assert!(summary.contains("required=2")); + assert!(summary.contains("erasure write quorum")); let shutdown_err = writer .shutdown() .await .expect_err("shutdown quorum above writer count must fail"); - let shutdown_err = shutdown_err.to_string(); - assert!(shutdown_err.contains("Failed to shutdown writers")); - assert!(shutdown_err.contains("required=2")); - assert!(shutdown_err.contains("erasure write quorum")); + assert_eq!( + shutdown_err.get_ref().and_then(|source| source.downcast_ref::()), + Some(&Error::ErasureWriteQuorum), + ); + let shutdown_err = StorageError::from(shutdown_err); + assert_eq!(shutdown_err, StorageError::ErasureWriteQuorum); + assert!(shutdown_err.is_quorum_error()); + let summary = build_write_quorum_failure_summary(&writer.errs, OBJECT_OP_IGNORED_ERRS, writer.write_quorum); + let summary = format_write_quorum_failure(&summary); + assert!(summary.contains("required=2")); + assert!(summary.contains("erasure write quorum")); } // The production wiring (`MultiWriter::new`) must arm a real deadline by @@ -1794,7 +1832,13 @@ mod tests { .write(four_shards()) .await .expect_err("two stalled writers must fail the write quorum instead of hanging"); - assert!(err.to_string().contains("Failed to write data")); + assert_eq!( + err.get_ref().and_then(|source| source.downcast_ref::()), + Some(&Error::ErasureWriteQuorum), + ); + let err = StorageError::from(err); + assert_eq!(err, StorageError::ErasureWriteQuorum); + assert!(err.is_quorum_error()); } // A small object whose bytes were fully buffered leaves `write` succeeding @@ -1839,7 +1883,13 @@ mod tests { .shutdown() .await .expect_err("two shutdown stalls must fail the shutdown quorum instead of hanging"); - assert!(err.to_string().contains("Failed to shutdown writers")); + assert_eq!( + err.get_ref().and_then(|source| source.downcast_ref::()), + Some(&Error::ErasureWriteQuorum), + ); + let err = StorageError::from(err); + assert_eq!(err, StorageError::ErasureWriteQuorum); + assert!(err.is_quorum_error()); } // A slow-but-honest writer that keeps completing shards (delay < stall @@ -2121,7 +2171,13 @@ mod tests { .await .expect_err("streaming encode must fail when write quorum is unavailable"); - assert!(err.to_string().contains("Failed to write data")); + assert_eq!( + err.get_ref().and_then(|source| source.downcast_ref::()), + Some(&Error::ErasureWriteQuorum), + ); + let err = StorageError::from(err); + assert_eq!(err, StorageError::ErasureWriteQuorum); + assert!(err.is_quorum_error()); } #[tokio::test] @@ -2145,7 +2201,13 @@ mod tests { .await .expect_err("write quorum failure must fail the inline encode"); - assert!(err.to_string().contains("Failed to write data")); + assert_eq!( + err.get_ref().and_then(|source| source.downcast_ref::()), + Some(&Error::ErasureWriteQuorum), + ); + let err = StorageError::from(err); + assert_eq!(err, StorageError::ErasureWriteQuorum); + assert!(err.is_quorum_error()); assert!( committed.lock().expect("committed buffer should be lockable").is_empty(), "successful writer must not be committed when write quorum fails before shutdown" @@ -2173,7 +2235,13 @@ mod tests { .await .expect_err("shutdown quorum failure must fail the inline encode"); - assert!(err.to_string().contains("Failed to shutdown writers")); + let err = StorageError::from(err); + assert!(matches!( + &err, + StorageError::Io(source) + if source.kind() == std::io::ErrorKind::Other && source.to_string() == "injected shutdown failure" + )); + assert!(!err.is_quorum_error()); assert!( !committed.lock().expect("committed buffer should be lockable").is_empty(), "the successful writer should have committed before shutdown quorum failure was reported" @@ -2395,7 +2463,13 @@ mod tests { .await .expect_err("batched encode must fail when write quorum is unavailable"); - assert!(err.to_string().contains("Failed to write data")); + assert_eq!( + err.get_ref().and_then(|source| source.downcast_ref::()), + Some(&Error::ErasureWriteQuorum), + ); + let err = StorageError::from(err); + assert_eq!(err, StorageError::ErasureWriteQuorum); + assert!(err.is_quorum_error()); } #[tokio::test] diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 6e551b829..c065b33d3 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -1869,7 +1869,7 @@ async fn test_submit_heal_request_returns_merged_for_duplicate() { let storage: Arc = Arc::new(MockStorage); let manager = HealManager::new(storage, None); - let request = HealRequest::new( + let mut request = HealRequest::new( HealType::Object { bucket: "bucket".to_string(), object: "object".to_string(), @@ -1886,6 +1886,7 @@ async fn test_submit_heal_request_returns_merged_for_duplicate() { .expect("first request should be accepted"), HealAdmissionResult::Accepted ); + request.id = uuid::Uuid::new_v4().to_string(); assert_eq!( manager .submit_heal_request(request) @@ -3725,7 +3726,7 @@ async fn test_submit_heal_request_returns_merged_before_full_for_duplicate() { }), ); - let request = HealRequest::new( + let mut request = HealRequest::new( HealType::Object { bucket: "bucket".to_string(), object: "object".to_string(), @@ -3742,6 +3743,7 @@ async fn test_submit_heal_request_returns_merged_before_full_for_duplicate() { .expect("first request should be accepted"), HealAdmissionResult::Accepted ); + request.id = uuid::Uuid::new_v4().to_string(); assert_eq!( manager .submit_heal_request(request) diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 48ae146ce..d244dce70 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -38,7 +38,7 @@ use crate::heal::manager::{HealManager, MrfRepairNoticeTarget}; use metrics::{counter, gauge}; use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfDurableRepairAnchor, MrfIngressResult, MrfIntent}; use rustfs_heal_contracts::heal_channel::{HealAdmissionDropReason, HealAdmissionResult}; -use std::collections::{HashSet, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::Arc; use std::time::Duration; use tokio::sync::mpsc; @@ -134,7 +134,10 @@ struct MrfQueueKey { } fn queue_key(intent: &MrfIntent) -> MrfQueueKey { - let version_id = intent.version_id.filter(|bytes| *bytes != [0; 16]); + let version_id = (!matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption)) + .then_some(intent.version_id) + .flatten() + .filter(|bytes| *bytes != [0; 16]); let scope = (!matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption)) .then_some(intent.scope) .flatten(); @@ -533,29 +536,101 @@ struct MrfRuntime { /// Partial-write responsibilities accepted from replay and waiting for an /// exact storage-owned proof before the startup journal can be deleted. durable_replay_anchors: Vec, + /// Startup responsibilities remain in every successor snapshot until an + /// exact verified repair discharges them. Live admissions never grow this + /// set, so its size is bounded by the decoded startup journal. + retained_replay_intents: HashMap, /// Earliest instant a full-admission retry may proceed. backoff_until: Option, } impl MrfRuntime { - fn snapshot(&self) -> (Vec, Vec) { + fn enqueue_batch(&mut self, intents: impl IntoIterator) -> usize { + // Reserve retained startup work before admitting live hints. Computing + // the union once per batch avoids scanning it for every incoming hint. + let mut snapshot_count = self.queue.depth(); + let mut snapshot_bytes = self.queue.bytes(); + for (key, intent) in &self.retained_replay_intents { + if !self.queue.pending_keys.contains(key) { + snapshot_count = snapshot_count.saturating_add(1); + snapshot_bytes = snapshot_bytes.saturating_add(intent.estimated_bytes()); + } + } + let mut enqueued = 0; + for intent in intents { + let key = queue_key(&intent); + let retained = self.retained_replay_intents.get(&key); + let additional = retained.is_none() && !self.queue.pending_keys.contains(&key); + let next_count = snapshot_count.saturating_add(usize::from(additional)); + let next_bytes = snapshot_bytes.saturating_add(if additional { intent.estimated_bytes() } else { 0 }); + let result = if retained.is_some_and(|retained| retained.lease != intent.lease) + || next_count > self.queue.capacity + || next_bytes > self.queue.byte_budget + { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "queue_overflow").increment(1); + MrfQueuePushResult::Rejected + } else { + self.queue.try_push_typed(intent.clone()) + }; + match result { + MrfQueuePushResult::Enqueued => { + snapshot_count = next_count; + snapshot_bytes = next_bytes; + enqueued += 1; + self.new_since_flush += 1; + self.dirty = true; + } + MrfQueuePushResult::Coalesced | MrfQueuePushResult::Rejected => { + rustfs_common::mrf_channel::release_mrf_intent(&intent); + } + } + } + enqueued + } + + fn snapshot(&self) -> Option<(Vec, Vec)> { let mut authoritative = Vec::new(); let mut legacy = Vec::new(); - for intent in self.queue.intents() { + let mut encoded = HashSet::new(); + for intent in self.queue.intents().chain(self.retained_replay_intents.values()) { + let key = queue_key(intent); + if self + .retained_replay_intents + .get(&key) + .is_some_and(|retained| retained.lease != intent.lease) + { + // Legacy records cannot distinguish concurrent responsibilities + // with different leases. Preserve the existing disk anchor. + return None; + } + if !encoded.insert(key) { + continue; + } + if encoded.len() > self.queue.capacity { + return None; + } let scoped_identity = !matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption) && intent.scope.is_some(); if !encode_intent(intent, &mut authoritative) { counter!("rustfs_heal_mrf_dropped_total", "reason" => "journal_identity_oversized").increment(1); + return None; + } + if authoritative.len() > self.queue.byte_budget { + return None; } if !scoped_identity && !encode_intent(intent, &mut legacy) { counter!("rustfs_heal_mrf_dropped_total", "reason" => "journal_identity_oversized").increment(1); + return None; } } - (authoritative, legacy) + Some((authoritative, legacy)) } async fn flush(&mut self) { - let (authoritative, legacy) = self.snapshot(); + let Some((authoritative, legacy)) = self.snapshot() else { + self.dirty = true; + return; + }; let authoritative_persisted = write_journal(MRF_SCOPED_JOURNAL_PATH, &authoritative).await; if !authoritative.is_empty() { counter!("rustfs_heal_mrf_journal_fsync_total").increment(1); @@ -594,11 +669,31 @@ impl MrfRuntime { // attempts counter) changes the encoded snapshot; mark it dirty // either way. self.dirty = true; + let replay_key = queue_key(&intent); + let replayed = self + .retained_replay_intents + .get(&replay_key) + .is_some_and(|retained| retained.lease == intent.lease); + if replayed { + if !matches!( + rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut intent), + MrfIngressResult::Enqueued + ) { + self.retain_replay_journal = true; + self.queue.push_back(intent); + self.backoff_until = Some(tokio::time::Instant::now() + self.config.admission_backoff); + break; + } + self.retained_replay_intents.insert(replay_key, intent.clone()); + } match submit_mrf_heal_request(manager, &intent).await { - // Accepted intents leave the pending set; the next flush persists the - // smaller snapshot. This is not a durable successor receipt and - // does not discharge the producer's existing retry hints. - Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {} + // Admission removes executable work from the pending queue, + // but startup responsibilities still require an exact proof. + Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => { + if replayed && let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await { + self.durable_replay_anchors.push(anchor); + } + } Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => { intent.attempts = intent.attempts.saturating_add(1); if intent.attempts >= MRF_MAX_ATTEMPTS { @@ -632,13 +727,14 @@ impl MrfRuntime { } fn retained_replay_journal(&self) -> bool { - self.retain_replay_journal || !self.durable_replay_anchors.is_empty() + self.retain_replay_journal || !self.retained_replay_intents.is_empty() } fn discharge_durable_replay_anchors(&mut self) { if self.durable_replay_anchors.is_empty() { return; } + let mut discharged_leases: HashSet<_> = self.durable_replay_anchors.iter().map(|anchor| anchor.lease).collect(); let mut buckets: Vec> = self .durable_replay_anchors .iter() @@ -652,6 +748,13 @@ impl MrfRuntime { &mut self.durable_replay_anchors, ); } + for anchor in &self.durable_replay_anchors { + discharged_leases.remove(&anchor.lease); + } + let before = self.retained_replay_intents.len(); + self.retained_replay_intents + .retain(|_, intent| !intent.lease.is_some_and(|lease| discharged_leases.contains(&lease))); + self.dirty |= self.retained_replay_intents.len() != before; } } @@ -704,6 +807,7 @@ struct ReplayOutcome { journal_on_disk: bool, retain_journal_for_replay: bool, durable_replay_anchors: Vec, + retained_replay_intents: HashMap, } fn replay_must_retain_journal( @@ -786,6 +890,7 @@ async fn replay_into( journal_on_disk: false, retain_journal_for_replay: false, durable_replay_anchors: Vec::new(), + retained_replay_intents: HashMap::new(), }; } Err(err) => { @@ -799,6 +904,7 @@ async fn replay_into( journal_on_disk: true, retain_journal_for_replay: true, durable_replay_anchors: Vec::new(), + retained_replay_intents: HashMap::new(), }; } }; @@ -837,6 +943,7 @@ async fn replay_into( } } } + let mut retained_replay_intents: HashMap<_, _> = queue.intents().map(|intent| (queue_key(intent), intent.clone())).collect(); // Drain the replayed intents immediately; whatever the manager refuses // stays armed in `queue` for the consumer's retry loop. @@ -851,6 +958,7 @@ async fn replay_into( *backoff_until = Some(tokio::time::Instant::now()); break; } + retained_replay_intents.insert(queue_key(&intent), intent.clone()); match submit_mrf_heal_request(manager, &intent).await { Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => { if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await { @@ -872,6 +980,7 @@ async fn replay_into( break; } Ok(HealAdmissionResult::Dropped(_)) => { + retained_replay_intents.remove(&queue_key(&intent)); rustfs_common::mrf_channel::release_mrf_intent(&intent); } Err(_) => { @@ -906,6 +1015,7 @@ async fn replay_into( journal_on_disk, retain_journal_for_replay, durable_replay_anchors, + retained_replay_intents, } } @@ -921,6 +1031,7 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive journal_on_disk: false, retain_replay_journal: false, durable_replay_anchors: Vec::new(), + retained_replay_intents: HashMap::new(), backoff_until: None, }; @@ -930,6 +1041,7 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive runtime.journal_on_disk = replay.journal_on_disk; runtime.retain_replay_journal = replay.retain_journal_for_replay; runtime.durable_replay_anchors = replay.durable_replay_anchors; + runtime.retained_replay_intents = replay.retained_replay_intents; // Anything still pending (e.g. the manager was full and backoff armed) // must be re-persisted by the next flush before replay can delete the // startup anchor. @@ -956,17 +1068,7 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive ); return; } - for intent in batch.drain(..) { - match runtime.queue.try_push_typed(intent.clone()) { - MrfQueuePushResult::Enqueued => { - runtime.new_since_flush += 1; - runtime.dirty = true; - } - MrfQueuePushResult::Coalesced | MrfQueuePushResult::Rejected => { - rustfs_common::mrf_channel::release_mrf_intent(&intent); - } - } - } + runtime.enqueue_batch(batch.drain(..)); runtime.dispatch(manager.as_ref()).await; if runtime.new_since_flush >= runtime.config.flush_threshold { runtime.flush().await; @@ -1118,6 +1220,7 @@ mod tests { journal_on_disk: true, retain_replay_journal: false, durable_replay_anchors: vec![anchor], + retained_replay_intents: HashMap::from([(queue_key(&intent), intent.clone())]), backoff_until: None, }; rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent { @@ -1135,14 +1238,196 @@ mod tests { runtime.retained_replay_journal(), "anchor must retain the startup journal before proof is consumed" ); + assert_eq!( + decode_journal(&runtime.snapshot().expect("retained snapshot").0).0.len(), + 1, + "an admitted responsibility must remain in the successor before proof" + ); runtime.discharge_durable_replay_anchors(); assert!( !runtime.retained_replay_journal(), "matching verified proof discharges the durable replay anchor" ); + assert!(runtime.dirty, "proof removal must be persisted by the next flush"); + assert!(runtime.snapshot().expect("discharged snapshot").0.is_empty()); rustfs_common::mrf_channel::release_mrf_intent(&intent); } + #[test] + fn runtime_snapshot_retains_admitted_replay_and_pending_successor() { + let accepted = intent("snapshot-bucket", "accepted", 0); + let pending = intent("snapshot-bucket", "pending", 2); + let mut queue = MrfQueue::new(4, 4096); + assert!(queue.try_push(pending.clone())); + let runtime = MrfRuntime { + queue, + config: MrfConsumerConfig::default(), + new_since_flush: 0, + dirty: true, + journal_on_disk: true, + retain_replay_journal: true, + durable_replay_anchors: Vec::new(), + retained_replay_intents: HashMap::from([ + (queue_key(&accepted), accepted), + (queue_key(&pending), intent("snapshot-bucket", "pending", 0)), + ]), + backoff_until: None, + }; + + let (authoritative, legacy) = runtime.snapshot().expect("complete successor should fit"); + for snapshot in [authoritative, legacy] { + let (recovered, truncated) = decode_journal(&snapshot); + assert_eq!(truncated, 0); + assert_eq!(recovered.len(), 2, "admission must not discard an unproven startup responsibility"); + assert!(recovered.iter().any(|intent| intent.object.as_ref() == "accepted")); + assert!( + recovered + .iter() + .any(|intent| intent.object.as_ref() == "pending" && intent.attempts == 2) + ); + } + } + + #[test] + fn runtime_snapshot_preserves_anchor_when_successor_exceeds_budget_or_changes_lease() { + let mut retained = intent("bounded-bucket", "retained", 0); + assert_eq!( + rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut retained), + MrfIngressResult::Enqueued + ); + let mut queue = MrfQueue::new(1, 4096); + assert!(queue.try_push(intent("bounded-bucket", "pending", 0))); + let mut runtime = MrfRuntime { + queue, + config: MrfConsumerConfig::default(), + new_since_flush: 0, + dirty: true, + journal_on_disk: true, + retain_replay_journal: false, + durable_replay_anchors: Vec::new(), + retained_replay_intents: HashMap::from([(queue_key(&retained), retained.clone())]), + backoff_until: None, + }; + assert!(runtime.snapshot().is_none(), "combined count must honor the queue ceiling"); + runtime.queue.capacity = 2; + runtime.queue.byte_budget = 1; + assert!(runtime.snapshot().is_none(), "oversized successor must not replace the startup journal"); + + runtime.queue = MrfQueue::new(2, 4096); + let mut newer = retained.clone(); + newer.lease = None; + assert_eq!( + rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut newer), + MrfIngressResult::Enqueued + ); + assert_ne!(retained.lease, newer.lease); + assert!(runtime.queue.try_push(newer)); + assert!( + runtime.snapshot().is_none(), + "legacy encoding cannot conflate distinct responsibilities for one object" + ); + } + + #[test] + fn runtime_admission_reserves_replay_budget_until_verified_repair() { + for count_limited in [true, false] { + let mut retained = intent("reserved-bucket", "retained", 0); + assert_eq!( + rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut retained), + MrfIngressResult::Enqueued + ); + let incarnation = Uuid::new_v4(); + let anchor = MrfDurableRepairAnchor::from_intent(&retained, incarnation).expect("replay anchor"); + let mut runtime = MrfRuntime { + queue: MrfQueue::new(if count_limited { 1 } else { 2 }, retained.estimated_bytes()), + config: MrfConsumerConfig::default(), + new_since_flush: 0, + dirty: false, + journal_on_disk: true, + retain_replay_journal: false, + durable_replay_anchors: vec![anchor], + retained_replay_intents: HashMap::from([(queue_key(&retained), retained.clone())]), + backoff_until: None, + }; + if count_limited { + runtime.queue.byte_budget = 4096; + } + let pending = intent("reserved-bucket", "pending", 0); + assert_eq!(runtime.enqueue_batch([pending.clone()]), 0, "retained work consumes admission budget"); + assert_eq!(runtime.queue.depth(), 0); + let (snapshot, _) = runtime.snapshot().expect("rejection must leave a writable retained snapshot"); + let (recovered, truncated) = decode_journal(&snapshot); + assert_eq!(truncated, 0); + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].object.as_ref(), "retained"); + + rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent { + kind: retained.kind, + bucket: retained.bucket.clone(), + object: retained.object.clone(), + version_id: retained.version_id, + scope: retained.scope, + lease: retained.lease, + bucket_incarnation_id: incarnation, + disposition: MrfVerifiedRepairDisposition::Repaired, + }); + runtime.discharge_durable_replay_anchors(); + assert_eq!(runtime.enqueue_batch([pending]), 1, "proof must release admission capacity for retry"); + let (snapshot, _) = runtime.snapshot().expect("new admitted work must be persistable"); + let (recovered, truncated) = decode_journal(&snapshot); + assert_eq!(truncated, 0); + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].object.as_ref(), "pending"); + } + } + + #[test] + fn runtime_admission_rejects_new_lease_without_blocking_other_successors() { + let mut retained = intent("lease-bucket", "retained", 0); + assert_eq!( + rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut retained), + MrfIngressResult::Enqueued + ); + let incarnation = Uuid::new_v4(); + let anchor = MrfDurableRepairAnchor::from_intent(&retained, incarnation).expect("replay anchor"); + let mut runtime = MrfRuntime { + queue: MrfQueue::new(2, 4096), + config: MrfConsumerConfig::default(), + new_since_flush: 0, + dirty: false, + journal_on_disk: true, + retain_replay_journal: false, + durable_replay_anchors: vec![anchor], + retained_replay_intents: HashMap::from([(queue_key(&retained), retained.clone())]), + backoff_until: None, + }; + let mut newer = retained.clone(); + newer.lease = None; + assert_eq!( + rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut newer), + MrfIngressResult::Enqueued + ); + assert_ne!(newer.lease, retained.lease); + assert_eq!(runtime.enqueue_batch([newer.clone(), intent("lease-bucket", "pending", 0)]), 1); + assert_eq!(runtime.queue.intents().next().expect("unrelated successor").object.as_ref(), "pending"); + assert_eq!(decode_journal(&runtime.snapshot().expect("unblocked successor").0).0.len(), 2); + + rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent { + kind: retained.kind, + bucket: retained.bucket.clone(), + object: retained.object.clone(), + version_id: retained.version_id, + scope: retained.scope, + lease: retained.lease, + bucket_incarnation_id: incarnation, + disposition: MrfVerifiedRepairDisposition::Repaired, + }); + runtime.discharge_durable_replay_anchors(); + assert_eq!(runtime.enqueue_batch([newer.clone()]), 1); + assert!(runtime.queue.intents().any(|intent| intent.lease == newer.lease)); + assert_eq!(decode_journal(&runtime.snapshot().expect("new lease successor").0).0.len(), 2); + } + #[test] fn durable_replay_acquires_a_fresh_lease_before_manager_admission() { let unique = uuid::Uuid::new_v4(); @@ -1177,6 +1462,46 @@ mod tests { rustfs_common::mrf_channel::release_mrf_intent(&replay); } + #[test] + fn metadata_replay_canonicalization_preserves_one_bounded_responsibility() { + let mut legacy = intent("metadata-replay-bucket", "object", 0); + legacy.kind = MrfKind::MetadataCorruption; + let mut bytes = Vec::new(); + assert!(encode_intent(&legacy, &mut bytes)); + let (mut decoded, truncated) = decode_journal(&bytes); + assert_eq!(truncated, 0); + let mut replay = decoded.pop().expect("legacy metadata record"); + assert!(replay.version_id.is_some(), "the legacy wire record carries an ignored version"); + let original_key = queue_key(&replay); + let mut retained_replay_intents = HashMap::from([(original_key.clone(), replay.clone())]); + assert_eq!( + rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut replay), + MrfIngressResult::Enqueued + ); + assert!(replay.version_id.is_none()); + assert_eq!(queue_key(&replay), original_key, "rearm must not create another retained key"); + retained_replay_intents.insert(queue_key(&replay), replay); + assert_eq!(retained_replay_intents.len(), 1); + let runtime = MrfRuntime { + queue: MrfQueue::new(1, 4096), + config: MrfConsumerConfig::default(), + new_since_flush: 0, + dirty: false, + journal_on_disk: true, + retain_replay_journal: true, + durable_replay_anchors: Vec::new(), + retained_replay_intents, + backoff_until: None, + }; + let (snapshot, _) = runtime + .snapshot() + .expect("canonical metadata fits the original one-record budget"); + let (recovered, truncated) = decode_journal(&snapshot); + assert_eq!(truncated, 0); + assert_eq!(recovered.len(), 1); + assert!(recovered[0].version_id.is_none()); + } + #[test] fn replay_can_arm_more_records_than_live_queue_budget() { let mut queue = MrfQueue::new(1, intent("bucket", "object-0", 0).estimated_bytes()); diff --git a/crates/heal/src/heal/task/heal_object.rs b/crates/heal/src/heal/task/heal_object.rs index 2265d1919..c7635a79f 100644 --- a/crates/heal/src/heal/task/heal_object.rs +++ b/crates/heal/src/heal/task/heal_object.rs @@ -162,11 +162,6 @@ impl HealTask { pool: self.options.pool_index, set: self.options.set_index, }; - let expected_bucket_incarnation_id = self.storage.bucket_incarnation_id(bucket).await?; - let mut expected_identity = - self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index); - expected_identity.bucket_incarnation_id = expected_bucket_incarnation_id; - let mut expected_identity = self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index); expected_identity.bucket_incarnation_id = self.outcome_bucket_incarnation_id(bucket, self.options.dry_run).await?; diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 9981d73d3..048946f2a 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -1339,6 +1339,10 @@ struct MockStorage { heal_object_outcomes: Mutex>>, heal_object_receipts: Mutex>>, bucket_incarnation_id: Mutex>, + bucket_incarnation_calls: AtomicU64, + bucket_incarnation_error: Mutex>, + block_bucket_incarnation: bool, + bucket_incarnation_started: tokio::sync::Notify, bucket_incarnation_after_object_heal: Mutex>, bucket_incarnation_unavailable: Mutex, format_no_heal_required: Mutex, @@ -1482,7 +1486,7 @@ async fn object_heal_records_matching_positive_storage_receipt() { }); let task = HealTask::from_request( HealRequest::object("bucket-a".to_string(), "object-a".to_string(), Some("version-a".to_string())), - storage, + storage.clone(), ); task.execute().await.expect("mock object heal should complete"); @@ -1495,6 +1499,114 @@ async fn object_heal_records_matching_positive_storage_receipt() { assert_eq!(object.identity.version_id.as_deref(), Some("version-a")); assert!(object.identity.bucket_incarnation_id.is_some()); assert_eq!(object.disposition, HealObjectDisposition::Repaired); + assert_eq!( + storage.bucket_incarnation_calls.load(Ordering::Relaxed), + 1, + "latch the owner exactly once before repair" + ); +} + +#[tokio::test] +async fn object_heal_owner_lookup_failure_preserves_unverified_repair() { + let storage = Arc::new(MockStorage { + bucket_incarnation_error: Mutex::new(Some(Error::other("owner metadata unavailable"))), + heal_object_receipts: Mutex::new(HashMap::from([( + "object-a".to_string(), + VecDeque::from([object_receipt( + "object-a", + None, + HealObjectDisposition::Repaired, + Uuid::new_v4(), + )]), + )])), + ..Default::default() + }); + let task = HealTask::from_request(HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None), storage.clone()); + + task.execute().await.expect("missing receipt owner must not prevent repair"); + + assert_eq!(storage.heal_object_calls.lock().expect("heal calls").as_slice(), ["object-a"]); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.healed, 0); + assert_eq!(outcome.counters.unknown, 1); + assert_eq!( + outcome.objects.front().expect("unverified outcome").disposition, + HealObjectDisposition::Unknown + ); +} + +#[tokio::test] +async fn object_heal_dry_run_skips_owner_lookup_and_positive_receipts() { + let storage = Arc::new(MockStorage { + bucket_incarnation_error: Mutex::new(Some(Error::other("dry-run must not query the receipt owner"))), + heal_object_receipts: Mutex::new(HashMap::from([( + "object-a".to_string(), + VecDeque::from([object_receipt( + "object-a", + None, + HealObjectDisposition::Repaired, + Uuid::new_v4(), + )]), + )])), + ..Default::default() + }); + let mut request = HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None); + request.options.dry_run = true; + let task = HealTask::from_request(request, storage.clone()); + + task.execute().await.expect("dry-run should complete without owner metadata"); + + assert!(storage.object_heal_opts.lock().expect("heal options")[0].dry_run); + assert_eq!(storage.bucket_incarnation_calls.load(Ordering::Relaxed), 0); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.healed, 0); + assert_eq!(outcome.counters.unknown, 0); + assert_eq!(outcome.counters.skipped, 1); + assert_eq!( + outcome.objects.front().expect("dry-run outcome").disposition, + HealObjectDisposition::DryRunObserved + ); +} + +#[tokio::test(start_paused = true)] +async fn object_heal_owner_lookup_obeys_task_timeout() { + let storage = Arc::new(MockStorage { + block_bucket_incarnation: true, + ..Default::default() + }); + let mut request = HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None); + request.options.timeout = Some(Duration::from_secs(5)); + let task = HealTask::from_request(request, storage.clone()); + + let result = tokio::time::timeout(Duration::from_secs(60), task.execute()) + .await + .expect("owner lookup must honor the task deadline"); + + assert!(matches!(result, Err(Error::TaskTimeout))); + assert!(storage.heal_object_calls.lock().expect("heal calls").is_empty()); +} + +#[tokio::test] +async fn object_heal_owner_lookup_obeys_cancellation() { + let storage = Arc::new(MockStorage { + block_bucket_incarnation: true, + ..Default::default() + }); + let mut request = HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None); + request.options.timeout = None; + let task = HealTask::from_request(request, storage.clone()); + + let (result, ()) = tokio::time::timeout(Duration::from_secs(5), async { + tokio::join!(task.execute(), async { + storage.bucket_incarnation_started.notified().await; + task.cancel().await.expect("cancel pending owner lookup"); + }) + }) + .await + .expect("cancellation must interrupt owner lookup"); + + assert!(matches!(result, Err(Error::TaskCancelled))); + assert!(storage.heal_object_calls.lock().expect("heal calls").is_empty()); } #[tokio::test] @@ -1844,6 +1956,14 @@ impl HealStorageAPI for MockStorage { } async fn bucket_incarnation_id(&self, _bucket: &str) -> Result> { + self.bucket_incarnation_calls.fetch_add(1, Ordering::Relaxed); + self.bucket_incarnation_started.notify_one(); + if self.block_bucket_incarnation { + std::future::pending::<()>().await; + } + if let Some(error) = self.bucket_incarnation_error.lock().expect("owner lookup error").take() { + return Err(error); + } if *self.bucket_incarnation_unavailable.lock().unwrap() { return Err(Error::Other("bucket incarnation unavailable".to_string())); } diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs index bf7ada9b2..6ca419342 100644 --- a/crates/heal/tests/mrf_pipeline_test.rs +++ b/crates/heal/tests/mrf_pipeline_test.rs @@ -626,7 +626,8 @@ fn mrf_successor_flush_child_process_fixture() { }), )); mrf_queue::spawn_mrf_consumer(manager.clone()); - let expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2); + let mut expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2); + expected_successor.extend(journal_record(1, "successor-bucket", "first-object", None, 0)); let flushed = wait_until(Duration::from_secs(10), || async { manager.operations_snapshot().await.queued_by_source.mrf == 1 && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor) @@ -673,7 +674,8 @@ fn mrf_successor_flush_waiting_child_process_fixture() { }), )); mrf_queue::spawn_mrf_consumer(manager.clone()); - let expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2); + let mut expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2); + expected_successor.extend(journal_record(1, "service-kill-bucket", "first-object", None, 0)); let flushed = wait_until(Duration::from_secs(10), || async { manager.operations_snapshot().await.queued_by_source.mrf == 1 && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor) @@ -713,7 +715,8 @@ fn mrf_authoritative_fsync_waiting_child_process_fixture() { write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &startup); write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &startup); - let successor = journal_record(1, "fsync-kill-bucket", "second-object", None, 2); + let mut successor = journal_record(1, "fsync-kill-bucket", "second-object", None, 2); + successor.extend(journal_record(1, "fsync-kill-bucket", "first-object", None, 0)); write_journal_path_to_disks_synced(&disk_paths, SCOPED_JOURNAL_REL, &successor); assert!( journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &successor) @@ -770,9 +773,8 @@ async fn journal_replay_retains_child_process_anchor_when_manager_is_full() { ); } -/// If a process crashes after flushing a smaller successor snapshot but before -/// deleting the startup anchor, the restarted process must replay the -/// successor tail rather than losing it or merging it with stale records. +/// A successor flush must preserve both pending work and accepted work whose +/// repair has not been proven when the process restarts. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[serial] async fn journal_replay_survives_successor_flush_before_delete() { @@ -787,7 +789,8 @@ async fn journal_replay_survives_successor_flush_before_delete() { assert_eq!(status.code(), Some(78), "child process did not reach the successor flush boundary"); let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await; - let expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2); + let mut expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2); + expected_successor.extend(journal_record(1, "successor-bucket", "first-object", None, 0)); assert!( journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor), "restarted process must see the pending successor snapshot" @@ -795,11 +798,11 @@ async fn journal_replay_survives_successor_flush_before_delete() { let restarted = make_manager(storage); let replayed = mrf_queue::replay_journal_once(&restarted).await; - assert_eq!(replayed, 1, "restart after successor flush must replay only the still-pending tail"); + assert_eq!(replayed, 2, "restart must replay both the admitted and pending responsibilities"); assert_eq!( restarted.operations_snapshot().await.queued_by_source.mrf, - 1, - "the successor tail must be accepted after restart" + 2, + "both unproven successor responsibilities must be accepted after restart" ); assert!( disk_paths.iter().all(|path| { @@ -811,8 +814,8 @@ async fn journal_replay_survives_successor_flush_before_delete() { } /// A service-style hard kill after successor flush must be equivalent to a -/// crash at the flush-before-delete boundary: restart may replay the smaller -/// successor snapshot, but must not lose or merge stale startup records. +/// crash at the flush-before-delete boundary: restart must recover every +/// unproven responsibility from the successor snapshot. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[serial] #[cfg(unix)] @@ -840,7 +843,8 @@ async fn journal_replay_survives_service_kill_after_successor_flush() { assert!(!status.success(), "child fixture must be terminated instead of exiting cleanly"); let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await; - let expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2); + let mut expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2); + expected_successor.extend(journal_record(1, "service-kill-bucket", "first-object", None, 0)); assert!( journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor), "restarted process must see the successor snapshot produced before the kill" @@ -848,11 +852,11 @@ async fn journal_replay_survives_service_kill_after_successor_flush() { let restarted = make_manager(storage); let replayed = mrf_queue::replay_journal_once(&restarted).await; - assert_eq!(replayed, 1, "restart after service kill must replay only the still-pending tail"); + assert_eq!(replayed, 2, "service-kill restart must preserve every unproven responsibility"); assert_eq!( restarted.operations_snapshot().await.queued_by_source.mrf, - 1, - "the successor tail must be accepted after service kill restart" + 2, + "both unproven responsibilities must be accepted after service kill restart" ); assert!( disk_paths.iter().all(|path| { @@ -864,9 +868,9 @@ async fn journal_replay_survives_service_kill_after_successor_flush() { } /// A hard kill between the authoritative successor fsync and the legacy mirror -/// rewrite must prefer the canonical successor tail over the stale legacy -/// startup epoch. This models the mixed-version boundary conservatively: new -/// readers must not merge epochs, while the old mirror remains crash-visible. +/// rewrite must prefer the canonical successor over the stale legacy startup +/// epoch while retaining every unproven responsibility. New readers must not +/// merge epochs, while the old mirror remains crash-visible. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[serial] #[cfg(unix)] @@ -894,7 +898,8 @@ async fn journal_replay_survives_sigkill_after_authoritative_successor_fsync_bef assert!(!status.success(), "child fixture must be terminated instead of exiting cleanly"); let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await; - let expected_successor = journal_record(1, "fsync-kill-bucket", "second-object", None, 2); + let mut expected_successor = journal_record(1, "fsync-kill-bucket", "second-object", None, 2); + expected_successor.extend(journal_record(1, "fsync-kill-bucket", "first-object", None, 0)); let stale_startup = { let mut startup = journal_record(1, "fsync-kill-bucket", "first-object", None, 0); startup.extend(journal_record(1, "fsync-kill-bucket", "second-object", None, 0)); @@ -911,11 +916,11 @@ async fn journal_replay_survives_sigkill_after_authoritative_successor_fsync_bef let restarted = make_manager(storage); let replayed = mrf_queue::replay_journal_once(&restarted).await; - assert_eq!(replayed, 1, "new reader must replay only the authoritative successor tail"); + assert_eq!(replayed, 2, "new reader must recover every responsibility in the authoritative successor"); assert_eq!( restarted.operations_snapshot().await.queued_by_source.mrf, - 1, - "the successor tail must be accepted after the fsync-boundary restart" + 2, + "both responsibilities must be accepted after the fsync-boundary restart" ); assert!( disk_paths.iter().all(|path| { diff --git a/crates/kms/src/backends/local.rs b/crates/kms/src/backends/local.rs index ac10274fc..002267189 100644 --- a/crates/kms/src/backends/local.rs +++ b/crates/kms/src/backends/local.rs @@ -1233,6 +1233,8 @@ impl LocalKmsClient { async fn decode_stored_key(&self, key_id: &str) -> Result<(StoredMasterKey, Vec)> { let key_path = self.master_key_path(key_id)?; if !fs::try_exists(&key_path).await? { + // A missing key is a caller error only while its storage directory is available. + let _ = fs::read_dir(&self.config.key_dir).await?; return Err(KmsError::key_not_found(key_id)); } @@ -2095,11 +2097,7 @@ impl KmsBackend for LocalKmsBackend { let _write_guard = self.client.lock_key_for_write(key_id).await; // First, load the key from disk to get the master key - let mut master_key = self - .client - .load_master_key(key_id) - .await - .map_err(|_| KmsError::key_not_found(format!("Key {key_id} not found")))?; + let mut master_key = self.client.load_master_key(key_id).await?; let (deletion_date_str, deletion_date_dt) = if request.force_immediate.unwrap_or(false) { // Tombstone first: mark the record Deleted before removing the @@ -2202,11 +2200,7 @@ impl KmsBackend for LocalKmsBackend { let _write_guard = self.client.lock_key_for_write(key_id).await; // Load the key from disk to get the master key - let mut master_key = self - .client - .load_master_key(key_id) - .await - .map_err(|_| KmsError::key_not_found(format!("Key {key_id} not found")))?; + let mut master_key = self.client.load_master_key(key_id).await?; if master_key.status != KeyStatus::PendingDeletion { return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion"))); @@ -2965,6 +2959,146 @@ mod tests { assert!(matches!(error, KmsError::InvalidKey { .. })); } + #[tokio::test] + async fn delete_key_preserves_directory_io_error() { + let (client, temp_dir) = create_dev_mode_client().await; + client.create_key("existing-key", "AES_256", None).await.expect("create key"); + let backend = LocalKmsBackend { client }; + let offline_dir = TempDir::new().expect("create offline directory"); + let offline_key_dir = offline_dir.path().join("keys"); + fs::rename(temp_dir.path(), &offline_key_dir) + .await + .expect("move key directory offline"); + fs::write(temp_dir.path(), b"not a directory") + .await + .expect("replace key directory with a file"); + + let error = backend + .delete_key(DeleteKeyRequest { + key_id: "existing-key".to_string(), + ..Default::default() + }) + .await + .expect_err("unreadable storage must prevent scheduling deletion"); + + fs::remove_file(temp_dir.path()).await.expect("remove replacement file"); + fs::rename(&offline_key_dir, temp_dir.path()) + .await + .expect("restore key directory"); + assert!(matches!(error, KmsError::IoError { .. }), "got {error:?}"); + let key = backend + .client + .load_master_key("existing-key") + .await + .expect("read retained key"); + assert_eq!(key.status, KeyStatus::Active, "failed deletion must not mutate key state"); + } + + #[tokio::test] + async fn cancel_key_deletion_preserves_directory_io_error() { + let (client, temp_dir) = create_dev_mode_client().await; + client.create_key("existing-key", "AES_256", None).await.expect("create key"); + let backend = LocalKmsBackend { client }; + backend + .delete_key(DeleteKeyRequest { + key_id: "existing-key".to_string(), + ..Default::default() + }) + .await + .expect("schedule key deletion"); + let offline_dir = TempDir::new().expect("create offline directory"); + let offline_key_dir = offline_dir.path().join("keys"); + fs::rename(temp_dir.path(), &offline_key_dir) + .await + .expect("move key directory offline"); + fs::write(temp_dir.path(), b"not a directory") + .await + .expect("replace key directory with a file"); + + let error = backend + .cancel_key_deletion(CancelKeyDeletionRequest { + key_id: "existing-key".to_string(), + }) + .await + .expect_err("unreadable storage must prevent cancelling deletion"); + + fs::remove_file(temp_dir.path()).await.expect("remove replacement file"); + fs::rename(&offline_key_dir, temp_dir.path()) + .await + .expect("restore key directory"); + assert!(matches!(error, KmsError::IoError { .. }), "got {error:?}"); + let key = backend + .client + .load_master_key("existing-key") + .await + .expect("read retained key"); + assert_eq!( + key.status, + KeyStatus::PendingDeletion, + "failed cancellation must retain the deletion state" + ); + } + + #[tokio::test] + async fn test_load_master_key_missing_key_remains_not_found() { + let (client, _temp_dir) = create_dev_mode_client().await; + + let error = client + .load_master_key("missing-key") + .await + .expect_err("missing key must fail"); + + assert!(matches!(error, KmsError::KeyNotFound { key_id } if key_id == "missing-key")); + } + + #[tokio::test] + async fn test_load_master_key_unavailable_directory_is_io_error() { + let (client, temp_dir) = create_dev_mode_client().await; + client.create_key("existing-key", "AES_256", None).await.expect("create key"); + let offline_dir = TempDir::new().expect("create offline directory"); + let offline_key_dir = offline_dir.path().join("keys"); + fs::rename(temp_dir.path(), &offline_key_dir) + .await + .expect("move key directory offline"); + + let error = client + .load_master_key("existing-key") + .await + .expect_err("unavailable key directory must fail"); + + fs::rename(&offline_key_dir, temp_dir.path()) + .await + .expect("restore key directory"); + assert!(matches!(error, KmsError::IoError { .. }), "got {error:?}"); + let key = client.load_master_key("existing-key").await.expect("read restored key"); + assert_eq!(key.key_id, "existing-key"); + } + + #[tokio::test] + async fn test_load_master_key_directory_replaced_by_file_is_io_error() { + let (client, temp_dir) = create_dev_mode_client().await; + client.create_key("existing-key", "AES_256", None).await.expect("create key"); + let offline_dir = TempDir::new().expect("create offline directory"); + let offline_key_dir = offline_dir.path().join("keys"); + fs::rename(temp_dir.path(), &offline_key_dir) + .await + .expect("move key directory offline"); + fs::write(temp_dir.path(), b"not a directory") + .await + .expect("replace key directory with a file"); + + let error = client + .load_master_key("existing-key") + .await + .expect_err("a file in place of the key directory must fail"); + + fs::remove_file(temp_dir.path()).await.expect("remove replacement file"); + fs::rename(&offline_key_dir, temp_dir.path()) + .await + .expect("restore key directory"); + assert!(matches!(error, KmsError::IoError { .. }), "got {error:?}"); + } + #[tokio::test] async fn test_load_master_key_accepts_legacy_rfc3339_timestamp() { let (client, _temp_dir) = create_dev_mode_client().await; diff --git a/docs/testing/ci-gates.md b/docs/testing/ci-gates.md index 5c313dc41..a7087d736 100644 --- a/docs/testing/ci-gates.md +++ b/docs/testing/ci-gates.md @@ -43,7 +43,8 @@ Promotion rule: never promote a report-only lane to required from one green run. | PR, non-doc change | `End-to-End Tests` | `ci.yml` `e2e-tests` | Report-only | `cargo nextest run --profile e2e-smoke -p e2e_test`, then `./scripts/e2e-run.sh ./target/debug/rustfs `; membership guards `scripts/check_test_wiring.py --check-profile e2e-smoke ` and `scripts/check_security_smoke_count.sh check ` | | PR, non-doc change | `S3 Implemented Tests` | `ci.yml` `s3-implemented-tests` | Report-only | build `rustfs`, then `scripts/s3-tests/run.sh` with the job's `DEPLOY_MODE` / `TEST_MODE` / `MAXFAIL` env | | PR, non-doc change | `S3 Lifecycle Behavior Tests` | `ci.yml` `s3-lifecycle-behavior-tests` | Report-only | `scripts/s3-tests/run.sh` with the job's accelerated-scanner env | -| PR touching `paths` in `audit.yml` | `Cargo Deny`, `Workflow Pin Report`, `Dependency Review` | `audit.yml` `cargo-deny`, `workflow-pin-report`, `dependency-review` | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` | +| PR to `main` or `release` touching `paths` in `audit.yml` | `Cargo Deny`, `Workflow Pin Report`, `Dependency Review` | `audit.yml` `cargo-deny`, `workflow-pin-report`, `dependency-review` | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` | +| Push to `main` or `release` touching `paths` in `audit.yml` | `Cargo Deny`, `Workflow Pin Report` | `audit.yml` `cargo-deny`, `workflow-pin-report` | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` | | PR touching `paths` in `architecture-migration-rules.yml` | `Architecture Migration Rules` | `architecture-migration-rules.yml` `architecture-migration-rules` | Report-only | `scripts/check_architecture_migration_rules.sh` | | PR touching `paths` in `nix.yml` | `Nix Build & Check` | `nix.yml` `nix-validation` | Report-only | `nix flake check` | | PR touching `paths` in `fuzz.yml` | `Build Fuzz Harness`, `Smoke / ` | `fuzz.yml` `fuzz-build`, `pr-fuzz-smoke` | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` | @@ -53,7 +54,7 @@ Promotion rule: never promote a report-only lane to required from one green run. | PR touching `paths` in `oidc-keycloak.yml` | `OIDC Keycloak live gate` | `oidc-keycloak.yml` `oidc-keycloak-live` | Report-only | `cargo build --locked -p rustfs --bin rustfs`, then `bash scripts/test/oidc_keycloak_live.sh ./target/debug/rustfs` | | PR touching `paths` in `targets-integration.yml` | `PostgreSQL, MySQL, AMQP, and NATS` | `targets-integration.yml` `targets-live` | Report-only | start the containers as in the job, export the `RUSTFS_TEST_*` DSNs, then the job's `cargo test --locked -p rustfs-targets --test -- --ignored --test-threads=1` commands | | PR limited to main-CI-excluded paths | `Quick Checks`, `Test and Lint` | `ci-docs-only.yml` `quick-checks`, `test-and-lint` | Required | `git diff --check`; `make doc-paths-check`; `scripts/check_no_planning_docs.sh` | -| `merge_group`; push to `main` | `End-to-End Tests (full merge gate)` | `ci.yml` `e2e-full` | Report-only | `cargo nextest run --profile e2e-full -p e2e_test` | +| `merge_group`; push to `main` or `release` | `End-to-End Tests (full merge gate)` | `ci.yml` `e2e-full` | Report-only | `cargo nextest run --profile e2e-full -p e2e_test` | e2e filters live in `.config/nextest.toml`; extend a profile instead of adding a second selector. Before a profile runs, `scripts/check_test_wiring.py` compares its listing to the committed digest in `.config/e2e--selection.txt`, so a silent test drop fails closed. @@ -62,7 +63,7 @@ cost. `data_usage_test` runs in the PR `e2e-smoke` lane so changes that affect authoritative scanner usage publication, quota-visible usage, or admin usage snapshots get an end-to-end signal before merge review. `heal_erasure_disk_rebuild_test` runs in `e2e-full` so core erasure heal rebuild regressions are caught no later -than the merge queue or `main` push lane; it also remains in `e2e-nightly` with +than the merge queue or `main`/`release` push lane; it also remains in `e2e-nightly` with the serialized cluster fault-domain suites for scheduled soak signal. ## Scheduled validation diff --git a/scripts/error-other-format-baseline.txt b/scripts/error-other-format-baseline.txt index e1ab9154b..9a903ff77 100644 --- a/scripts/error-other-format-baseline.txt +++ b/scripts/error-other-format-baseline.txt @@ -33,7 +33,7 @@ 1|crates/ecstore/src/disk/mod.rs 5|crates/ecstore/src/erasure/codec/bridge.rs 1|crates/ecstore/src/erasure/coding/decode_reader.rs -10|crates/ecstore/src/erasure/coding/encode.rs +8|crates/ecstore/src/erasure/coding/encode.rs 25|crates/ecstore/src/erasure/coding/erasure.rs 3|crates/ecstore/src/layout/disks_layout.rs 2|crates/ecstore/src/layout/endpoint.rs