diff --git a/Cargo.lock b/Cargo.lock index a9fac1098..a9a0d63a7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10745,6 +10745,7 @@ dependencies = [ "rustfs-data-usage", "rustfs-ecstore", "rustfs-filemeta", + "rustfs-heal", "rustfs-heal-contracts", "rustfs-lifecycle", "rustfs-lock", diff --git a/crates/common/src/mrf_channel.rs b/crates/common/src/mrf_channel.rs index f83b78f96..83e1663ea 100644 --- a/crates/common/src/mrf_channel.rs +++ b/crates/common/src/mrf_channel.rs @@ -422,9 +422,9 @@ fn unix_now_ms() -> u64 { .unwrap_or(0) } -/// A repair the MRF consumer landed, fanned out so retry ledgers can drop -/// entries the journal no longer tracks (backlog#1894 axis B). The payload -/// mirrors the intent identity so consumers match without re-parsing. +/// Legacy, unverified repair notice. Its identity lacks kind, set scope, +/// bucket incarnation and responsibility generation. Consumers must not use +/// it to discharge persisted repair responsibility. #[derive(Debug, Clone, PartialEq, Eq)] pub struct MrfRepairedEvent { pub bucket: Arc, @@ -439,8 +439,8 @@ const MRF_REPAIRED_EVENT_CAP: usize = 4096; static MRF_REPAIRED_EVENTS: OnceLock>> = OnceLock::new(); -/// Record that the MRF consumer landed a repair. Never blocks: the critical -/// section is a deque push under a std mutex. +/// Record a legacy notification for compatibility. This is not an +/// acknowledgement of storage verification or durable repair completion. pub fn note_mrf_repaired(bucket: &str, object: &str, version_id: Option<[u8; 16]>) { let registry = MRF_REPAIRED_EVENTS.get_or_init(|| std::sync::Mutex::new(std::collections::VecDeque::new())); let Ok(mut events) = registry.lock() else { @@ -515,6 +515,9 @@ mod tests { } coalescer_release(&key, Some(lease)); let retry_lease = coalescer_admit(key.clone()).expect("released identity must admit a retry"); + assert_ne!(lease, retry_lease); + coalescer_release(&key, Some(lease)); + assert_eq!(coalescer_admit(key.clone()), Err(MrfIngressResult::Coalesced)); coalescer_release(&key, Some(retry_lease)); } diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index 5247c9ea7..f5e839900 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/crates/heal/src/heal/manager/scheduler.rs @@ -313,7 +313,6 @@ impl HealManager { completed_status_entry.outcome = Some(Arc::new(task.get_outcome().await)); } let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. }); - let successful_completion = matches!(completed_status, HealTaskStatus::Completed); // Keep retry ownership continuous: status snapshots acquire // these locks in the same active -> retrying order. let mut retrying_heals_guard = if let (Some((request, _, error)), Some(cancel_token)) = @@ -375,11 +374,11 @@ impl HealManager { drop(stats); if terminal_completion { let notice_targets = take_mrf_repair_notice_targets(&mrf_repair_notice_targets_clone, &task_id); - if successful_completion { - emit_mrf_repaired_events(notice_targets); - } else { - release_mrf_repair_notice_targets(notice_targets); - } + // Neither task status nor the diagnostic outcome + // window supplies a storage-owned repair receipt. + // Release only the ingress lease for rediscovery; + // preserve the producer's existing retry hints. + release_mrf_repair_notice_targets(notice_targets); } } @@ -706,20 +705,6 @@ fn move_mrf_repair_notice_targets( } } -fn emit_mrf_repaired_events(targets: Vec) { - for target in targets { - rustfs_common::mrf_channel::note_mrf_repaired(&target.bucket, &target.object, target.version_id); - rustfs_common::mrf_channel::release_mrf_identity( - target.kind, - &target.bucket, - &target.object, - target.version_id, - target.scope, - target.lease, - ); - } -} - fn release_mrf_repair_notice_targets(targets: Vec) { for target in targets { rustfs_common::mrf_channel::release_mrf_identity( diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 364f594f0..4a370cebc 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -3096,7 +3096,7 @@ async fn test_cancel_task_removes_queued_request() { } #[tokio::test] -async fn test_mrf_repaired_notice_waits_for_successful_completion() { +async fn mrf_ownership_unverified_completion_does_not_emit_repaired() { let bucket = "mrf-completion-success"; let object = "object"; let version_id = Some([9u8; 16]); @@ -3126,21 +3126,81 @@ async fn test_mrf_repaired_notice_waits_for_successful_completion() { ); process_manager_queue_once(&manager).await; - for _ in 0..100 { - let events = rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket); - if !events.is_empty() { - assert_eq!(events.len(), 1); - assert_eq!(events[0].object.as_ref(), object); - assert_eq!(events[0].version_id, version_id); - return; + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let stats = manager.get_statistics().await; + if stats.successful_tasks + stats.failed_tasks > 0 { + break; + } + tokio::task::yield_now().await; } - tokio::time::sleep(Duration::from_millis(10)).await; - } - panic!("successful MRF-owned heal should emit one repaired event"); + }) + .await + .expect("scheduler completes the task"); + assert!(rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket).is_empty()); + assert!(!lock_mrf_repair_notice_targets(&manager.mrf_repair_notice_targets).contains_key(&receipt.task_id)); } #[tokio::test] -async fn test_mrf_repaired_notice_removed_on_queued_cancel_without_event() { +async fn mrf_ownership_dry_run_and_empty_window_do_not_emit_repaired() { + for empty_window in [false, true] { + let bucket = if empty_window { + "mrf-empty-outcome" + } else { + "mrf-dry-run-outcome" + }; + let manager = HealManager::new(Arc::new(MockStorage), None); + let request = HealRequest::new( + if empty_window { + HealType::Cluster + } else { + HealType::Object { + bucket: bucket.to_string(), + object: "object".to_string(), + version_id: None, + } + }, + HealOptions { + recursive: true, + dry_run: !empty_window, + recreate_missing: true, + ..Default::default() + }, + HealPriority::Normal, + ); + let receipt = manager + .submit_mrf_heal_request_with_receipt(request, Arc::from(bucket), Arc::from("object"), None) + .await + .expect("notice target registered"); + process_manager_queue_once(&manager).await; + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let stats = manager.get_statistics().await; + if stats.successful_tasks + stats.failed_tasks > 0 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("scheduler completed"); + let report = manager.get_task_report(&receipt.task_id).await.expect("completed report"); + assert_eq!(report.status, HealTaskStatus::Completed); + let outcome = report.outcome.expect("canonical outcome"); + if empty_window { + assert!(outcome.objects.is_empty()); + } else { + assert_eq!( + outcome.objects[0].disposition, + crate::heal::outcome::HealObjectDisposition::DryRunObserved + ); + } + assert!(rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket).is_empty()); + } +} + +#[tokio::test] +async fn mrf_ownership_queued_cancel_does_not_emit_repaired() { let bucket = "mrf-completion-cancel"; let object = "object"; let _ = rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket); diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 901064823..8ebb461ff 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -25,12 +25,13 @@ //! set, rewritten on a group-commit cadence (every flush interval or flush //! threshold new intents). A rewrite is atomic at the record level only — a //! torn tail simply truncates during replay because every record carries its -//! own CRC32. Losing the last flush window (≤500 ms) is acceptable because -//! every producer keeps its own safety net: read-repair re-detects on the -//! next failing read, and the scanner's corrupt-metadata branch leaves a -//! pending-ledger entry behind even when its MRF intent is accepted -//! (backlog#1894 axis A), so a lost intent is retried by the ledger rather -//! than waiting for the failed-object TTL to re-scan the path. +//! own CRC32. Neither ingress nor manager admission is a durable ownership +//! receipt. The last flush window can be lost. Read-repair can rediscover a +//! failed read; the scanner retains bounded, expiring retry hints. Partial +//! writes also use a best-effort in-memory fast path, not a durable successor. +//! These mechanisms must not be reported as verified repair completion. +//! The partial-write caller's restart-survival requirement remains unmet by +//! admission alone; a verified durable handoff is still required. use super::{DiskStore, HealDiskExt as _, local_disk_map_read}; use crate::heal::manager::{HealManager, MrfRepairNoticeTarget}; @@ -585,8 +586,8 @@ impl MrfRuntime { self.dirty = true; match submit_mrf_heal_request(manager, &intent).await { // Accepted intents leave the pending set; the next flush persists the - // smaller snapshot. The scanner ledger is cleared later, when the - // canonical heal task reaches a successful terminal completion. + // 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) => {} Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => { intent.attempts = intent.attempts.saturating_add(1); diff --git a/crates/scanner/Cargo.toml b/crates/scanner/Cargo.toml index 7061a5e0f..635fe3423 100644 --- a/crates/scanner/Cargo.toml +++ b/crates/scanner/Cargo.toml @@ -106,6 +106,7 @@ bytes.workspace = true hex-simd.workspace = true [dev-dependencies] +rustfs-heal.workspace = true tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] } serial_test = { workspace = true } temp-env = { workspace = true, features = ["async_closure"] } diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 0c2a8c455..19703c262 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -710,6 +710,10 @@ pub struct FolderScanner { coverage_frontier: Option, resume_frontier: Option, coverage_gap: bool, + pending_heal_sync_deferred: bool, + pending_heal_batch_dirty: bool, + #[cfg(test)] + pending_heal_sync_count: usize, pending_size_reconciliation_keys: HashSet, pending_size_reconciliation_scopes: HashSet, pending_size_reconciliation_truncated: bool, @@ -1080,7 +1084,7 @@ impl FolderScanner { scan_mode, result, ); - if result.is_admitted() { + if result.is_admitted() || matches!(priority, HealChannelPriority::Low) { return Ok(result); } @@ -2435,6 +2439,10 @@ pub async fn scan_data_folder( coverage_frontier: resume_frontier.clone(), resume_frontier, coverage_gap: false, + pending_heal_sync_deferred: false, + pending_heal_batch_dirty: false, + #[cfg(test)] + pending_heal_sync_count: 0, pending_size_reconciliation_keys: HashSet::new(), pending_size_reconciliation_scopes: HashSet::new(), pending_size_reconciliation_truncated: false, diff --git a/crates/scanner/src/scanner_folder/ledger.rs b/crates/scanner/src/scanner_folder/ledger.rs index 167a1425d..29d2cc4b6 100644 --- a/crates/scanner/src/scanner_folder/ledger.rs +++ b/crates/scanner/src/scanner_folder/ledger.rs @@ -11,13 +11,58 @@ // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. -/// The pending-scanner-heal ledger: durable heal intents recorded during scans and retried after MRF consumption. +/// Persisted best-effort retry hints. Existing TTL/count pruning applies only +/// to this hint cache, never to a committed durable repair obligation. use super::*; +const PENDING_HEAL_RETRY_BASE_SECS: u64 = 15 * 60; +const PENDING_HEAL_RETRY_CAP_SECS: u64 = 6 * 60 * 60; + +pub(super) struct PendingHealSyncBatch<'a> { + pub(super) scanner: &'a mut FolderScanner, +} + +impl<'a> PendingHealSyncBatch<'a> { + pub(super) fn new(scanner: &'a mut FolderScanner) -> Self { + scanner.pending_heal_sync_deferred = true; + scanner.pending_heal_batch_dirty = false; + Self { scanner } + } +} + +impl Drop for PendingHealSyncBatch<'_> { + fn drop(&mut self) { + self.scanner.pending_heal_sync_deferred = false; + if self.scanner.pending_heal_batch_dirty { + self.scanner.pending_heal_batch_dirty = false; + self.scanner.sync_pending_heals(); + } + } +} + +pub(super) fn record_pending_heal_attempt(entry: &mut PendingScannerHeal, now: u64) { + entry.last_attempt = now; + entry.attempts = entry.attempts.saturating_add(1); +} + +pub(super) fn observe_pending_heal_admission(entry: &mut PendingScannerHeal, result: HealAdmissionResult) { + // Rediscovery and coalesced admissions must not postpone an armed retry. + entry.last_admission_result = result.result_label().to_string(); + entry.last_admission_reason = result.reason_label().to_string(); +} + impl FolderScanner { pub(super) fn sync_pending_heals(&mut self) { - self.update_cache.info.pending_heals = self.new_cache.info.pending_heals.clone(); self.pending_heals_changed = true; + if self.pending_heal_sync_deferred { + self.pending_heal_batch_dirty = true; + return; + } + self.update_cache.info.pending_heals = self.new_cache.info.pending_heals.clone(); + #[cfg(test)] + { + self.pending_heal_sync_count += 1; + } } pub(super) fn clear_pending_scanner_heal( @@ -37,32 +82,6 @@ impl FolderScanner { } } - /// Batched variant of [`Self::clear_pending_scanner_heal`] for repaired - /// notices (backlog#1894 axis B): one retain pass and one ledger sync - /// for the whole notice set, so a mass-recovery first sweep cannot turn - /// into thousands of full-table clones on the scan task. Only Object - /// entries match — bucket-level heals are never the MRF consumer's work. - pub(super) fn clear_pending_scanner_heals_for_repaired(&mut self, events: &[rustfs_common::mrf_channel::MrfRepairedEvent]) { - // Pre-resolve the notice version strings once; each ledger entry then - // compares against plain Option<&str>. - let targets: Vec<(&str, &str, Option)> = events - .iter() - .map(|event| (event.bucket.as_ref(), event.object.as_ref(), mrf_repaired_version_id(event.version_id))) - .collect(); - let before = self.new_cache.info.pending_heals.len(); - self.new_cache.info.pending_heals.retain(|entry| { - entry.kind != PendingScannerHealKind::Object - || !targets.iter().any(|(bucket, object, version)| { - entry.bucket.as_str() == *bucket - && entry.object.as_deref() == Some(*object) - && entry.version_id.as_deref() == version.as_deref() - }) - }); - if self.new_cache.info.pending_heals.len() != before { - self.sync_pending_heals(); - } - } - pub(super) fn record_pending_scanner_heal( &mut self, kind: PendingScannerHealKind, @@ -80,10 +99,7 @@ impl FolderScanner { .iter_mut() .find(|entry| pending_scanner_heal_matches(entry, kind, bucket, object, version_id)) { - entry.last_attempt = now; - entry.attempts = entry.attempts.saturating_add(1); - entry.last_admission_result = result.result_label().to_string(); - entry.last_admission_reason = result.reason_label().to_string(); + observe_pending_heal_admission(entry, result); self.sync_pending_heals(); return; } @@ -198,47 +214,54 @@ impl FolderScanner { result: HealAdmissionResult, ) { match result { - HealAdmissionResult::Accepted | HealAdmissionResult::Merged => { - self.clear_pending_scanner_heal(kind, bucket, object, version_id); - } HealAdmissionResult::Full | HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull) => { self.record_pending_scanner_heal(kind, bucket, object, version_id, scan_mode, result); } - HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped) => { - self.clear_pending_scanner_heal(kind, bucket, object, version_id); - } - // Admin-only overlap rejections (HS-06); the scanner never sees - // them, but if it ever does, treat them as terminal like any - // other policy drop rather than endlessly retrying. - HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning) + HealAdmissionResult::Accepted + | HealAdmissionResult::Merged + | HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped) + | HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning) | HealAdmissionResult::Dropped(HealAdmissionDropReason::OverlappingPaths) => { - self.clear_pending_scanner_heal(kind, bucket, object, version_id); + // Admission is neither repair completion nor a durable + // successor receipt. Preserve existing responsibility without + // turning every newly admitted hint into a persisted intent. + if let Some(entry) = self + .new_cache + .info + .pending_heals + .iter_mut() + .find(|entry| pending_scanner_heal_matches(entry, kind, bucket, object, version_id)) + { + observe_pending_heal_admission(entry, result); + self.sync_pending_heals(); + } } } } pub(super) async fn retry_pending_scanner_heals(&mut self) -> Result<(), ScannerError> { - if !self.should_heal().await { + let batch = PendingHealSyncBatch::new(self); + let scanner = &mut *batch.scanner; + if !scanner.should_heal().await { return Ok(()); } - let bucket = self.new_cache.info.name.clone(); - // Backlog#1894 axis B: repairs the MRF consumer landed hand the - // manager the heal task, so the matching pending-ledger entries are - // retried nowhere — drop them here. Best-effort: a lost notice just - // leaves the entry to expire through its own attempts/age limits. + let bucket = scanner.new_cache.info.name.clone(); + // Legacy notices cannot bind a verified disposition to the current + // incarnation, kind, set scope and responsibility generation. let repaired = rustfs_common::mrf_channel::take_mrf_repaired_events_for(&bucket); if !repaired.is_empty() { - self.clear_pending_scanner_heals_for_repaired(&repaired); + counter!("rustfs_scanner_unverified_repair_notices_total") + .increment(u64::try_from(repaired.len()).unwrap_or(u64::MAX)); } - self.prune_pending_scanner_heals(); - for pending in pending_scanner_heal_retry_candidates(&self.new_cache.info.pending_heals, &bucket) { - if !self.should_heal().await { + scanner.prune_pending_scanner_heals(); + for pending in pending_scanner_heal_retry_candidates(&scanner.new_cache.info.pending_heals, &bucket) { + if !scanner.should_heal().await { break; } let Some(request) = build_pending_scanner_heal_request(&pending) else { - self.clear_pending_scanner_heal(pending.kind, &pending.bucket, None, pending.version_id.as_deref()); + scanner.clear_pending_scanner_heal(pending.kind, &pending.bucket, None, pending.version_id.as_deref()); counter!( METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL, "bucket" => pending.bucket.clone(), @@ -257,14 +280,27 @@ impl FolderScanner { continue; }; - self.send_required_scanner_heal_request( - pending.kind, - pending.bucket.clone(), - pending.object.clone(), - pending.version_id.clone(), - request, - ) - .await?; + if let Some(entry) = scanner.new_cache.info.pending_heals.iter_mut().find(|entry| { + pending_scanner_heal_matches( + entry, + pending.kind, + &pending.bucket, + pending.object.as_deref(), + pending.version_id.as_deref(), + ) + }) { + record_pending_heal_attempt(entry, Self::now_secs()); + scanner.sync_pending_heals(); + } + scanner + .send_required_scanner_heal_request( + pending.kind, + pending.bucket.clone(), + pending.object.clone(), + pending.version_id.clone(), + request, + ) + .await?; } Ok(()) @@ -295,16 +331,6 @@ pub(super) fn pending_scanner_heal_identity(entry: &PendingScannerHeal) -> (u8, (kind, entry.bucket.as_str(), entry.object.as_deref(), entry.version_id.as_deref()) } -/// Decode an MRF repaired-notice version id for ledger matching. A nil UUID -/// means "no value" per the repo-wide defensive-UUID invariant, so it maps -/// to `None` and matches unversioned ledger entries only. -pub(super) fn mrf_repaired_version_id(version_id: Option<[u8; 16]>) -> Option { - version_id - .map(uuid::Uuid::from_bytes) - .filter(|uuid| !uuid.is_nil()) - .map(|uuid| uuid.to_string()) -} - pub(super) fn sort_pending_scanner_heals_for_retry(entries: &mut [PendingScannerHeal]) { entries.sort_by(|a, b| { a.last_attempt @@ -318,30 +344,56 @@ pub(super) fn pending_scanner_heal_retry_candidates( pending_heals: &[PendingScannerHeal], bucket: &str, ) -> Vec { - let mut entries: Vec = pending_heals.iter().filter(|entry| entry.bucket == bucket).cloned().collect(); - sort_pending_scanner_heals_for_retry(&mut entries); + pending_scanner_heal_retry_candidates_at(pending_heals, bucket, FolderScanner::now_secs()) +} + +pub(super) fn pending_scanner_heal_retry_candidates_at( + pending_heals: &[PendingScannerHeal], + bucket: &str, + now: u64, +) -> Vec { + // Schedule across scanner cycles rather than allocating a timer per hint. + // A later Full response must not reset an already retried hint's backoff. + let mut entries: Vec<&PendingScannerHeal> = pending_heals + .iter() + .filter(|entry| { + let exponent = entry.attempts.saturating_sub(1).min(31); + let delay = PENDING_HEAL_RETRY_BASE_SECS + .saturating_mul(1_u64 << exponent) + .min(PENDING_HEAL_RETRY_CAP_SECS); + entry.bucket == bucket && now.checked_sub(entry.last_attempt).is_some_and(|age| age >= delay) + }) + .collect(); + entries.sort_by(|a, b| { + a.last_attempt + .cmp(&b.last_attempt) + .then_with(|| a.attempts.cmp(&b.attempts)) + .then_with(|| pending_scanner_heal_identity(a).cmp(&pending_scanner_heal_identity(b))) + }); entries.truncate(MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET); - entries + entries.into_iter().cloned().collect() } pub(super) fn build_pending_scanner_heal_request(entry: &PendingScannerHeal) -> Option { + let priority = if entry.last_admission_result == "full" + || (entry.last_admission_result == "dropped" && entry.last_admission_reason == "queue_full") + { + HealChannelPriority::High + } else { + HealChannelPriority::Low + }; match entry.kind { - PendingScannerHealKind::Bucket => Some(build_bucket_heal_request(entry.bucket.clone(), HealChannelPriority::High)), + PendingScannerHealKind::Bucket => Some(build_bucket_heal_request(entry.bucket.clone(), priority)), PendingScannerHealKind::Object => entry.object.as_ref().map(|object| { if entry.version_id.is_none() { - build_non_destructive_object_heal_request( - entry.bucket.clone(), - object.clone(), - entry.scan_mode, - HealChannelPriority::High, - ) + build_non_destructive_object_heal_request(entry.bucket.clone(), object.clone(), entry.scan_mode, priority) } else { build_object_heal_request( entry.bucket.clone(), object.clone(), entry.version_id.clone(), entry.scan_mode, - HealChannelPriority::High, + priority, ) } }), diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index d47b17acc..723c79207 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/crates/scanner/src/scanner_folder/tests.rs @@ -15,6 +15,8 @@ use crate::SCANNER_SLEEPER; use super::*; + +mod mrf_ownership; use crate::storage_api::VersionPurgeStatusType; use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass}; use rustfs_filemeta::{FileInfo, FileMeta, MetadataResolutionParams}; @@ -350,6 +352,9 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) { coverage_frontier: None, resume_frontier: None, coverage_gap: false, + pending_heal_sync_deferred: false, + pending_heal_batch_dirty: false, + pending_heal_sync_count: 0, pending_size_reconciliation_keys: HashSet::new(), pending_size_reconciliation_scopes: HashSet::new(), pending_size_reconciliation_truncated: false, @@ -1132,23 +1137,8 @@ fn pending_heal( } } -/// The nil-UUID branch of the defensive-UUID invariant: a nil version in -/// a repaired notice means "no value" and must match unversioned ledger -/// entries only. -#[test] -fn test_mrf_repaired_version_id_maps_nil_to_none() { - assert_eq!(mrf_repaired_version_id(None), None); - assert_eq!(mrf_repaired_version_id(Some([0u8; 16])), None); - let uuid = Uuid::new_v4(); - assert_eq!(mrf_repaired_version_id(Some(*uuid.as_bytes())), Some(uuid.to_string())); -} - -/// Full wiring of backlog#1894 axis B: notes taken for the scanned bucket -/// clear exactly the matching Object ledger entries — bucket-level -/// entries, other buckets' entries, and version-mismatched entries -/// survive; a real (non-nil) version matches only the same version. #[tokio::test] -async fn test_mrf_repaired_notices_clear_matching_ledger_entries() { +async fn mrf_ownership_legacy_notices_preserve_pending_entries() { use rustfs_common::mrf_channel::note_mrf_repaired; let (mut scanner, temp_dir) = build_test_scanner().await; @@ -1176,8 +1166,7 @@ async fn test_mrf_repaired_notices_clear_matching_ledger_entries() { note_mrf_repaired("bucket", "object-a", None); note_mrf_repaired("bucket", "object-b", Some(*Uuid::parse_str(&version).unwrap().as_bytes())); - // A nil-UUID notice for object-c means "no value": it clears the - // unversioned entry but must not touch the versioned one. + // Neither nil nor a matching version proves incarnation, scope or owner. note_mrf_repaired("bucket", "object-c", Some([0u8; 16])); // A notice for a target the ledger does not track must be a no-op. note_mrf_repaired("bucket", "object-untracked", None); @@ -1194,12 +1183,12 @@ async fn test_mrf_repaired_notices_clear_matching_ledger_entries() { .iter() .map(|entry| (entry.kind, entry.bucket.as_str(), entry.object.as_deref(), entry.version_id.as_deref())) .collect(); - // Cleared: object-a (no version), object-b (exact version match), and - // object-c's unversioned entry (the nil branch matched no-version - // only — the versioned object-c entry survives). assert_eq!( survivors, vec![ + (PendingScannerHealKind::Object, "bucket", Some("object-a"), None), + (PendingScannerHealKind::Object, "bucket", Some("object-b"), Some(version.as_str())), + (PendingScannerHealKind::Object, "bucket", Some("object-c"), None), ( PendingScannerHealKind::Object, "bucket", @@ -1338,7 +1327,7 @@ async fn test_pending_heal_update_keeps_stale_entry_until_retry_prune() { ); assert_eq!(scanner.new_cache.info.pending_heals.len(), 1); - assert_eq!(scanner.new_cache.info.pending_heals[0].attempts, 2); + assert_eq!(scanner.new_cache.info.pending_heals[0].attempts, 1); assert_eq!(scanner.new_cache.info.pending_heals[0].object.as_deref(), Some("object")); assert_eq!(scanner.update_cache.info.pending_heals, scanner.new_cache.info.pending_heals); } @@ -1371,7 +1360,7 @@ async fn test_pending_heal_queue_full_deduplicates_object_entry() { let pending = &scanner.new_cache.info.pending_heals[0]; assert_eq!(pending.object.as_deref(), Some("object")); assert_eq!(pending.version_id.as_deref(), Some("version-a")); - assert_eq!(pending.attempts, 2); + assert_eq!(pending.attempts, 1); assert_eq!(pending.last_admission_result, "dropped"); assert_eq!(pending.last_admission_reason, "queue_full"); assert_eq!(scanner.update_cache.info.pending_heals, scanner.new_cache.info.pending_heals); @@ -1379,7 +1368,7 @@ async fn test_pending_heal_queue_full_deduplicates_object_entry() { } #[tokio::test] -async fn test_pending_heal_admitted_results_clear_matching_entry() { +async fn mrf_ownership_admission_preserves_existing_pending() { let (mut scanner, temp_dir) = build_test_scanner().await; let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir); @@ -1400,7 +1389,8 @@ async fn test_pending_heal_admitted_results_clear_matching_entry() { HealAdmissionResult::Accepted, ); - assert!(scanner.new_cache.info.pending_heals.is_empty()); + assert_eq!(scanner.new_cache.info.pending_heals.len(), 1); + assert_eq!(scanner.new_cache.info.pending_heals[0].last_admission_result, "accepted"); scanner.update_pending_scanner_heal_after_admission( PendingScannerHealKind::Bucket, @@ -1419,11 +1409,12 @@ async fn test_pending_heal_admitted_results_clear_matching_entry() { HealAdmissionResult::Merged, ); - assert!(scanner.new_cache.info.pending_heals.is_empty()); + assert_eq!(scanner.new_cache.info.pending_heals.len(), 2); + assert_eq!(scanner.new_cache.info.pending_heals[1].last_admission_result, "merged"); } #[tokio::test] -async fn test_pending_heal_policy_dropped_clears_without_creating_entry() { +async fn mrf_ownership_policy_drop_does_not_discharge_existing_pending() { let (mut scanner, temp_dir) = build_test_scanner().await; let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir); @@ -1454,7 +1445,7 @@ async fn test_pending_heal_policy_dropped_clears_without_creating_entry() { HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped), ); - assert!(scanner.new_cache.info.pending_heals.is_empty()); + assert_eq!(scanner.new_cache.info.pending_heals.len(), 1); } #[test] diff --git a/crates/scanner/src/scanner_folder/tests/mrf_ownership.rs b/crates/scanner/src/scanner_folder/tests/mrf_ownership.rs new file mode 100644 index 000000000..142008cef --- /dev/null +++ b/crates/scanner/src/scanner_folder/tests/mrf_ownership.rs @@ -0,0 +1,410 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::*; +use crate::storage_api::EcstoreHealResultItem as HealItem; +use crate::storage_api::scanner_io::BucketInfo; +use rustfs_common::mrf_channel::{ + MrfIngressResult, MrfKind, MrfScope, note_mrf_repaired, take_mrf_repaired_events_for, try_send_mrf_intent_typed, +}; +use rustfs_heal::heal::{ + manager::{HealConfig, HealManager}, + mrf_queue::spawn_mrf_consumer, + storage::{HealListItem, HealObjectInfo, HealStorageAPI}, +}; +use rustfs_heal_contracts::heal_channel::HealOpts; + +#[tokio::test] +async fn mrf_ownership_admission_observation_does_not_postpone_retry() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir); + scanner.new_cache.info.pending_heals.push(pending_heal( + PendingScannerHealKind::Object, + "bucket", + Some("object"), + None, + 100, + 2, + )); + for result in [ + HealAdmissionResult::Accepted, + HealAdmissionResult::Merged, + HealAdmissionResult::Full, + HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull), + HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped), + ] { + scanner.update_pending_scanner_heal_after_admission( + PendingScannerHealKind::Object, + "bucket", + Some("object"), + None, + HealScanMode::Deep, + result, + ); + let entry = &scanner.new_cache.info.pending_heals[0]; + assert_eq!((entry.last_attempt, entry.attempts), (100, 2)); + assert!(pending_scanner_heal_retry_candidates_at(&scanner.new_cache.info.pending_heals, "bucket", 1899).is_empty()); + assert_eq!( + pending_scanner_heal_retry_candidates_at(&scanner.new_cache.info.pending_heals, "bucket", 1900).len(), + 1 + ); + } + scanner.update_pending_scanner_heal_after_admission( + PendingScannerHealKind::Object, + "bucket", + Some("new-object"), + None, + HealScanMode::Deep, + HealAdmissionResult::Accepted, + ); + assert_eq!( + scanner.new_cache.info.pending_heals.len(), + 1, + "successful admission does not create a new ledger" + ); +} + +#[test] +fn mrf_ownership_retry_due_boundaries_and_priority_are_bounded() { + let mut entry = pending_heal(PendingScannerHealKind::Object, "bucket", Some("object"), None, 100, 1); + entry.last_admission_result = "accepted".to_string(); + assert!(pending_scanner_heal_retry_candidates_at(std::slice::from_ref(&entry), "bucket", 999).is_empty()); + assert_eq!( + pending_scanner_heal_retry_candidates_at(std::slice::from_ref(&entry), "bucket", 1000).len(), + 1 + ); + assert_eq!( + build_pending_scanner_heal_request(&entry).expect("request").priority, + HealChannelPriority::Low + ); + record_pending_heal_attempt(&mut entry, 1000); + observe_pending_heal_admission(&mut entry, HealAdmissionResult::Full); + assert!(pending_scanner_heal_retry_candidates_at(std::slice::from_ref(&entry), "bucket", 2799).is_empty()); + assert_eq!( + pending_scanner_heal_retry_candidates_at(std::slice::from_ref(&entry), "bucket", 2800).len(), + 1 + ); + assert_eq!( + build_pending_scanner_heal_request(&entry).expect("request").priority, + HealChannelPriority::High + ); + entry.attempts = u32::MAX; + assert!(pending_scanner_heal_retry_candidates_at(std::slice::from_ref(&entry), "bucket", 22599).is_empty()); + assert_eq!( + pending_scanner_heal_retry_candidates_at(std::slice::from_ref(&entry), "bucket", 22600).len(), + 1 + ); + entry.last_attempt = u64::MAX; + assert!(pending_scanner_heal_retry_candidates_at(&[entry], "bucket", 22600).is_empty()); +} + +#[tokio::test] +async fn mrf_ownership_full_hint_table_has_bounded_multicycle_work_and_sync() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir); + let base = 1_700_000_000; + scanner.new_cache.info.pending_heals = (0..MAX_PENDING_SCANNER_HEALS_PER_BUCKET) + .map(|index| { + let mut entry = + pending_heal(PendingScannerHealKind::Object, "bucket", Some(&format!("object-{index}")), None, base, 1); + entry.first_seen = base; + entry.last_admission_result = "accepted".to_string(); + entry + }) + .collect(); + let indices: HashMap = scanner + .new_cache + .info + .pending_heals + .iter() + .enumerate() + .map(|(index, entry)| (entry.object.clone().expect("object identity"), index)) + .collect(); + scanner.sync_pending_heals(); + let initial_syncs = scanner.pending_heal_sync_count; + let mut requests = 0usize; + let mut nonempty_batches = 0; + for minute in 0..24 * 60 { + let now = base + minute * 60; + let candidates = pending_scanner_heal_retry_candidates_at(&scanner.new_cache.info.pending_heals, "bucket", now); + assert!(candidates.len() <= MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET); + if candidates.is_empty() { + continue; + } + nonempty_batches += 1; + let before = scanner.pending_heal_sync_count; + { + let batch = PendingHealSyncBatch::new(&mut scanner); + for candidate in candidates { + assert_eq!( + build_pending_scanner_heal_request(&candidate) + .expect("retry request") + .priority, + HealChannelPriority::Low + ); + let index = indices[candidate.object.as_ref().expect("object identity")]; + record_pending_heal_attempt(&mut batch.scanner.new_cache.info.pending_heals[index], now); + observe_pending_heal_admission( + &mut batch.scanner.new_cache.info.pending_heals[index], + HealAdmissionResult::Accepted, + ); + batch.scanner.sync_pending_heals(); + requests += 1; + } + } + assert_eq!(scanner.pending_heal_sync_count, before + 1, "one table clone per changed retry batch"); + assert_eq!(scanner.new_cache.info.pending_heals.len(), MAX_PENDING_SCANNER_HEALS_PER_BUCKET); + } + assert!(requests >= MAX_PENDING_SCANNER_HEALS_PER_BUCKET, "every retained hint receives a retry"); + assert!( + requests <= 7 * MAX_PENDING_SCANNER_HEALS_PER_BUCKET, + "15min..6h backoff bounds repeated work within 24h" + ); + assert!(scanner.new_cache.info.pending_heals.iter().all(|entry| entry.attempts >= 2)); + assert_eq!(scanner.pending_heal_sync_count - initial_syncs, nonempty_batches); + assert_eq!(scanner.update_cache.info.pending_heals, scanner.new_cache.info.pending_heals); +} + +#[tokio::test] +async fn mrf_ownership_cancelled_batch_restores_sync_without_per_item_clones() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir); + scanner + .new_cache + .info + .pending_heals + .push(pending_heal(PendingScannerHealKind::Object, "bucket", Some("object"), None, 1, 1)); + let before = scanner.pending_heal_sync_count; + let mut work = Box::pin(async { + let batch = PendingHealSyncBatch::new(&mut scanner); + record_pending_heal_attempt(&mut batch.scanner.new_cache.info.pending_heals[0], 100); + batch.scanner.sync_pending_heals(); + batch.scanner.sync_pending_heals(); + std::future::pending::<()>().await; + }); + assert!(futures::poll!(&mut work).is_pending()); + drop(work); + assert!(!scanner.pending_heal_sync_deferred); + assert!(!scanner.pending_heal_batch_dirty); + assert_eq!(scanner.pending_heal_sync_count, before + 1); + assert_eq!(scanner.new_cache.info.pending_heals[0].attempts, 2); + assert!(pending_scanner_heal_retry_candidates_at(&scanner.new_cache.info.pending_heals, "bucket", 101).is_empty()); + assert_eq!(scanner.update_cache.info.pending_heals, scanner.new_cache.info.pending_heals); + let result: std::result::Result<(), &'static str> = async { + let batch = PendingHealSyncBatch::new(&mut scanner); + record_pending_heal_attempt(&mut batch.scanner.new_cache.info.pending_heals[0], 200); + observe_pending_heal_admission(&mut batch.scanner.new_cache.info.pending_heals[0], HealAdmissionResult::Merged); + batch.scanner.sync_pending_heals(); + Err("injected retry batch failure") + } + .await; + assert!(result.is_err()); + assert_eq!(scanner.pending_heal_sync_count, before + 2); + assert!(!scanner.pending_heal_sync_deferred); + assert_eq!(scanner.update_cache.info.pending_heals, scanner.new_cache.info.pending_heals); +} + +#[derive(Default)] +struct NoticeStorage { + calls: std::sync::Mutex>, + retry_started: tokio::sync::Notify, +} + +#[async_trait::async_trait] +impl HealStorageAPI for NoticeStorage { + async fn get_object_meta(&self, _: &str, _: &str) -> rustfs_heal::Result> { + Ok(None) + } + async fn ec_decode_rebuild(&self, _: &str, _: &str) -> rustfs_heal::Result> { + Err(rustfs_heal::Error::other("unused decode fixture")) + } + async fn get_bucket_info(&self, bucket: &str) -> rustfs_heal::Result> { + Ok(Some(BucketInfo { + name: bucket.to_string(), + ..Default::default() + })) + } + async fn list_buckets(&self) -> rustfs_heal::Result> { + Ok(Vec::new()) + } + async fn object_exists(&self, _: &str, _: &str) -> rustfs_heal::Result { + Ok(true) + } + async fn heal_object( + &self, + _: &str, + object: &str, + _: Option<&str>, + _: &HealOpts, + ) -> rustfs_heal::Result<(HealItem, Option)> { + let retry = { + let mut calls = self.calls.lock().expect("fixture calls"); + let count = calls.entry(object.to_string()).or_default(); + *count += 1; + *count > 1 + }; + if retry { + self.retry_started.notify_one(); + std::future::pending::<()>().await; + } + match object { + "grace" => Ok(( + HealItem::default(), + Some(rustfs_heal::Error::Disk(crate::DiskError::other( + "dangling object deletion deferred by heal grace window; retry_after_secs=3599; grace_secs=3600", + ))), + )), + "failed" => Err(rustfs_heal::Error::other("permanent fixture failure")), + "cancelled" => Err(rustfs_heal::Error::TaskCancelled), + _ => Ok((HealItem::default(), None)), + } + } + async fn heal_bucket(&self, _: &str, _: &HealOpts) -> rustfs_heal::Result { + Ok(HealItem::default()) + } + async fn heal_format(&self, _: bool) -> rustfs_heal::Result<(HealItem, Option)> { + Ok((HealItem::default(), None)) + } + async fn list_objects_for_heal_page( + &self, + _: &str, + _: &str, + _: Option<&str>, + _: bool, + ) -> rustfs_heal::Result<(Vec, Option, bool)> { + Ok((Vec::new(), None, false)) + } + async fn get_disk_for_resume(&self, _: &str) -> rustfs_heal::Result { + Err(rustfs_heal::Error::other("unused resume fixture")) + } +} + +#[tokio::test] +#[serial] +async fn mrf_ownership_manager_completion_preserves_scanner_pending() { + const CHILD: &str = "RUSTFS_MRF_OWNERSHIP_TEST_CHILD"; + if std::env::var_os(CHILD).is_none() { + let output = std::process::Command::new(std::env::current_exe().expect("test executable")) + .args([ + "--exact", + "scanner_folder::tests::mrf_ownership::mrf_ownership_manager_completion_preserves_scanner_pending", + "--nocapture", + ]) + .env(CHILD, "1") + .env("RUSTFS_HEAL_MRF_ENABLE", "true") + .output() + .expect("isolated ingress test process"); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + output.status.success() && stdout.contains("1 passed;"), + "{stdout}\n{}", + String::from_utf8_lossy(&output.stderr) + ); + return; + } + // The production ingress channel is a process singleton; isolation keeps + // its receiver and lease generations independent from other scanner tests. + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir); + let bucket = format!("mrf-ownership-{}", Uuid::new_v4()); + scanner.new_cache.info.name = bucket.clone(); + scanner.update_cache.info.name = bucket.clone(); + scanner.heal_object_select = 1; + let storage = Arc::new(NoticeStorage::default()); + let manager = Arc::new(HealManager::new( + storage.clone(), + Some(HealConfig { + enable_auto_heal: false, + mainline_throttle_enable: false, + heal_interval: Duration::from_millis(10), + ..Default::default() + }), + )); + manager.start().await.expect("production manager starts"); + spawn_mrf_consumer(manager.clone()); + for (index, object) in ["grace", "unknown", "failed", "cancelled"].iter().enumerate() { + let version = Uuid::new_v4(); + scanner.new_cache.info.pending_heals.push(pending_heal( + PendingScannerHealKind::Object, + &bucket, + Some(object), + Some(&version.to_string()), + 1, + 1, + )); + let scope = Some(MrfScope { + pool_index: 0, + set_index: 0, + }); + assert_eq!( + try_send_mrf_intent_typed(MrfKind::PartialWrite, &bucket, object, Some(version), scope), + MrfIngressResult::Enqueued + ); + // Re-admission establishes that the first terminal callback released + // its ingress lease. Statistics alone precede notice publication. + tokio::time::timeout(Duration::from_secs(5), async { + loop { + match try_send_mrf_intent_typed(MrfKind::PartialWrite, &bucket, object, Some(version), scope) { + MrfIngressResult::Enqueued => break, + MrfIngressResult::Coalesced => tokio::task::yield_now().await, + other => panic!("unexpected retry ingress result: {other:?}"), + } + } + }) + .await + .expect("production terminal releases its ingress lease"); + tokio::time::timeout(Duration::from_secs(5), storage.retry_started.notified()) + .await + .expect("the real consumer starts the second generation"); + assert!( + take_mrf_repaired_events_for(&bucket).is_empty(), + "{object}: task completion must not emit an unproved repair" + ); + if *object == "unknown" { + assert_eq!( + manager.get_statistics().await.total_objects_healed, + 1, + "legacy healed count is not repair proof" + ); + } + assert_eq!( + try_send_mrf_intent_typed(MrfKind::PartialWrite, &bucket, object, Some(version), scope), + MrfIngressResult::Coalesced, + "the in-flight retry retains its new ingress lease" + ); + note_mrf_repaired(&bucket, object, Some(*version.as_bytes())); + let syncs_before_retry = scanner.pending_heal_sync_count; + scanner + .retry_pending_scanner_heals() + .await + .expect("real scanner ledger retry"); + assert_eq!(scanner.pending_heal_sync_count, syncs_before_retry + 1, "the real retry batch syncs once"); + assert_eq!( + scanner.new_cache.info.pending_heals.len(), + index + 1, + "{object}: pending responsibility survives" + ); + let restored = DataUsageCache::unmarshal(&scanner.new_cache.marshal_msg().expect("serialize pending cache")) + .expect("decode pending cache"); + assert_eq!(restored.info.pending_heals.len(), index + 1); + assert_eq!( + manager + .cancel_tasks_for_path(&format!("{bucket}/{object}")) + .await + .expect("cancel blocked retry"), + 1 + ); + } + manager.stop().await.expect("production manager stops"); +} diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index a1981df30..26a1eb935 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -126,6 +126,9 @@ pub(crate) use rustfs_lifecycle::{ }; use rustfs_storage_api as storage_contracts; +#[cfg(test)] +pub(crate) type EcstoreHealResultItem = ::HealResultItem; + pub(crate) mod owner { #[cfg(test)] pub(crate) use rustfs_ecstore::api::set_disk::test_util::hold_namespace_commit as ecstore_hold_namespace_commit; diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index 55e40cc31..c931f3302 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -301,6 +301,24 @@ The pacing gate holds neither namespace locks nor I/O/page permits while sleepin A missing provider, zero pause, or both class thresholds set to zero preserves unpaced execution. Missing counts follow the existing shared pressure interpreter; they are observations, not health, quorum or resource-ownership proof. The current provider exposes node-level workload classes, so this does not claim independent per-set foreground measurements or a hard global resource budget. Runtime waits increment `rustfs_heal_mainline_throttle_total` with `source=admin`, `result=delayed`, and a foreground-pressure or `recovery_window` reason. Real p99/throughput protection requires the separate W20 fixed-load ABBA measurements. +## Pending Heal Hints + +The scanner's persisted pending-heal cache is a best-effort retry backstop, bounded to 10,000 hints per bucket and a 24-hour age limit. These limits do not authorize garbage collection of committed durable repair obligations. A durable owner must retain its independent replay record until verified object completion or an equivalent durable successor permits removal. + +Accepted, merged, or policy-dropped admission does not clear an existing hint. Task completion and legacy repair notices also lack the incarnation, set scope, generation, and storage verification needed to prove repair responsibility was discharged. The legacy success path therefore retains hints conservatively; this is not a complete durable MRF handoff protocol. + +Pending retries use their persisted attempt count and last-attempt timestamp, starting at 15 minutes and doubling up to six hours. Each bucket submits at most 128 due hints per cycle. Already admitted or policy-dropped hints retry at Low priority; queue-full hints keep High priority but obey the same due-time bound. A changed retry batch synchronizes its pending cache once, including when cancelled, rather than copying the entire table after every admission. + +Rediscovery and admission observations update the recorded result but do not postpone an already armed retry. The actual retry loop advances the attempt count and timestamp before awaiting admission, so cancellation or repeated queue-full results cannot reset the retry budget. + +| Producer | Current identity and compensation | Durable handoff boundary | +|---|---|---| +| Scanner corrupt metadata | Metadata kind with no invented version/set; an existing pending-cache hint remains available for bounded retries. | Cache publication is separate from MRF ingress. Its age/count limits mean it is not an irrevocable repair-obligation ledger. | +| Read decode failure | Decode kind, available version and erasure-set scope; a later failing read can rediscover the repair. | Nonblocking ingress and in-memory read-repair admission do not acknowledge durable acceptance. | +| Partial write | Partial-write kind, available version and erasure-set scope; an in-memory heal request is the fast path. | The caller's documented restart-survival requirement is not fulfilled by ignoring the ingress result or by removing the unaccepted journal record at manager admission. Verified durable ownership remains pending. | + +Legacy notices carry only bucket/object/version, not a verified storage disposition, incarnation, scope, or durable responsibility generation. They are drained without clearing hints. Terminal callbacks release only their exact node-local ingress lease so rediscovery remains possible; lease generations are not durable successor receipts. Pending migration staging is not activated, and this change does not enable durable tombstones or garbage collection. Positive cleanup requires a storage-owner receipt with the complete responsibility identity and validated commit/fence evidence; neither task status nor the bounded diagnostic outcome window supplies it. + ## Deliberate non-parity with MinIO These differences from MinIO are design decisions, recorded so they are not re-filed as gaps.