From 97e600ae697c82287bd61b7f60bd0b40888beaa8 Mon Sep 17 00:00:00 2001 From: Chris Date: Thu, 22 Oct 2026 00:04:16 +0900 Subject: [PATCH] fix(heal): hold unobserved legacy MRF responsibilities (#8772) --- crates/heal/src/heal/mrf_queue.rs | 17 ++- .../heal/src/heal/mrf_queue/partial_write.rs | 130 +++++++++++++----- crates/heal/tests/mrf_partial_write_test.rs | 49 ++++++- crates/heal/tests/mrf_pipeline_test.rs | 11 +- docs/operations/scanner-runtime-controls.md | 4 +- rustfs/src/admin/handlers/heal.rs | 37 ++++- 6 files changed, 192 insertions(+), 56 deletions(-) diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 2e98b9865..185446cd2 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -139,7 +139,7 @@ pub async fn recheck_legacy_responsibility( pub async fn refresh_legacy_responsibility( responsibility_id: Uuid, - expected_bucket_incarnation_id: Uuid, + expected_bucket_incarnation_id: Option, ) -> Result<(), MrfLifecycleControlError> { let sender = GLOBAL_MRF_LIFECYCLE_SENDER .get() @@ -181,7 +181,7 @@ enum MrfLifecycleCommand { }, Refresh { responsibility_id: Uuid, - expected_bucket_incarnation_id: Uuid, + expected_bucket_incarnation_id: Option, response: tokio::sync::oneshot::Sender>, }, AcceptUnverifiedRisk { @@ -1012,7 +1012,7 @@ impl MrfRuntime { fn list_unverified_legacy(&self, after: Option, limit: usize) -> MrfLegacyResponsibilitySnapshot { let (responsibilities, next_cursor) = self.partial_writes.list_unverified_legacy(after, limit); MrfLegacyResponsibilitySnapshot { - contract_version: 1, + contract_version: 2, node_local: true, checkpoint_owner: self.runtime_checkpoint.map(|(owner, _)| owner), checkpoint_sequence: self.runtime_checkpoint.map(|(_, sequence)| sequence), @@ -1085,20 +1085,25 @@ impl MrfRuntime { &mut self, manager: &HealManager, responsibility_id: Uuid, - expected_bucket_incarnation_id: Uuid, + expected_bucket_incarnation_id: Option, ) -> Result<(), MrfLifecycleControlError> { let (intent, state, source_incarnation) = self .partial_writes .intent_for_responsibility(responsibility_id) .ok_or(MrfLifecycleControlError::StaleGeneration)?; - if state.bucket_incarnation_id() != Some(expected_bucket_incarnation_id) { + if state.bucket_incarnation_id() != expected_bucket_incarnation_id { + return Err(MrfLifecycleControlError::StaleGeneration); + } + if expected_bucket_incarnation_id.is_none() + && (source_incarnation.is_some() || !matches!(state, partial_write::ResponsibilityState::Active)) + { return Err(MrfLifecycleControlError::StaleGeneration); } let anchor = manager .durable_mrf_repair_anchor(&intent) .await .ok_or(MrfLifecycleControlError::IncarnationChanged)?; - if anchor.bucket_incarnation_id == expected_bucket_incarnation_id { + if Some(anchor.bucket_incarnation_id) == expected_bucket_incarnation_id { return Ok(()); } let previous_acceptance = self.partial_writes.last_operator_acceptance(responsibility_id); diff --git a/crates/heal/src/heal/mrf_queue/partial_write.rs b/crates/heal/src/heal/mrf_queue/partial_write.rs index 2a5a6e409..f39a814b4 100644 --- a/crates/heal/src/heal/mrf_queue/partial_write.rs +++ b/crates/heal/src/heal/mrf_queue/partial_write.rs @@ -210,6 +210,17 @@ struct Responsibility { } impl Responsibility { + fn generation_unobserved(&self) -> bool { + self.source_bucket_incarnation_id.is_none() && matches!(self.state, ResponsibilityState::Active) + } + + fn is_parked(&self) -> bool { + // Legacy source identity cannot be reconstructed from a bucket name. + // Derive this hold from the existing durable fields so old checkpoints + // need no new state tag and replay cannot restart automatic retries. + self.state.is_parked() || self.generation_unobserved() + } + fn defer_unverified_attempt(&mut self, now: Instant, base: Duration) { self.attempt_in_flight = false; self.consecutive_unverified_attempts = self.consecutive_unverified_attempts.saturating_add(1); @@ -289,7 +300,7 @@ impl PartialWrites { } let key = PartialWriteKey::new(&intent, source_bucket_incarnation_id); let previous = self.entries.get(&key); - let previous_was_held = previous.is_some_and(|entry| entry.state.is_parked()); + let previous_was_held = previous.is_some_and(|entry| entry.is_parked()); let mut retry_queued = previous.is_some_and(|entry| entry.retry_queued); let old_cost = previous.map_or(0, Self::entry_cost); let next_bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::cost(&intent)); @@ -407,7 +418,7 @@ impl PartialWrites { return false; }; if entry.anchor.as_ref() != Some(anchor) - || entry.state.is_parked() + || entry.is_parked() || anchor.bucket_incarnation_id.is_nil() || entry.source_bucket_incarnation_id != Some(anchor.bucket_incarnation_id) { @@ -439,7 +450,9 @@ impl PartialWrites { pub(super) fn legacy_generation_unknown_count(&self) -> usize { self.entries .values() - .filter(|entry| matches!(&entry.state, ResponsibilityState::LegacyGenerationUnknown { .. })) + .filter(|entry| { + entry.generation_unobserved() || matches!(&entry.state, ResponsibilityState::LegacyGenerationUnknown { .. }) + }) .count() } @@ -502,7 +515,7 @@ impl PartialWrites { entry.last_operator_acceptance = record.last_operator_acceptance.clone(); entry.state = record.state.clone(); self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); - if entry.state.is_parked() { + if entry.is_parked() { entry.retry_queued = false; } true @@ -522,7 +535,7 @@ impl PartialWrites { self.entries .values() .find(|entry| entry.responsibility_id == responsibility_id) - .is_some_and(|entry| matches!(&entry.state, ResponsibilityState::Active)) + .is_some_and(|entry| !entry.is_parked()) } pub(super) fn last_operator_acceptance(&self, responsibility_id: Uuid) -> Option { @@ -703,7 +716,7 @@ impl PartialWrites { entry.last_operator_acceptance = last_operator_acceptance; self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); entry.persisted = true; - if entry.state.is_parked() { + if entry.is_parked() { entry.retry_queued = false; } true @@ -717,23 +730,24 @@ impl PartialWrites { let mut selected = BTreeMap::new(); for entry in self.entries.values() { let (status, bucket_incarnation_id, held_since_ms) = match &entry.state { + ResponsibilityState::Active if entry.generation_unobserved() => ("legacy_generation_unobserved", None, None), ResponsibilityState::Active => continue, ResponsibilityState::HeldUnverifiedLegacy { bucket_incarnation_id, since_ms, - } => ("held_unverified_legacy", *bucket_incarnation_id, Some(*since_ms)), + } => ("held_unverified_legacy", Some(*bucket_incarnation_id), Some(*since_ms)), ResponsibilityState::LegacyGenerationUnknown { observed_bucket_incarnation_id, detected_at_ms, - } => ("legacy_generation_unknown", *observed_bucket_incarnation_id, Some(*detected_at_ms)), + } => ("legacy_generation_unknown", Some(*observed_bucket_incarnation_id), Some(*detected_at_ms)), ResponsibilityState::BucketIncarnationChanged { observed_bucket_incarnation_id, detected_at_ms, .. - } => ("bucket_incarnation_changed", *observed_bucket_incarnation_id, Some(*detected_at_ms)), + } => ("bucket_incarnation_changed", Some(*observed_bucket_incarnation_id), Some(*detected_at_ms)), ResponsibilityState::OperatorAcceptedUnverified { bucket_incarnation_id, .. - } => ("operator_accepted_unverified", *bucket_incarnation_id, None), + } => ("operator_accepted_unverified", Some(*bucket_incarnation_id), None), }; if after.is_some_and(|after| entry.responsibility_id <= after) { continue; @@ -783,7 +797,7 @@ impl PartialWrites { pub(super) fn replace_observed_incarnation( &mut self, responsibility_id: Uuid, - expected_bucket_incarnation_id: Uuid, + expected_bucket_incarnation_id: Option, observed_bucket_incarnation_id: Uuid, source_bucket_incarnation_id: Option, ) -> Result<(ResponsibilityState, Option), &'static str> { @@ -792,9 +806,16 @@ impl PartialWrites { .values_mut() .find(|entry| entry.responsibility_id == responsibility_id) .ok_or("responsibility not found or generation changed")?; - if entry.state.bucket_incarnation_id() != Some(expected_bucket_incarnation_id) { + if entry.state.bucket_incarnation_id() != expected_bucket_incarnation_id { return Err("responsibility changed; refresh the listing"); } + if observed_bucket_incarnation_id.is_nil() + || expected_bucket_incarnation_id.is_some_and(|id| id.is_nil()) + || (expected_bucket_incarnation_id.is_none() && !entry.generation_unobserved()) + || entry.source_bucket_incarnation_id != source_bucket_incarnation_id + { + return Err("responsibility identity changed; refresh the listing"); + } let previous = entry.state.clone(); let previous_acceptance = entry.last_operator_acceptance.clone(); if source_bucket_incarnation_id.is_some_and(|source| source != observed_bucket_incarnation_id) { @@ -859,20 +880,6 @@ impl PartialWrites { // parked, so every dispatch must compare against a fresh identity. if let Some(anchor) = manager.durable_mrf_repair_anchor(&entry.intent).await { entry.anchor = Some(anchor.clone()); - if entry.source_bucket_incarnation_id.is_none() { - let old_cost = Self::entry_cost(entry); - let detected_at_ms = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map_or(0, |duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)); - entry.state = ResponsibilityState::LegacyGenerationUnknown { - observed_bucket_incarnation_id: anchor.bucket_incarnation_id, - detected_at_ms, - }; - self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); - entry.retry_queued = false; - lifecycle_changed = true; - continue; - } if let Some(source_bucket_incarnation_id) = entry.source_bucket_incarnation_id && source_bucket_incarnation_id != anchor.bucket_incarnation_id { @@ -923,7 +930,7 @@ impl PartialWrites { continue; }; entry.retry_queued = false; - if entry.state.is_parked() { + if entry.is_parked() { // Held obligations remain in the durable responsibility map, // but leave the hot retry index until a new generation arrives. continue; @@ -981,7 +988,7 @@ pub struct MrfLegacyResponsibility { pub version_id: Option, pub pool_index: Option, pub set_index: Option, - pub bucket_incarnation_id: Uuid, + pub bucket_incarnation_id: Option, pub enqueued_at_ms: u64, pub status: &'static str, pub held_since_ms: Option, @@ -1057,6 +1064,62 @@ mod tests { } } + #[test] + fn unobserved_legacy_is_held_and_visible_after_checkpoint_replay() { + let original = intent("deleted-bucket-legacy"); + let mut writes = PartialWrites::default(); + writes.admit(original.clone(), 1, 8192).expect("retain legacy responsibility"); + writes.mark_persisted(); + assert!(writes.ready_keys(Instant::now() + Duration::from_secs(86400), 1).is_empty()); + assert!(writes.retry_order.is_empty(), "unobserved legacy work must leave the hot retry queue"); + let record = writes.checkpoint_records(|_| Some([7; 32])).remove(0); + assert!(record.is_valid()); + assert!(matches!(record.state, ResponsibilityState::Active), "keep the legacy disk encoding"); + assert_eq!(record.source_bucket_incarnation_id, None); + let (rows, next) = writes.list_unverified_legacy(None, 1); + assert_eq!(rows.len(), 1); + assert!(next.is_none()); + assert_eq!(rows[0].status, "legacy_generation_unobserved"); + assert_eq!(rows[0].bucket_incarnation_id, None); + assert_eq!(writes.legacy_generation_unknown_count(), 1); + assert!(!writes.is_active_responsibility(record.responsibility_id)); + let json = serde_json::to_value(&rows[0]).expect("encode unknown observation"); + assert!(json["bucketIncarnationId"].is_null()); + + let encoded = serde_json::to_vec(&record).expect("encode unchanged checkpoint schema"); + let decoded: ResponsibilityCheckpoint = serde_json::from_slice(&encoded).expect("decode legacy schema"); + let mut replay = PartialWrites::default(); + replay.admit(original.clone(), 1, 8192).expect("replay intent"); + assert!(replay.restore_state(&original, &decoded)); + replay.mark_persisted(); + assert!(replay.ready_keys(Instant::now() + Duration::from_secs(86400), 1).is_empty()); + assert_eq!(replay.checkpoint_records(|_| Some([7; 32])), vec![record]); + assert_eq!(replay.list_unverified_legacy(None, 1).0[0].responsibility_id, rows[0].responsibility_id); + } + + #[test] + fn unobserved_legacy_refresh_preserves_unknown_source_and_rejects_stale_actions() { + let original = intent("unobserved-refresh"); + let mut writes = PartialWrites::default(); + writes.admit(original, 1, 8192).expect("retain legacy responsibility"); + let id = writes.list_unverified_legacy(None, 1).0[0].responsibility_id; + assert!(writes.replace_observed_incarnation(id, None, Uuid::nil(), None).is_err()); + let observed = Uuid::new_v4(); + writes + .replace_observed_incarnation(id, None, observed, None) + .expect("observe a current generation"); + let row = writes.list_unverified_legacy(None, 1).0.remove(0); + assert_eq!(row.status, "legacy_generation_unknown"); + assert_eq!(row.bucket_incarnation_id, Some(observed)); + assert_eq!(row.source_bucket_incarnation_id, None); + assert!(writes.replace_observed_incarnation(id, None, Uuid::new_v4(), None).is_err()); + assert!( + writes.recheck(id, observed).is_err(), + "observation cannot authorize repair of an unknown source" + ); + assert!(writes.ready_keys(Instant::now() + Duration::from_secs(86400), 1).is_empty()); + } + #[test] fn durable_retry_backoff_is_bounded_staggered_and_not_checkpointed() { let item = intent("retry-backoff"); @@ -1409,7 +1472,9 @@ mod tests { fn partial_write_retention_retries_rotate_past_a_failing_prefix() { let mut writes = PartialWrites::default(); for object in ["a", "b", "c"] { - writes.admit(intent(object), 3, 8192).expect("bounded backlog should fit"); + writes + .admit_with_source_incarnation(intent(object), Some(Uuid::new_v4()), 3, 8192) + .expect("bounded source-bound backlog should fit"); } let now = Instant::now(); assert!(writes.ready_keys(now, 2).is_empty(), "uncommitted responsibility must not be dispatched"); @@ -1597,7 +1662,7 @@ mod tests { let (previous, _) = restored .replace_observed_incarnation( checkpoint.responsibility_id, - original_incarnation, + Some(original_incarnation), recreated_incarnation, Some(original_incarnation), ) @@ -1955,11 +2020,12 @@ mod tests { .admit_with_source_incarnation(legacy.clone(), None, 2, 64 * 1024) .expect("retain a legacy responsibility without inventing its source"); writes.mark_persisted(); - assert!(writes.dispatch_collecting_lifecycle_change(&manager, &config).await); + assert!(!writes.dispatch_collecting_lifecycle_change(&manager, &config).await); assert_eq!(writes.legacy_generation_unknown_count(), 1); let original_unknown = writes.checkpoint_records(super::super::intent_digest).remove(0); assert_eq!(original_unknown.source_bucket_incarnation_id, None); - assert!(matches!(original_unknown.state, ResponsibilityState::LegacyGenerationUnknown { .. })); + assert!(matches!(original_unknown.state, ResponsibilityState::Active)); + assert_eq!(writes.list_unverified_legacy(None, 10).0[0].status, "legacy_generation_unobserved"); let current = intent("same-object-unknown-source"); assert_eq!(super::super::intent_digest(&legacy), super::super::intent_digest(¤t)); diff --git a/crates/heal/tests/mrf_partial_write_test.rs b/crates/heal/tests/mrf_partial_write_test.rs index 1a7243782..99cbe5e57 100644 --- a/crates/heal/tests/mrf_partial_write_test.rs +++ b/crates/heal/tests/mrf_partial_write_test.rs @@ -30,8 +30,8 @@ use tokio::io::AsyncReadExt; mod storage_api; use storage_api::endpoint_index::{EndpointServerPools, Endpoints, init_local_disks}; use storage_api::integration::{ - DiskAPI, DiskError, DiskStore, EcstoreStorageError, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader, - RUSTFS_META_BUCKET, ReadOptions, + BucketOperations, DeleteBucketOptions, DiskAPI, DiskError, DiskStore, EcstoreStorageError, ObjectIO, ObjectOperations, + ObjectOptions, PutObjReader, RUSTFS_META_BUCKET, ReadOptions, }; const SNAPSHOT_LIMIT: usize = 64 * 1024 * 1024; @@ -131,6 +131,10 @@ async fn legacy_unbound_generation_is_parked_and_requires_explicit_risk_acceptan ) .await .expect("old-format intent should still be durably admitted"); + env.ecstore + .delete_bucket("unbound-lifecycle", &DeleteBucketOptions::default()) + .await + .expect("delete the legacy intent's bucket through the supported lifecycle"); assert!( wait_until(|| async { @@ -152,7 +156,9 @@ async fn legacy_unbound_generation_is_parked_and_requires_explicit_risk_acceptan .iter() .find(|item| item.bucket == "unbound-lifecycle" && item.object == "legacy.bin") .expect("visible generation-unknown entry"); - assert_eq!(entry.status, "legacy_generation_unknown"); + assert_eq!(entry.status, "legacy_generation_unobserved"); + assert_eq!(entry.bucket_incarnation_id, None); + assert_eq!(snapshot.contract_version, 2); assert!(entry.source_bucket_incarnation_id.is_none()); assert!( snapshot_contains("legacy.bin").await, @@ -164,13 +170,42 @@ async fn legacy_unbound_generation_is_parked_and_requires_explicit_risk_acceptan "unbound old intent must not be sent to the current bucket generation" ); + assert!( + matches!( + mrf_queue::refresh_legacy_responsibility(entry.responsibility_id, None).await, + Err(rustfs_heal::heal::mrf_queue::MrfLifecycleControlError::IncarnationChanged) + ), + "an absent bucket is not retirement proof" + ); + assert!(snapshot_contains("legacy.bin").await, "failed observation must retain the durable intent"); + env.make_bucket("unbound-lifecycle", false).await; + mrf_queue::refresh_legacy_responsibility(entry.responsibility_id, None) + .await + .expect("explicit refresh may observe a recreated bucket without binding the old source"); + let refreshed = mrf_queue::list_legacy_responsibilities(None, 8) + .await + .expect("read refreshed observation"); + let refreshed = refreshed + .responsibilities + .iter() + .find(|row| row.responsibility_id == entry.responsibility_id) + .expect("refresh must preserve the responsibility ID"); + assert_eq!(refreshed.status, "legacy_generation_unknown"); + assert_eq!(refreshed.source_bucket_incarnation_id, None); + let observed = refreshed + .bucket_incarnation_id + .expect("refresh records a real current identity"); assert!(matches!( - mrf_queue::recheck_legacy_responsibility(entry.responsibility_id, entry.bucket_incarnation_id).await, + mrf_queue::refresh_legacy_responsibility(entry.responsibility_id, None).await, + Err(rustfs_heal::heal::mrf_queue::MrfLifecycleControlError::StaleGeneration) + )); + assert!(matches!( + mrf_queue::recheck_legacy_responsibility(entry.responsibility_id, observed).await, Err(rustfs_heal::heal::mrf_queue::MrfLifecycleControlError::InvalidAction(_)) )); mrf_queue::accept_unverified_legacy_risk(mrf_queue::MrfLegacyRiskAcceptanceRequest { responsibility_id: entry.responsibility_id, - expected_bucket_incarnation_id: entry.bucket_incarnation_id, + expected_bucket_incarnation_id: observed, acknowledge_unknown_source_incarnation: true, acknowledge_incarnation_mismatch: false, actor: "integration-test-operator".to_string(), @@ -975,7 +1010,9 @@ async fn partial_write_sigkill_replay_scenario(protected: bool) { "the producer-bound bucket generation must survive journal replay" ); let responsibility_id = held.responsibility_id; - let bucket_incarnation_id = held.bucket_incarnation_id; + let bucket_incarnation_id = held + .bucket_incarnation_id + .expect("source-bound hold must retain its observed identity"); let request_id = uuid::Uuid::new_v4(); mrf_queue::accept_unverified_legacy_risk(mrf_queue::MrfLegacyRiskAcceptanceRequest { responsibility_id, diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs index f72a91b98..682c4d8f5 100644 --- a/crates/heal/tests/mrf_pipeline_test.rs +++ b/crates/heal/tests/mrf_pipeline_test.rs @@ -315,9 +315,9 @@ fn committed_checkpoint_matches_on_all_disks(disk_paths: &[PathBuf], sequence: u async fn assert_legacy_responsibilities_parked( disk_paths: &[PathBuf], expected_records: &[&[u8]], - observed_bucket_incarnation_id: uuid::Uuid, + current_bucket_incarnation_id: uuid::Uuid, ) { - assert!(!observed_bucket_incarnation_id.is_nil(), "the observed bucket generation must be real"); + assert!(!current_bucket_incarnation_id.is_nil(), "the fixture current generation must be real"); let checkpoint = mrf_queue::snapshot::inspect_local_committed_snapshot(rustfs_config::DEFAULT_HEAL_MRF_JOURNAL_MAX_BYTES) .await .expect("inspect the committed legacy responsibility checkpoint") @@ -384,12 +384,11 @@ async fn assert_legacy_responsibilities_parked( Some(&serde_json::Value::Null), "replay must not fabricate operator authorization" ); - assert_eq!(record["state"]["state"], "legacy_generation_unknown"); assert_eq!( - record["state"]["observed_bucket_incarnation_id"], - observed_bucket_incarnation_id.to_string() + record["state"], + serde_json::json!({"state": "active"}), + "the source-unbound hold must preserve the old disk state without inventing an observation" ); - assert!(record["state"]["detected_at_ms"].as_u64().is_some_and(|time| time > 0)); } } diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index ae4915866..d2dcb39d3 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -397,7 +397,9 @@ Durable partial-write responsibilities also have node-local MRF metrics: These gauges are emitted per node through the configured OTLP metrics exporter (`RUSTFS_OBS_ENDPOINT` or `RUSTFS_OBS_METRIC_ENDPOINT`). The background-heal status endpoint reports execution tasks; it does not include durable MRF responsibilities. Use the authenticated node-local `GET /rustfs/admin/v4/heal/mrf/responsibilities?limit=100&cursor=` endpoint to list held, generation-unknown, incarnation-mismatched, or operator-accepted entries, including the stable responsibility ID, exact object/version/scope, source/current bucket incarnation, checkpoint owner/sequence, and any risk-acceptance record. `limit` is bounded to 1–256 and `nextCursor` is absent on the last page. If the listed bucket incarnation may have changed since the responsibility was parked, call the action endpoint with `action: "refresh"`, the listed `responsibilityId`, and its current `expectedBucketIncarnationId`. Refresh reads the live bucket identity, persists a changed-generation classification, and never dispatches a heal. Then review the fresh listing before taking another action. Use `action: "recheck"` to resume one exact source-bound held generation after restoring a trusted source or preparing an exact-version deletion. A recheck is persisted before the task is dispatched. -An operator may instead choose `action: "acceptUnverifiedRisk"` only with `acknowledgeUnverifiedDataRisk: true`, explicit booleans for both source-incarnation acknowledgments, a non-empty reason, an audit reference, and a non-nil request ID. When `sourceBucketIncarnationId` is absent, set `acknowledgeUnknownSourceIncarnation: true`; when it differs from the current `bucketIncarnationId`, set `acknowledgeBucketIncarnationMismatch: true`. A generation with a known source/current mismatch is parked before dispatch, so its old intent cannot be applied to a recreated bucket. Legacy journal records without a source incarnation are listed as `legacy_generation_unknown` and are not automatically dispatched; refresh can record the live generation but does not bind the old responsibility to it. This does not create a payload proof, delete the object, or remove the durable intent. It persists an `operator_accepted_unverified` disposition that stops automatic retries while leaving the responsibility visible. The actor, reason, reference, time, request ID, target identity, source/current bucket incarnations, and acknowledgment flags are kept in a checksummed lifecycle record in the same disk replica and alternating slot as its committed MRF checkpoint. If that lifecycle record is missing, corrupt, or belongs to another checkpoint, unbound records return to `legacy_generation_unknown`; they are not treated as proof or dispatched against an unbound bucket. Pre-lifecycle binaries ignore this state and can heal a compatibility-journal object into a recreated bucket generation; downgrading while any durable MRF responsibility remains is unsupported. See `backlog-2682` in the compatibility cleanup register. +The listing's `contractVersion: 2` also includes `legacy_generation_unobserved` entries. Their `sourceBucketIncarnationId` and `bucketIncarnationId` are null: neither the original generation nor a current observation is available. These entries remain held across replay without automatic bucket-identity retry loops. Startup replay retains its existing one-time anchor lookup; a missing identity cannot resume the steady-state retry loop. The hold is derived from the existing source-unbound active checkpoint encoding; no new disk state tag is written. Use `action: "refresh"` with `expectedBucketIncarnationId: null` to request an observation. If no current identity can be read, refresh fails and retains the responsibility. If one is available, refresh records it as `legacy_generation_unknown`; it never manufactures a source identity. A stale null expectation is rejected once an observation has been recorded. Recheck and risk acceptance still require a known, non-nil current incarnation. Older server binaries can resume the old identity retry behavior after rollback; they must not be used to discharge this unresolved work. + +An operator may instead choose `action: "acceptUnverifiedRisk"` only with `acknowledgeUnverifiedDataRisk: true`, explicit booleans for both source-incarnation acknowledgments, a non-empty reason, an audit reference, and a non-nil request ID. When `sourceBucketIncarnationId` is absent, set `acknowledgeUnknownSourceIncarnation: true`; when it differs from the current `bucketIncarnationId`, set `acknowledgeBucketIncarnationMismatch: true`. A generation with a known source/current mismatch is parked before dispatch, so its old intent cannot be applied to a recreated bucket. Legacy journal records without a source incarnation are listed as `legacy_generation_unobserved` until refresh records a current identity, then as `legacy_generation_unknown` and are not automatically dispatched; refresh can record the live generation but does not bind the old responsibility to it. This does not create a payload proof, delete the object, or remove the durable intent. It persists an `operator_accepted_unverified` disposition that stops automatic retries while leaving the responsibility visible. The actor, reason, reference, time, request ID, target identity, source/current bucket incarnations, and acknowledgment flags are kept in a checksummed lifecycle record in the same disk replica and alternating slot as its committed MRF checkpoint. If that lifecycle record is missing, corrupt, or belongs to another checkpoint, unbound records return to `legacy_generation_unobserved`; they are not treated as proof or dispatched against an unbound bucket. Pre-lifecycle binaries ignore this state and can heal a compatibility-journal object into a recreated bucket generation; downgrading while any durable MRF responsibility remains is unsupported. See `backlog-2682` in the compatibility cleanup register. Example risk-acceptance body (substitute values from the node-local listing): diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index f0ef8c491..33ac358eb 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -1615,7 +1615,7 @@ pub struct BackgroundHealStatusHandler {} struct MrfLegacyResponsibilityActionRequest { action: MrfLegacyResponsibilityAction, responsibility_id: uuid::Uuid, - expected_bucket_incarnation_id: uuid::Uuid, + expected_bucket_incarnation_id: Option, #[serde(default)] acknowledge_unverified_data_risk: Option, #[serde(default)] @@ -1639,9 +1639,12 @@ enum MrfLegacyResponsibilityAction { } fn validate_mrf_legacy_responsibility_action(action: &MrfLegacyResponsibilityActionRequest) -> Result<(), &'static str> { - if action.responsibility_id.is_nil() || action.expected_bucket_incarnation_id.is_nil() { + if action.responsibility_id.is_nil() || action.expected_bucket_incarnation_id.is_some_and(|id| id.is_nil()) { return Err("responsibility and bucket incarnation IDs must be non-nil"); } + if action.expected_bucket_incarnation_id.is_none() && !matches!(action.action, MrfLegacyResponsibilityAction::Refresh) { + return Err("a known bucket incarnation is required for recheck and risk acceptance"); + } match action.action { MrfLegacyResponsibilityAction::Refresh | MrfLegacyResponsibilityAction::Recheck => { if action.acknowledge_unverified_data_risk.is_some() @@ -1797,7 +1800,9 @@ impl Operation for MrfLegacyResponsibilitiesActionHandler { Duration::from_secs(30), rustfs_heal::heal::mrf_queue::recheck_legacy_responsibility( action.responsibility_id, - action.expected_bucket_incarnation_id, + action + .expected_bucket_incarnation_id + .ok_or_else(|| admin_error(S3ErrorCode::InvalidRequest, "expectedBucketIncarnationId is required"))?, ), ) .await @@ -1820,7 +1825,9 @@ impl Operation for MrfLegacyResponsibilitiesActionHandler { rustfs_heal::heal::mrf_queue::accept_unverified_legacy_risk( rustfs_heal::heal::mrf_queue::MrfLegacyRiskAcceptanceRequest { responsibility_id: action.responsibility_id, - expected_bucket_incarnation_id: action.expected_bucket_incarnation_id, + expected_bucket_incarnation_id: action.expected_bucket_incarnation_id.ok_or_else(|| { + admin_error(S3ErrorCode::InvalidRequest, "expectedBucketIncarnationId is required") + })?, acknowledge_unknown_source_incarnation: action.acknowledge_unknown_source_incarnation == Some(true), acknowledge_incarnation_mismatch: action.acknowledge_bucket_incarnation_mismatch == Some(true), actor, @@ -1972,6 +1979,26 @@ mod tests { use std::sync::atomic::{AtomicBool, Ordering}; use uuid::Uuid; + #[test] + fn mrf_legacy_only_refresh_accepts_an_unobserved_incarnation() { + for action in ["refresh", "recheck", "acceptUnverifiedRisk"] { + let request: MrfLegacyResponsibilityActionRequest = serde_json::from_value(json!({ + "action": action, + "responsibilityId": Uuid::new_v4(), + "expectedBucketIncarnationId": null, + })) + .expect("decode explicit unknown observation"); + assert_eq!(validate_mrf_legacy_responsibility_action(&request).is_ok(), action == "refresh"); + } + let request: MrfLegacyResponsibilityActionRequest = serde_json::from_value(json!({ + "action": "refresh", + "responsibilityId": Uuid::new_v4(), + "expectedBucketIncarnationId": Uuid::nil(), + })) + .expect("decode nil identity"); + assert!(validate_mrf_legacy_responsibility_action(&request).is_err()); + } + #[test] fn mrf_legacy_risk_action_requires_explicit_acknowledgment_and_audit_fields() { let base = json!({ @@ -2021,7 +2048,7 @@ mod tests { let recheck = MrfLegacyResponsibilityActionRequest { action: MrfLegacyResponsibilityAction::Recheck, responsibility_id: Uuid::new_v4(), - expected_bucket_incarnation_id: Uuid::new_v4(), + expected_bucket_incarnation_id: Some(Uuid::new_v4()), acknowledge_unverified_data_risk: None, acknowledge_unknown_source_incarnation: None, acknowledge_bucket_incarnation_mismatch: None,