Merge branch 'main' into houseme/fix/backlog-2992

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/core/pool_meta_native_integration_tests.rs

@@ -18,6 +18,7 @@ use crate::storage_api_contracts::namespace::NamespaceLocking as _;
 use codec::{fixture_inline_pool_meta_replicas as replicas, fixture_split_inline_pool_meta_replicas as split};
 use std::path::PathBuf;
 use tokio::fs;
+use uuid::Uuid;
 
 struct PhysicalFixture {
     _directories: Vec<tempfile::TempDir>,
@@ -628,3 +629,559 @@ async fn pool_meta_native_physical_three_plus_one_checkpoint_converges_all_shard
         );
     }
 }
+
+async fn owner_repair_no_backups_fixture() -> PhysicalFixture {
+    let fixture = fixture(0).await;
+    for pool in disks(&fixture.store).await {
+        for disk in pool {
+            let object = object_path(&disk);
+            let mut entries = fs::read_dir(&object).await.expect("fixture metadata directory");
+            while let Some(entry) = entries.next_entry().await.expect("fixture directory entry") {
+                if entry.file_type().await.expect("fixture entry type").is_dir() {
+                    fs::remove_dir_all(entry.path())
+                        .await
+                        .expect("remove task-owned fixture rollback directory");
+                }
+            }
+        }
+    }
+    fixture
+}
+
+async fn owner_repair_identity_heads(store: &ECStore) -> Vec<Vec<(Uuid, Bytes)>> {
+    let mut all = Vec::new();
+    for pool in &store.pools {
+        let set = pool.get_disks_by_key(POOL_META_IDENTITY_NAME);
+        let members = set
+            .disks
+            .read()
+            .await
+            .iter()
+            .cloned()
+            .map(|disk| disk.expect("identity fixture disk must be online"))
+            .collect::<Vec<_>>();
+        let mut heads = Vec::new();
+        for disk in members {
+            match disk
+                .pool_meta_native_recovery(Request::InspectIdentity {})
+                .await
+                .expect("physical identity inspection")
+            {
+                Response::IdentityInspected { disk_id, raw } => heads.push((disk_id, raw)),
+                _ => panic!("identity inspection must return its physical head"),
+            }
+        }
+        all.push(heads);
+    }
+    all
+}
+
+async fn legacy_native_owner_fixture(version: u16, stale: bool) -> (PhysicalFixture, Vec<u8>) {
+    let fixture = owner_repair_no_backups_fixture().await;
+    let mut meta = match decode_pool_meta_replica(fixture.canonical.clone()) {
+        PoolMetaReplica::Valid {
+            meta, committed: true, ..
+        } => meta,
+        _ => panic!("fixture must provide valid committed pool metadata"),
+    };
+    meta.version = version;
+    let raw = meta
+        .encode_config_data_for_v2_gate(version == POOL_META_VERSION)
+        .expect("legacy fixture should encode in its original version");
+    assert_eq!(u16::from_le_bytes([raw[2], raw[3]]), version);
+    let all = disks(&fixture.store).await;
+    let now = OffsetDateTime::now_utc();
+    for pool in &all {
+        let heads = replicas(&raw, now).await;
+        for (index, disk) in pool.iter().enumerate() {
+            fs::write(object_path(disk).join("xl.meta"), &heads[index])
+                .await
+                .expect("install exact legacy head");
+        }
+    }
+    if stale {
+        let old = replicas(&raw, now - Duration::seconds(10)).await;
+        fs::write(object_path(&all[0][0]).join("xl.meta"), &old[0])
+            .await
+            .expect("install stale same-raw legacy head");
+    }
+    (fixture, raw)
+}
+
+#[tokio::test]
+#[serial_test::serial(pool_meta_version_env)]
+async fn pool_meta_native_heal_v1_with_all_upgrade_gates_disabled_keeps_exact_raw() {
+    temp_env::async_with_vars(
+        [
+            (rustfs_config::ENV_POOL_META_V2_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V2_FLEET_CONFIRMED, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_FLEET_CONFIRMED, Some("false")),
+        ],
+        async {
+            let (fixture, raw) = legacy_native_owner_fixture(POOL_META_V1_VERSION, false).await;
+            let guard = owner(&fixture.store).await;
+            let mut state = fixture.store.pool_meta_save_gate.lock().await;
+            let result = tokio::time::timeout(
+                std::time::Duration::from_secs(30),
+                fixture.store.heal_pool_meta_replica_under_fence(0, guard, &mut state),
+            )
+            .await
+            .expect("legacy native heal must terminate");
+            assert!(result.is_ok(), "fully online legacy native heal should succeed: {result:?}");
+            drop(state);
+            let reads = read_pool_meta_replicas(fixture.store.pools.clone(), true).await;
+            for read in &reads {
+                assert!(
+                    matches!(&read.replica, PoolMetaReplica::Valid { raw: bytes, revision, .. }
+                    if bytes == &raw && revision.version == POOL_META_V1_VERSION),
+                    "native heal must not upgrade a V1 raw payload when both upgrade gates are disabled"
+                );
+            }
+        },
+    )
+    .await;
+}
+
+#[tokio::test]
+#[serial_test::serial(pool_meta_version_env)]
+async fn pool_meta_native_heal_rejects_legacy_source_with_trailing_bytes_without_writes() {
+    temp_env::async_with_vars(
+        [
+            (rustfs_config::ENV_POOL_META_V2_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V2_FLEET_CONFIRMED, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_FLEET_CONFIRMED, Some("false")),
+        ],
+        async {
+            for version in [POOL_META_V1_VERSION, POOL_META_VERSION] {
+                let (fixture, raw) = legacy_native_owner_fixture(version, false).await;
+                let mut source_raw = raw.clone();
+                // A second MessagePack value is readable by legacy decoding but
+                // must not be copied as part of an unverified repair payload.
+                source_raw.push(0xc0);
+                assert!(
+                    matches!(decode_pool_meta_replica(source_raw.clone()), PoolMetaReplica::Valid { raw, .. } if raw == source_raw),
+                    "ordinary legacy reads must remain compatible with trailing bytes"
+                );
+                let all_disks = disks(&fixture.store).await;
+                let source_heads = replicas(&source_raw, OffsetDateTime::now_utc() + Duration::seconds(1)).await;
+                for (index, disk) in all_disks[0].iter().enumerate() {
+                    fs::write(object_path(disk).join("xl.meta"), &source_heads[index])
+                        .await
+                        .expect("install readable legacy source with trailing bytes");
+                }
+                let selection = load_pool_meta_replicas(fixture.store.pools.clone(), true)
+                    .await
+                    .expect("legacy semantic selection must remain readable");
+                assert_eq!(selection.legacy_raw.as_deref(), Some(source_raw.as_slice()));
+                let before = snapshots(&fixture.store).await;
+                let identities = owner_repair_identity_heads(&fixture.store).await;
+                let guard = owner(&fixture.store).await;
+                let mut state = fixture.store.pool_meta_save_gate.lock().await;
+                let result = tokio::time::timeout(
+                    std::time::Duration::from_secs(30),
+                    fixture.store.heal_pool_meta_replica_under_fence(1, guard, &mut state),
+                )
+                .await
+                .expect("legacy native heal must terminate");
+                drop(state);
+                assert!(result.is_err(), "version {version}: unparsed trailing source bytes must not be republished");
+                assert_eq!(snapshots(&fixture.store).await, before, "version {version}: no pool.bin head or backup may change");
+                assert_eq!(
+                    owner_repair_identity_heads(&fixture.store).await,
+                    identities,
+                    "version {version}: no identity shard may change"
+                );
+                let reads = read_pool_meta_replicas(fixture.store.pools.clone(), true).await;
+                assert!(matches!(&reads[0].replica, PoolMetaReplica::Valid { raw, .. } if raw == &source_raw));
+                assert!(matches!(&reads[1].replica, PoolMetaReplica::Valid { raw: target, .. } if target == &raw));
+            }
+        },
+    )
+    .await;
+}
+
+#[tokio::test]
+#[serial_test::serial(pool_meta_version_env)]
+async fn pool_meta_native_heal_mixed_v1_v2_preserves_observed_format_floor() {
+    temp_env::async_with_vars(
+        [
+            (rustfs_config::ENV_POOL_META_V2_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V2_FLEET_CONFIRMED, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_FLEET_CONFIRMED, Some("false")),
+        ],
+        async {
+            for first_version in [POOL_META_V1_VERSION, POOL_META_VERSION] {
+                for target_pool in [0, 1] {
+                    let (fixture, raw_v1) = legacy_native_owner_fixture(POOL_META_V1_VERSION, false).await;
+                    let mut meta = match decode_pool_meta_replica(raw_v1.clone()) {
+                        PoolMetaReplica::Valid { meta, committed: true, .. } => meta,
+                        _ => panic!("initial V1 metadata must decode"),
+                    };
+                    meta.version = POOL_META_VERSION;
+                    let raw_v2 = meta.encode_config_data_for_v2_gate(true).expect("equivalent V2 payload must encode");
+                    assert_eq!(u16::from_le_bytes([raw_v1[2], raw_v1[3]]), POOL_META_V1_VERSION);
+                    assert_eq!(u16::from_le_bytes([raw_v2[2], raw_v2[3]]), POOL_META_VERSION);
+                    let semantic_v1 = match decode_pool_meta_replica(raw_v1.clone()) {
+                        PoolMetaReplica::Valid { canonical, .. } => canonical,
+                        _ => panic!("V1 payload must remain valid"),
+                    };
+                    let semantic_v2 = match decode_pool_meta_replica(raw_v2.clone()) {
+                        PoolMetaReplica::Valid { canonical, .. } => canonical,
+                        _ => panic!("V2 payload must remain valid"),
+                    };
+                    assert_eq!(semantic_v1, semantic_v2, "the two formats must carry equivalent metadata");
+                    let other_version = if first_version == POOL_META_V1_VERSION { POOL_META_VERSION } else { POOL_META_V1_VERSION };
+                    // Pool two follows pool zero; pools zero and one cover both
+                    // orderings and both target formats in separate runs.
+                    let versions = [first_version, other_version, first_version];
+                    let physical = disks(&fixture.store).await;
+                    let timestamp = OffsetDateTime::now_utc();
+                    for (pool_index, pool) in physical.iter().enumerate() {
+                        let raw = if versions[pool_index] == POOL_META_V1_VERSION { &raw_v1 } else { &raw_v2 };
+                        let heads = replicas(raw, timestamp).await;
+                        for (index, disk) in pool.iter().enumerate() {
+                            fs::write(object_path(disk).join("xl.meta"), &heads[index]).await.expect("install mixed legacy pool head");
+                        }
+                    }
+                    let before = snapshots(&fixture.store).await;
+                    let selection = load_pool_meta_replicas(fixture.store.pools.clone(), true)
+                        .await.expect("equivalent legacy pools should be selectable");
+                    assert_eq!(selection.revision.version, POOL_META_VERSION, "the observed V2 format floor must be retained");
+                    assert_eq!(selection.canonical.as_deref(), Some(semantic_v2.as_slice()));
+                    let guard = owner(&fixture.store).await;
+                    let mut state = fixture.store.pool_meta_save_gate.lock().await;
+                    tokio::time::timeout(std::time::Duration::from_secs(30),
+                        fixture.store.heal_pool_meta_replica_under_fence(target_pool, guard, &mut state))
+                        .await.expect("mixed legacy heal must terminate")
+                        .expect("equivalent legacy metadata should heal without downgrading the format floor");
+                    drop(state);
+                    let after = snapshots(&fixture.store).await;
+                    let reads = read_pool_meta_replicas(fixture.store.pools.clone(), true).await;
+                    for (pool_index, read) in reads.iter().enumerate() {
+                        let expected_version = if pool_index == target_pool { POOL_META_VERSION } else { versions[pool_index] };
+                        let expected_raw = if expected_version == POOL_META_V1_VERSION { &raw_v1 } else { &raw_v2 };
+                        assert!(matches!(&read.replica,
+                            PoolMetaReplica::Valid { raw, canonical, revision, committed: true, .. }
+                                if raw == expected_raw && canonical == &semantic_v2 && revision.version == expected_version
+                        ), "pool {pool_index}, first={first_version}, target={target_pool}: preserve equivalent payload and expected legacy version");
+                        for (index, snapshot) in after[pool_index].iter().enumerate() {
+                            codec::validate_inline_pool_meta_shard(snapshot.current.as_ref().expect("physical head"), index, expected_raw)
+                                .await.expect("every physical shard must match its selected legacy payload");
+                        }
+                        if pool_index != target_pool {
+                            assert_eq!(after[pool_index], before[pool_index], "healing one pool must not modify other pools");
+                        }
+                    }
+                }
+            }
+        },
+    ).await;
+}
+
+#[tokio::test]
+#[serial_test::serial(pool_meta_version_env)]
+async fn pool_meta_native_heal_rejects_v1_and_v3_prepare_in_either_pool_order() {
+    temp_env::async_with_vars(
+        [
+            (rustfs_config::ENV_POOL_META_V2_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V2_FLEET_CONFIRMED, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_FLEET_CONFIRMED, Some("false")),
+        ],
+        async {
+            for prepare_pool in [0, 1] {
+                let (fixture, raw_v1) = legacy_native_owner_fixture(POOL_META_V1_VERSION, false).await;
+                let (legacy_meta, legacy_canonical) = match decode_pool_meta_replica(raw_v1.clone()) {
+                    PoolMetaReplica::Valid {
+                        meta,
+                        canonical,
+                        revision,
+                        committed: true,
+                        ..
+                    } if revision.version == POOL_META_V1_VERSION => (meta, canonical),
+                    _ => panic!("baseline V1 metadata must be committed and readable"),
+                };
+                let previous = PoolMetaCommittedCandidate {
+                    canonical: legacy_canonical,
+                    meta: legacy_meta.clone(),
+                    revision: PoolMetaRevision::legacy(POOL_META_V1_VERSION),
+                };
+                let mut next = legacy_meta;
+                next.version = POOL_META_GENERATION_VERSION;
+                next.pools[0].last_update += Duration::seconds(1);
+                let revision = PoolMetaRevision {
+                    version: POOL_META_GENERATION_VERSION,
+                    cluster_id: fixture.revision.cluster_id,
+                    epoch: fixture.revision.epoch,
+                    generation: 1,
+                    transaction_id: Some(Uuid::new_v4()),
+                };
+                let pending = encode_pool_meta_v3_envelope(&next, revision, false, Some(&previous))
+                    .expect("V3 PREPARE with V1 predecessor must encode");
+                assert!(matches!(decode_pool_meta_replica(pending.clone()),
+                    PoolMetaReplica::Valid { committed: false, revision: actual, .. } if actual == revision));
+                let heads = replicas(&pending, OffsetDateTime::now_utc() + Duration::seconds(1)).await;
+                let physical = disks(&fixture.store).await;
+                for (index, disk) in physical[prepare_pool].iter().enumerate() {
+                    fs::write(object_path(disk).join("xl.meta"), &heads[index])
+                        .await
+                        .expect("install task-owned V3 PREPARE evidence");
+                }
+                let reads = read_pool_meta_replicas(fixture.store.pools.clone(), true).await;
+                assert!(matches!(&reads[prepare_pool].replica,
+                    PoolMetaReplica::Valid { raw, committed: false, revision: actual, .. }
+                        if raw == &pending && *actual == revision));
+                let other_pool = 1 - prepare_pool;
+                assert!(matches!(&reads[other_pool].replica,
+                    PoolMetaReplica::Valid { raw, committed: true, revision: actual, .. }
+                        if raw == &raw_v1 && actual.version == POOL_META_V1_VERSION));
+                let before = snapshots(&fixture.store).await;
+                let identities = owner_repair_identity_heads(&fixture.store).await;
+                let guard = owner(&fixture.store).await;
+                let mut state = fixture.store.pool_meta_save_gate.lock().await;
+                let result = tokio::time::timeout(
+                    std::time::Duration::from_secs(30),
+                    fixture
+                        .store
+                        .heal_pool_meta_replica_under_fence(prepare_pool, guard, &mut state),
+                )
+                .await
+                .expect("mixed V1/V3 PREPARE heal must terminate");
+                drop(state);
+                assert!(
+                    result.is_err(),
+                    "pool {prepare_pool}: native heal must not overwrite a V3 PREPARE with legacy data"
+                );
+                assert_eq!(
+                    snapshots(&fixture.store).await,
+                    before,
+                    "pool {prepare_pool}: rejection must retain all physical head and backup evidence"
+                );
+                assert_eq!(
+                    owner_repair_identity_heads(&fixture.store).await,
+                    identities,
+                    "pool {prepare_pool}: rejection must not mutate identity evidence"
+                );
+            }
+        },
+    )
+    .await;
+}
+
+#[tokio::test]
+#[serial_test::serial(pool_meta_version_env)]
+async fn pool_meta_native_heal_rejects_v1_legacy_dont_save_true_without_writes() {
+    temp_env::async_with_vars(
+        [
+            (rustfs_config::ENV_POOL_META_V2_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V2_FLEET_CONFIRMED, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_FLEET_CONFIRMED, Some("false")),
+        ],
+        async {
+            let (fixture, safe_v1) = legacy_native_owner_fixture(POOL_META_V1_VERSION, false).await;
+            let meta = match decode_pool_meta_replica(safe_v1.clone()) {
+                PoolMetaReplica::Valid {
+                    meta, committed: true, ..
+                } => meta,
+                _ => panic!("baseline V1 metadata must decode"),
+            };
+            let statuses = meta
+                .pools
+                .iter()
+                .map(|pool| {
+                    assert!(pool.decommission.is_none(), "this regression fixture does not inject decommission state");
+                    LegacyPoolStatus {
+                        id: pool.id,
+                        cmd_line: pool.cmd_line.clone(),
+                        last_update: pool.last_update,
+                        decommission: None,
+                    }
+                })
+                .collect();
+            let legacy = LegacyPoolMeta {
+                version: POOL_META_V1_VERSION,
+                pools: statuses,
+                dont_save: true,
+            };
+            let mut dangerous_raw = Vec::new();
+            dangerous_raw.extend_from_slice(&POOL_META_FORMAT.to_le_bytes());
+            dangerous_raw.extend_from_slice(&POOL_META_V1_VERSION.to_le_bytes());
+            dangerous_raw.extend_from_slice(&rmp_serde::to_vec(&legacy).expect("legacy three-field tuple must serialize"));
+            let old_reader: LegacyPoolMeta =
+                rmp_serde::from_slice(&dangerous_raw[4..]).expect("legacy reader must decode source");
+            assert!(old_reader.dont_save, "the old reader would treat this source as permanently do-not-save");
+            assert_eq!(rmp::decode::read_array_len(&mut &dangerous_raw[4..]).expect("tuple length"), 3);
+            let source = match decode_pool_meta_replica(dangerous_raw.clone()) {
+                PoolMetaReplica::Valid {
+                    canonical,
+                    revision,
+                    committed: true,
+                    ..
+                } => {
+                    assert_eq!(revision.version, POOL_META_V1_VERSION);
+                    canonical
+                }
+                _ => panic!("current reader must recognize the legacy source"),
+            };
+            assert!(matches!(decode_pool_meta_replica(safe_v1.clone()),
+                PoolMetaReplica::Valid { canonical, revision, committed: true, .. }
+                    if canonical == source && revision.version == POOL_META_V1_VERSION));
+            let physical = disks(&fixture.store).await;
+            let source_heads = replicas(&dangerous_raw, OffsetDateTime::now_utc() + Duration::seconds(1)).await;
+            for (index, disk) in physical[0].iter().enumerate() {
+                fs::write(object_path(disk).join("xl.meta"), &source_heads[index])
+                    .await
+                    .expect("install legacy source on pool zero");
+            }
+            let reads = read_pool_meta_replicas(fixture.store.pools.clone(), true).await;
+            assert!(matches!(&reads[0].replica, PoolMetaReplica::Valid { raw, committed: true, .. } if raw == &dangerous_raw));
+            assert!(matches!(&reads[1].replica, PoolMetaReplica::Valid { raw, committed: true, .. } if raw == &safe_v1));
+            let before = snapshots(&fixture.store).await;
+            let identities = owner_repair_identity_heads(&fixture.store).await;
+            let guard = owner(&fixture.store).await;
+            let mut state = fixture.store.pool_meta_save_gate.lock().await;
+            let result = tokio::time::timeout(
+                std::time::Duration::from_secs(30),
+                fixture.store.heal_pool_meta_replica_under_fence(1, guard, &mut state),
+            )
+            .await
+            .expect("unsafe legacy heal must terminate");
+            drop(state);
+            let error = result.expect_err("native heal must reject a source whose old-reader dont_save is true");
+            assert!(
+                error.to_string().contains("dont_save"),
+                "rejection must identify the unsafe legacy field: {error}"
+            );
+            assert_eq!(
+                snapshots(&fixture.store).await,
+                before,
+                "rejection must preserve every original head and backup"
+            );
+            assert_eq!(
+                owner_repair_identity_heads(&fixture.store).await,
+                identities,
+                "rejection must not write identity shards"
+            );
+            let after = read_pool_meta_replicas(fixture.store.pools.clone(), true).await;
+            assert!(matches!(&after[0].replica, PoolMetaReplica::Valid { raw, .. } if raw == &dangerous_raw));
+            assert!(
+                matches!(&after[1].replica, PoolMetaReplica::Valid { raw, .. } if raw == &safe_v1),
+                "the dangerous source bytes must not be copied into the target pool"
+            );
+        },
+    )
+    .await;
+}
+
+#[tokio::test]
+#[serial_test::serial(pool_meta_version_env)]
+async fn pool_meta_native_heal_rejects_target_v1_dont_save_true_without_writes() {
+    temp_env::async_with_vars(
+        [
+            (rustfs_config::ENV_POOL_META_V2_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V2_FLEET_CONFIRMED, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_WRITE, Some("false")),
+            (rustfs_config::ENV_POOL_META_V3_FLEET_CONFIRMED, Some("false")),
+        ],
+        async {
+            let (fixture, safe_v1) = legacy_native_owner_fixture(POOL_META_V1_VERSION, false).await;
+            let meta = match decode_pool_meta_replica(safe_v1.clone()) {
+                PoolMetaReplica::Valid {
+                    meta, committed: true, ..
+                } => meta,
+                _ => panic!("baseline V1 metadata must decode"),
+            };
+            let statuses = meta
+                .pools
+                .iter()
+                .map(|pool| {
+                    assert!(pool.decommission.is_none(), "this regression fixture does not inject decommission state");
+                    LegacyPoolStatus {
+                        id: pool.id,
+                        cmd_line: pool.cmd_line.clone(),
+                        last_update: pool.last_update,
+                        decommission: None,
+                    }
+                })
+                .collect();
+            let legacy = LegacyPoolMeta {
+                version: POOL_META_V1_VERSION,
+                pools: statuses,
+                dont_save: true,
+            };
+            let mut dangerous_raw = Vec::new();
+            dangerous_raw.extend_from_slice(&POOL_META_FORMAT.to_le_bytes());
+            dangerous_raw.extend_from_slice(&POOL_META_V1_VERSION.to_le_bytes());
+            dangerous_raw.extend_from_slice(&rmp_serde::to_vec(&legacy).expect("legacy three-field tuple must serialize"));
+            let old_reader: LegacyPoolMeta =
+                rmp_serde::from_slice(&dangerous_raw[4..]).expect("legacy reader must decode target");
+            assert!(old_reader.dont_save, "the old reader interprets the target as do-not-save");
+            assert_eq!(rmp::decode::read_array_len(&mut &dangerous_raw[4..]).expect("tuple length"), 3);
+            let source_canonical = match decode_pool_meta_replica(safe_v1.clone()) {
+                PoolMetaReplica::Valid {
+                    canonical,
+                    revision,
+                    committed: true,
+                    ..
+                } => {
+                    assert_eq!(revision.version, POOL_META_V1_VERSION);
+                    canonical
+                }
+                _ => panic!("safe pool-zero source must decode"),
+            };
+            assert!(
+                matches!(decode_pool_meta_replica(dangerous_raw.clone()),
+                PoolMetaReplica::Valid { canonical, revision, committed: true, .. }
+                    if canonical == source_canonical && revision.version == POOL_META_V1_VERSION),
+                "the dangerous target must remain semantically equivalent to the safe source"
+            );
+            let physical = disks(&fixture.store).await;
+            let target_heads = replicas(&dangerous_raw, OffsetDateTime::now_utc() + Duration::seconds(1)).await;
+            for (index, disk) in physical[1].iter().enumerate() {
+                fs::write(object_path(disk).join("xl.meta"), &target_heads[index])
+                    .await
+                    .expect("install legacy dangerous target head");
+            }
+            let reads = read_pool_meta_replicas(fixture.store.pools.clone(), true).await;
+            assert!(matches!(&reads[0].replica, PoolMetaReplica::Valid { raw, committed: true, .. } if raw == &safe_v1));
+            assert!(matches!(&reads[1].replica, PoolMetaReplica::Valid { raw, committed: true, .. } if raw == &dangerous_raw));
+            let before = snapshots(&fixture.store).await;
+            let identities = owner_repair_identity_heads(&fixture.store).await;
+            let guard = owner(&fixture.store).await;
+            let mut state = fixture.store.pool_meta_save_gate.lock().await;
+            let result = tokio::time::timeout(
+                std::time::Duration::from_secs(30),
+                fixture.store.heal_pool_meta_replica_under_fence(1, guard, &mut state),
+            )
+            .await
+            .expect("unsafe legacy target heal must terminate");
+            drop(state);
+            let error = result.expect_err("native heal must reject a target whose old-reader dont_save is true");
+            assert!(
+                error.to_string().contains("dont_save"),
+                "rejection must identify the unsafe legacy field: {error}"
+            );
+            assert_eq!(
+                snapshots(&fixture.store).await,
+                before,
+                "rejection must preserve all physical heads and backups"
+            );
+            assert_eq!(
+                owner_repair_identity_heads(&fixture.store).await,
+                identities,
+                "rejection must not write identity shards"
+            );
+            let after = read_pool_meta_replicas(fixture.store.pools.clone(), true).await;
+            assert!(matches!(&after[0].replica, PoolMetaReplica::Valid { raw, .. } if raw == &safe_v1));
+            assert!(
+                matches!(&after[1].replica, PoolMetaReplica::Valid { raw, .. } if raw == &dangerous_raw),
+                "unsafe target evidence must not be overwritten by the safe source"
+            );
+        },
+    )
+    .await;
+}

crates/ecstore/src/core/pools.rs

@@ -94,7 +94,6 @@ use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::collections::{HashMap, HashSet};
 use std::fmt::Display;
-#[cfg(test)]
 use std::io::Cursor;
 use std::io::Write;
 use std::path::PathBuf;
@@ -6839,11 +6838,48 @@ impl Drop for PoolMetaTransactionArm {
     }
 }
 
+fn ensure_legacy_pool_meta_republish_safe(data: &[u8]) -> Result<()> {
+    if data.len() >= 4
+        && LittleEndian::read_u16(&data[2..4]) == POOL_META_V1_VERSION
+        // The selection already validates the payload. The runtime-only
+        // flag can only appear in the legacy three-field sequence or map.
+        && rmp::decode::read_array_len(&mut &data[4..])
+            .or_else(|_| rmp::decode::read_map_len(&mut &data[4..]))
+            .ok()
+            == Some(3)
+        && let Ok(legacy) = rmp_serde::from_slice::<LegacyPoolMeta>(&data[4..])
+        && legacy.dont_save
+    {
+        return Err(Error::other("pool metadata legacy source contains dont_save=true"));
+    }
+    Ok(())
+}
+
+// Ordinary legacy reads retain their compatibility behavior, but bytes that
+// will be copied to another pool must not contain an unparsed extension.
+fn ensure_legacy_pool_meta_payload_complete(data: &[u8]) -> Result<()> {
+    let payload = data
+        .get(4..)
+        .ok_or_else(|| Error::other("pool metadata legacy source is too short"))?;
+    let mut decoder = rmp_serde::Deserializer::new(Cursor::new(payload));
+    serde::de::IgnoredAny::deserialize(&mut decoder)
+        .map_err(|error| Error::other(format!("pool metadata legacy source is invalid: {error}")))?;
+    let payload_len = u64::try_from(payload.len()).map_err(|_| Error::other("pool metadata legacy source is too large"))?;
+    if decoder.get_ref().position() != payload_len {
+        return Err(Error::other("pool metadata legacy source has trailing bytes"));
+    }
+    Ok(())
+}
+
 #[derive(Debug)]
 struct PoolMetaSelection {
     meta: PoolMeta,
     revision: PoolMetaRevision,
     canonical: Option<Vec<u8>>,
+    // Legacy semantic comparison canonicalizes to V2; a heal must instead
+    // preserve verified raw bytes at the highest observed format floor.
+    legacy_raw: Option<Vec<u8>>,
+    legacy_has_unsafe_flag: bool,
     replica_state: PoolMetaReplicaState,
     cas_tokens: Vec<PoolMetaCasToken>,
     absent: bool,
@@ -7234,6 +7270,8 @@ fn select_pool_meta_replica_reads(reads: Vec<PoolMetaReplicaRead>) -> Result<Poo
             meta: PoolMeta::default(),
             revision: PoolMetaRevision::legacy(0),
             canonical: None,
+            legacy_raw: None,
+            legacy_has_unsafe_flag: false,
             replica_state: PoolMetaReplicaState {
                 needs_repair: false,
                 repair_write_safe: true,
@@ -7288,6 +7326,8 @@ fn select_pool_meta_replica_reads(reads: Vec<PoolMetaReplicaRead>) -> Result<Poo
             meta: selected.meta.clone(),
             revision: selected.revision,
             canonical: Some(selected.canonical.clone()),
+            legacy_raw: None,
+            legacy_has_unsafe_flag: false,
             replica_state: PoolMetaReplicaState {
                 needs_repair,
                 repair_write_safe,
@@ -7301,6 +7341,10 @@ fn select_pool_meta_replica_reads(reads: Vec<PoolMetaReplicaRead>) -> Result<Poo
 
     // Legacy V1/V2 has no durable generation. Pool zero remains the commit
     // record; divergent backups without it are ambiguous and fail closed.
+    let legacy_has_unsafe_flag = committed
+        .iter()
+        .any(|(_, raw, _)| ensure_legacy_pool_meta_republish_safe(raw).is_err());
+    let mut floor_raw = None;
     let mut selected: Option<(usize, Vec<u8>, PoolMetaCommittedCandidate)> = None;
     for (idx, raw, candidate) in committed {
         if let Some((selected_idx, selected_raw, selected_candidate)) = selected.as_ref() {
@@ -7314,19 +7358,36 @@ fn select_pool_meta_replica_reads(reads: Vec<PoolMetaReplicaRead>) -> Result<Poo
                 }
             } else {
                 needs_repair |= selected_raw != &raw;
+                if floor_raw.is_none()
+                    && candidate.revision.version == observed_version
+                    && raw
+                        .get(2..4)
+                        .is_some_and(|version| LittleEndian::read_u16(version) == observed_version)
+                {
+                    floor_raw = Some(raw);
+                }
             }
         } else {
             selected = Some((idx, raw, candidate));
         }
     }
-    let (_, _, mut selected) =
+    let (_, selected_raw, mut selected) =
         selected.ok_or_else(|| Error::other("pool metadata recovery required: no valid legacy replica is available"))?;
+    if selected.revision.version == observed_version
+        && selected_raw
+            .get(2..4)
+            .is_some_and(|version| LittleEndian::read_u16(version) == observed_version)
+    {
+        floor_raw = Some(selected_raw);
+    }
     selected.meta.version = observed_version;
     selected.revision.version = observed_version;
     Ok(PoolMetaSelection {
         meta: selected.meta,
         revision: selected.revision,
         canonical: Some(selected.canonical),
+        legacy_raw: floor_raw,
+        legacy_has_unsafe_flag,
         replica_state: PoolMetaReplicaState {
             needs_repair,
             repair_write_safe,
@@ -14546,6 +14607,23 @@ impl ECStore {
             .canonical
             .as_ref()
             .ok_or_else(|| Error::other("pool metadata heal has no committed source"))?;
+        if selection.generation_protocol_observed && !selection.revision.is_generation_protocol() {
+            return Err(Error::other("pool metadata heal requires recovery of the observed generation protocol"));
+        }
+        if selection.legacy_has_unsafe_flag {
+            return Err(Error::other("pool metadata legacy replica contains dont_save=true"));
+        }
+        let repair_data = if selection.revision.is_generation_protocol() {
+            canonical.clone()
+        } else {
+            let raw = selection
+                .legacy_raw
+                .clone()
+                .ok_or_else(|| Error::other("pool metadata heal cannot prove the legacy format floor"))?;
+            ensure_legacy_pool_meta_payload_complete(&raw)?;
+            raw
+        };
+        ensure_legacy_pool_meta_republish_safe(&repair_data)?;
         let pool = self
             .pools
             .get(pool_index)
@@ -14556,10 +14634,10 @@ impl ECStore {
             .get(pool_index)
             .ok_or_else(|| Error::other("pool metadata heal has no target revision"))?;
         let mut arm = state.arm_transaction();
-        save_pool_meta_object_cas(pool.clone(), POOL_META_NAME, canonical.clone(), token, &fence, "native_heal", &mut arm)
+        save_pool_meta_object_cas(pool.clone(), POOL_META_NAME, repair_data.clone(), token, &fence, "native_heal", &mut arm)
             .await?;
         let (confirmed, _) = read_config_no_lock_preserve_empty_with_metadata(pool, POOL_META_NAME).await?;
-        if confirmed != *canonical {
+        if confirmed != repair_data {
             return Err(Error::other("pool metadata native heal readback diverged"));
         }
         fence.ensure_held()?;
@@ -29482,6 +29560,53 @@ mod tests {
         );
     }
 
+    #[test]
+    fn legacy_republish_guard_checks_only_three_field_sequence_or_map() {
+        let frame = |payload: Vec<u8>| {
+            let mut data = Vec::new();
+            data.extend_from_slice(&POOL_META_FORMAT.to_le_bytes());
+            data.extend_from_slice(&POOL_META_V1_VERSION.to_le_bytes());
+            data.extend_from_slice(&payload);
+            data
+        };
+        let two_fields = PersistedPoolMetaV1 {
+            version: POOL_META_V1_VERSION,
+            pools: Vec::new(),
+        };
+        for payload in [rmp_serde::to_vec(&two_fields), rmp_serde::to_vec_named(&two_fields)] {
+            let data = frame(payload.expect("two-field V1 payload must serialize"));
+            assert!(matches!(decode_pool_meta_replica(data.clone()), PoolMetaReplica::Valid { .. }));
+            assert!(ensure_legacy_pool_meta_republish_safe(&data).is_ok());
+            assert!(ensure_legacy_pool_meta_payload_complete(&data).is_ok());
+            let mut trailing = data;
+            trailing.push(0xc0);
+            assert!(ensure_legacy_pool_meta_payload_complete(&trailing).is_err());
+        }
+        for dont_save in [false, true] {
+            let legacy = LegacyPoolMeta {
+                version: POOL_META_V1_VERSION,
+                pools: Vec::new(),
+                dont_save,
+            };
+            for payload in [rmp_serde::to_vec(&legacy), rmp_serde::to_vec_named(&legacy)] {
+                let data = frame(payload.expect("three-field V1 payload must serialize"));
+                assert!(matches!(decode_pool_meta_replica(data.clone()), PoolMetaReplica::Valid { .. }));
+                assert_eq!(ensure_legacy_pool_meta_republish_safe(&data).is_err(), dont_save);
+                assert!(ensure_legacy_pool_meta_payload_complete(&data).is_ok());
+                let mut trailing = data;
+                trailing.push(0xc0);
+                assert!(ensure_legacy_pool_meta_payload_complete(&trailing).is_err());
+            }
+        }
+        // This guard is only called for already-decoded replicas; malformed
+        // containers must fail selection rather than becoming repair sources.
+        let truncated = frame(vec![0x93, 0x01]);
+        assert!(matches!(
+            decode_pool_meta_replica(truncated),
+            PoolMetaReplica::Corrupt(_) | PoolMetaReplica::Incompatible(_)
+        ));
+    }
+
     #[test]
     fn pool_meta_decode_supports_legacy_payload() {
         let start_time = OffsetDateTime::now_utc();

crates/ecstore/src/set_disk/mod.rs

@@ -4696,6 +4696,17 @@ pub(crate) fn same_distributed_lock_domain(left: &[Arc<dyn LockClient>], right:
             .all(|right_client| left.iter().any(|left_client| Arc::ptr_eq(left_client, right_client)))
 }
 
+// Reusing a held writer additionally requires the same quorum, so repeated
+// clients must not give different weight to an otherwise identical domain.
+fn same_distributed_lock_quorum(left: &[Arc<dyn LockClient>], right: &[Arc<dyn LockClient>]) -> bool {
+    !left.is_empty()
+        && left.len() == right.len()
+        && left.iter().all(|client| {
+            left.iter().filter(|other| Arc::ptr_eq(client, other)).count() == 1
+                && right.iter().filter(|other| Arc::ptr_eq(client, other)).count() == 1
+        })
+}
+
 const ERASURE_CACHE_MAX_ENTRIES: usize = 32;
 
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
@@ -5645,9 +5656,9 @@ impl SetDisks {
     pub(crate) async fn shares_namespace_lock_domain(&self, other: &Self) -> bool {
         match (self.ctx.is_dist_erasure().await, other.ctx.is_dist_erasure().await) {
             (false, false) => Arc::ptr_eq(&self.local_lock_manager, &other.local_lock_manager),
-            (true, true) => {
-                self.set_lock_namespace == other.set_lock_namespace && same_distributed_lock_domain(&self.lockers, &other.lockers)
-            }
+            // Distributed lock clients receive the raw ObjectKey; the pool
+            // namespace is not part of the key stored by their backends.
+            (true, true) => same_distributed_lock_quorum(&self.lockers, &other.lockers),
             _ => false,
         }
     }
@@ -9882,7 +9893,7 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
-    async fn namespace_lock_domain_includes_pool_namespace() {
+    async fn distributed_namespace_lock_domain_follows_backend_not_pool_name() {
         let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
 
         let first: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
@@ -9921,11 +9932,25 @@ mod tests {
             "sets in the same pool share the object namespace lock domain"
         );
         assert!(
-            !same_pool_first_set.shares_namespace_lock_domain(&other_pool_set).await,
-            "different pool namespaces must not be deduplicated solely by identical clients"
+            same_pool_first_set.shares_namespace_lock_domain(&other_pool_set).await,
+            "different pools using identical backends lock the same raw object key"
         );
     }
 
+    #[test]
+    fn distributed_lock_domain_rejects_partial_or_duplicate_backends() {
+        let a: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
+        let b: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
+        let c: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
+        assert!(same_distributed_lock_quorum(&[a.clone(), b.clone()], &[b.clone(), a.clone()]));
+        assert!(!same_distributed_lock_quorum(&[a.clone(), b.clone()], &[a.clone(), c]));
+        assert!(!same_distributed_lock_quorum(
+            &[a.clone(), a.clone(), b.clone()],
+            &[a.clone(), b.clone(), b]
+        ));
+        assert!(!same_distributed_lock_quorum(&[], &[]));
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn streaming_reader_holds_read_lock_until_eof() {

crates/ecstore/src/store/heal.rs

@@ -3692,6 +3692,177 @@ mod tests {
         );
     }
 
+    // The RPC lock service uses the raw ObjectKey and one backend per node,
+    // even when the requesting sets carry different pool namespace strings.
+    async fn pool_meta_heal_store_with_lock_domains(source: &Arc<ECStore>, shared: bool) -> Arc<ECStore> {
+        use rustfs_lock::client::{LockClient, local::LocalClient};
+
+        let clients = || {
+            (0..2)
+                .map(|_| {
+                    Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new()))) as Arc<dyn LockClient>
+                })
+                .collect::<Vec<_>>()
+        };
+        let first = clients();
+        let second = if shared { first.clone() } else { clients() };
+        let mut pools = source.pools.clone();
+        for (pool, lockers) in pools.iter_mut().zip([first, second]) {
+            for set in &mut Arc::make_mut(pool).disk_set {
+                Arc::make_mut(set).lockers = lockers.clone();
+            }
+        }
+        let endpoints: EndpointServerPools = pools.iter().map(|pool| pool.endpoints.clone()).collect::<Vec<_>>().into();
+        let state = source.pool_meta_save_gate.lock().await.independent_clone_for_test();
+        Arc::new(ECStore {
+            id: Uuid::new_v4(),
+            disk_map: Default::default(),
+            pools,
+            peer_sys: crate::cluster::rpc::S3PeerSys::new_with_instance_ctx(&endpoints, Arc::clone(&source.ctx)),
+            pool_meta: tokio::sync::RwLock::new(source.pool_meta.read().await.clone()),
+            rebalance_meta: tokio::sync::RwLock::new(None),
+            decommission_cancelers: tokio::sync::RwLock::new(vec![None; 2]),
+            start_gate: tokio::sync::Mutex::new(()),
+            pool_meta_save_gate: tokio::sync::Mutex::new(state),
+            ctx: Arc::clone(&source.ctx),
+            bucket_fence_registry: Arc::default(),
+        })
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn pool_meta_same_distributed_backend_has_one_raw_lock_across_pools() {
+        let (_dirs, source, _peer) = test_two_pool_stores(None).await;
+        let store = pool_meta_heal_store_with_lock_domains(&source, true).await;
+        let first = store.pools[0].get_disks_by_key(POOL_META_NAME);
+        let second = store.pools[1].get_disks_by_key(POOL_META_NAME);
+        let first_lock = first
+            .new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME)
+            .await
+            .expect("first pool lock");
+        let second_lock = second
+            .new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME)
+            .await
+            .expect("second pool lock");
+        let guard = first_lock
+            .get_write_lock(std::time::Duration::from_secs(2))
+            .await
+            .expect("first writer");
+        assert!(
+            second_lock
+                .get_write_lock(std::time::Duration::from_millis(150))
+                .await
+                .is_err(),
+            "different pool namespaces still conflict on the same backend and object key"
+        );
+        drop(guard);
+        second_lock
+            .get_write_lock(std::time::Duration::from_secs(3))
+            .await
+            .expect("second writer should acquire after the first releases");
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn pool_meta_heal_reuses_same_distributed_backend_across_pools() {
+        let (_dirs, source, _peer) = test_two_pool_stores(None).await;
+        let store = pool_meta_heal_store_with_lock_domains(&source, true).await;
+        let (item, error) = tokio::time::timeout(
+            std::time::Duration::from_secs(3),
+            store.handle_heal_object(
+                RUSTFS_META_BUCKET,
+                POOL_META_NAME,
+                "",
+                &HealOpts {
+                    pool: Some(1),
+                    dry_run: true,
+                    ..Default::default()
+                },
+            ),
+        )
+        .await
+        .expect("inspection must reuse the fixed pool's writer when the lock backend is shared")
+        .expect("pool metadata heal inspection should return a result");
+        assert!(error.is_none(), "inspection should succeed: {error:?}");
+        assert_eq!(item.object, POOL_META_NAME);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn pool_meta_heal_repairs_other_pool_with_shared_distributed_backend() {
+        let (_dirs, source, _peer) = test_two_pool_stores(None).await;
+        let store = pool_meta_heal_store_with_lock_domains(&source, true).await;
+        let missing = remove_pool_meta_shard(&store, 1).await;
+        let (item, error) = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            store.handle_heal_object(
+                RUSTFS_META_BUCKET,
+                POOL_META_NAME,
+                "",
+                &HealOpts {
+                    pool: Some(1),
+                    ..Default::default()
+                },
+            ),
+        )
+        .await
+        .expect("cross-pool repair must not reenter the shared writer")
+        .expect("pool metadata repair should return a result");
+        assert!(error.is_none(), "cross-pool repair should complete: {error:?}");
+        assert_eq!(item.object, POOL_META_NAME);
+        assert!(
+            missing.read_xl(RUSTFS_META_BUCKET, POOL_META_NAME, false).await.is_ok(),
+            "the selected missing shard must be restored"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn pool_meta_heal_keeps_target_lock_for_independent_distributed_backend() {
+        let (_dirs, source, _peer) = test_two_pool_stores(None).await;
+        let store = pool_meta_heal_store_with_lock_domains(&source, false).await;
+        let target = store.pools[1].get_disks_by_key(POOL_META_NAME);
+        let lock = target
+            .new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME)
+            .await
+            .expect("target lock");
+        let guard = lock
+            .get_write_lock(std::time::Duration::from_secs(2))
+            .await
+            .expect("hold target lock");
+        let barrier = DecommissionCapacityLockOrderBarrier::install(store.id, store.id);
+        let heal_store = Arc::clone(&store);
+        let mut heal = tokio::spawn(async move {
+            heal_store
+                .handle_heal_object(
+                    RUSTFS_META_BUCKET,
+                    POOL_META_NAME,
+                    "",
+                    &HealOpts {
+                        pool: Some(1),
+                        dry_run: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+        tokio::time::timeout(
+            std::time::Duration::from_secs(3),
+            barrier.wait_until_external_heal_target_lock_attempted(),
+        )
+        .await
+        .expect("inspection must attempt the independent target lock");
+        assert!(!heal.is_finished(), "inspection must wait while the target writer is held");
+        drop(guard);
+        let (item, error) = tokio::time::timeout(std::time::Duration::from_secs(3), &mut heal)
+            .await
+            .expect("inspection should progress after target lock release")
+            .expect("heal task should not panic")
+            .expect("pool metadata heal inspection should return a result");
+        assert!(error.is_none(), "inspection should succeed: {error:?}");
+        assert_eq!(item.object, POOL_META_NAME);
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn pool_meta_heal_cas_repairs_only_the_requested_pool() {

crates/heal/src/heal/manager.rs

@@ -810,7 +810,14 @@ fn publish_heal_queue_length(queue: &PriorityHealQueue) {
 fn active_heal_for_dedup_key(active_heals: &HashMap<String, Arc<HealTask>>, key: &str) -> Option<(String, HealType)> {
     active_heals
         .iter()
-        .find(|(_, task)| PriorityHealQueue::make_dedup_key_for_scope(&task.heal_type, &task.options) == key)
+        .find(|(_, task)| {
+            PriorityHealQueue::make_dedup_key_for_owner(
+                &task.heal_type,
+                &task.options,
+                task.source,
+                task.expected_mrf_bucket_incarnation_id,
+            ) == key
+        })
         .map(|(task_id, task)| (task_id.clone(), task.heal_type.clone()))
 }
 

crates/heal/src/heal/manager/queue.rs

@@ -611,7 +611,104 @@ impl PriorityHealQueue {
 
     /// Create a deduplication key from a heal request
     pub(super) fn make_dedup_key(request: &HealRequest) -> String {
-        Self::make_dedup_key_for_scope(&request.heal_type, &request.options)
+        Self::make_dedup_key_for_owner(
+            &request.heal_type,
+            &request.options,
+            request.source,
+            request.expected_mrf_bucket_incarnation_id,
+        )
+    }
+
+    /// Bound MRF work can only merge with work for the same bucket lifetime.
+    /// Length-prefix user-controlled components so a key cannot forge another
+    /// version, scope, or incarnation through delimiter characters.
+    pub(super) fn make_dedup_key_for_owner(
+        heal_type: &HealType,
+        options: &HealOptions,
+        source: HealRequestSource,
+        incarnation: Option<Uuid>,
+    ) -> String {
+        if source == HealRequestSource::Mrf
+            && let Some(incarnation) = incarnation
+        {
+            use std::fmt::Write as _;
+
+            let (pool, set) = heal_scope_indices(heal_type, options);
+            let mut key = format!("mrf-bound-v1:{incarnation}:{pool:?}:{set:?}:");
+            let mut part = |value: &str| {
+                // Formatting into a String is infallible.
+                let _ = write!(key, "{}:{value}", value.len());
+            };
+            match heal_type {
+                HealType::Object {
+                    bucket,
+                    object,
+                    version_id,
+                } => {
+                    part("object");
+                    part(bucket);
+                    part(object);
+                    match version_id {
+                        None => part("none"),
+                        Some(version_id) => {
+                            part("some");
+                            part(version_id);
+                        }
+                    }
+                }
+                HealType::Metadata { bucket, object } => {
+                    part("metadata");
+                    part(bucket);
+                    part(object);
+                }
+                HealType::ECDecode {
+                    bucket,
+                    object,
+                    version_id,
+                } => {
+                    part("ecdecode");
+                    part(bucket);
+                    part(object);
+                    match version_id {
+                        None => part("none"),
+                        Some(version_id) => {
+                            part("some");
+                            part(version_id);
+                        }
+                    }
+                }
+                HealType::DeleteMarkerPurge {
+                    bucket,
+                    object,
+                    version_id,
+                    purge,
+                } => {
+                    part("delete-marker-purge");
+                    part(bucket);
+                    part(object);
+                    part(version_id);
+                    part(&purge.bucket_incarnation_id.to_string());
+                    part(&purge.marker_incarnation_id.to_string());
+                    part(&base64_simd::URL_SAFE_NO_PAD.encode_to_string(purge.marker_identity));
+                }
+                HealType::Bucket { bucket } => {
+                    part("bucket");
+                    part(bucket);
+                }
+                HealType::Prefix { bucket, prefix } => {
+                    part("prefix");
+                    part(bucket);
+                    part(prefix);
+                }
+                // Cluster and erasure-set work are not bucket-bound; preserve
+                // their existing keys, including contains_erasure_set lookups.
+                HealType::Cluster | HealType::ErasureSet { .. } => {
+                    return Self::make_dedup_key_for_scope(heal_type, options);
+                }
+            }
+            return key;
+        }
+        Self::make_dedup_key_for_scope(heal_type, options)
     }
 
     pub(super) fn make_dedup_key_for_scope(heal_type: &HealType, options: &HealOptions) -> String {

crates/heal/src/heal/manager/tests.rs

@@ -117,6 +117,453 @@ async fn process_manager_queue_once(manager: &HealManager) {
 
 struct MockStorage;
 
+fn incarnation_bound_mrf_fallback(incarnation: Uuid) -> HealRequest {
+    let mut request = HealRequest::object("mrf-recreated-bucket".to_owned(), "object".to_owned(), None);
+    request.source = HealRequestSource::Mrf;
+    request.expected_mrf_bucket_incarnation_id = Some(incarnation);
+    request.options.pool_index = Some(0);
+    request.options.set_index = Some(0);
+    request
+}
+
+async fn assert_recreated_bucket_fallback_has_its_own_owner(existing_state: &str) {
+    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::clone(&storage), None);
+    let old_incarnation = Uuid::new_v4();
+    let new_incarnation = Uuid::new_v4();
+    let old_request = incarnation_bound_mrf_fallback(old_incarnation);
+    let old_id = old_request.id.clone();
+    match existing_state {
+        "queued" => {
+            let receipt = manager
+                .submit_heal_request_with_receipt(old_request)
+                .await
+                .expect("old fallback must be admitted");
+            assert_eq!(receipt.result, HealAdmissionResult::Accepted);
+        }
+        "active" => {
+            manager
+                .active_heals
+                .lock()
+                .await
+                .insert(old_id.clone(), Arc::new(HealTask::from_request(old_request, Arc::clone(&storage))));
+        }
+        "retrying" => {
+            manager.retrying_heals.lock().await.insert(
+                old_id.clone(),
+                RetryingHeal {
+                    request: old_request,
+                    error: "retry pending".to_owned(),
+                    cancel_token: CancellationToken::new(),
+                },
+            );
+        }
+        other => panic!("unknown fixture state: {other}"),
+    }
+
+    let new_request = incarnation_bound_mrf_fallback(new_incarnation);
+    let new_id = new_request.id.clone();
+    assert_ne!(new_id, old_id, "the fallback must use a new request ID");
+    let receipt = manager
+        .submit_heal_request_with_receipt(new_request)
+        .await
+        .expect("new incarnation admission must be decided");
+    assert_eq!(
+        receipt.result,
+        HealAdmissionResult::Accepted,
+        "a {existing_state} fallback for an older bucket incarnation cannot own the new repair"
+    );
+    assert_eq!(receipt.task_id, new_id, "the new incarnation must retain its own task identity");
+    let queue = manager.heal_queue.lock().await;
+    assert!(
+        queue
+            .requests()
+            .any(|request| { request.id == new_id && request.expected_mrf_bucket_incarnation_id == Some(new_incarnation) })
+    );
+    if existing_state == "queued" {
+        assert!(
+            queue
+                .requests()
+                .any(|request| { request.id == old_id && request.expected_mrf_bucket_incarnation_id == Some(old_incarnation) })
+        );
+    }
+}
+
+#[tokio::test]
+async fn mrf_fallback_recreated_bucket_does_not_merge_into_queued_old_incarnation() {
+    assert_recreated_bucket_fallback_has_its_own_owner("queued").await;
+}
+
+#[tokio::test]
+async fn mrf_fallback_recreated_bucket_does_not_merge_into_active_old_incarnation() {
+    assert_recreated_bucket_fallback_has_its_own_owner("active").await;
+}
+
+#[tokio::test]
+async fn mrf_fallback_recreated_bucket_does_not_merge_into_retrying_old_incarnation() {
+    assert_recreated_bucket_fallback_has_its_own_owner("retrying").await;
+}
+
+fn incarnation_bound_mrf_fallback_for_version(incarnation: Uuid, version_id: Option<&str>) -> HealRequest {
+    let mut request = incarnation_bound_mrf_fallback(incarnation);
+    if let HealType::Object { version_id: id, .. } = &mut request.heal_type {
+        *id = version_id.map(str::to_owned);
+    } else {
+        unreachable!("fallback fixture must be an object request");
+    }
+    request
+}
+
+#[tokio::test]
+async fn versioned_mrf_fallbacks_merge_into_queued_owner() {
+    for version in [Uuid::nil(), Uuid::new_v4()] {
+        let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+        let incarnation = Uuid::new_v4();
+        let version_id = version.to_string();
+        let original = incarnation_bound_mrf_fallback_for_version(incarnation, Some(&version_id));
+        let original_id = original.id.clone();
+        let accepted = manager
+            .submit_heal_request_with_receipt(original)
+            .await
+            .expect("first queued fallback");
+        assert_eq!(accepted.result, HealAdmissionResult::Accepted);
+        let merged = manager
+            .submit_heal_request_with_receipt(incarnation_bound_mrf_fallback_for_version(incarnation, Some(&version_id)))
+            .await
+            .expect("duplicate queued fallback");
+        assert_eq!(merged.result, HealAdmissionResult::Merged);
+        assert_eq!(merged.task_id, original_id);
+        assert_eq!(manager.heal_queue.lock().await.len(), 1);
+    }
+}
+
+#[tokio::test]
+async fn mrf_fallback_same_incarnation_merges_but_version_and_scope_remain_distinct() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let incarnation = Uuid::new_v4();
+    let original = incarnation_bound_mrf_fallback(incarnation);
+    let original_id = original.id.clone();
+    assert_eq!(
+        manager
+            .submit_heal_request_with_receipt(original)
+            .await
+            .expect("first fallback")
+            .result,
+        HealAdmissionResult::Accepted
+    );
+    let same = manager
+        .submit_heal_request_with_receipt(incarnation_bound_mrf_fallback(incarnation))
+        .await
+        .expect("same-incarnation fallback");
+    assert_eq!(same.result, HealAdmissionResult::Merged);
+    assert_eq!(same.task_id, original_id);
+
+    let mut versioned = incarnation_bound_mrf_fallback(incarnation);
+    versioned.heal_type = HealType::Object {
+        bucket: "mrf-recreated-bucket".to_owned(),
+        object: "object".to_owned(),
+        version_id: Some(Uuid::new_v4().to_string()),
+    };
+    assert_eq!(
+        manager
+            .submit_heal_request_with_receipt(versioned)
+            .await
+            .expect("different version")
+            .result,
+        HealAdmissionResult::Accepted
+    );
+    let mut other_scope = incarnation_bound_mrf_fallback(incarnation);
+    other_scope.options.pool_index = Some(1);
+    assert_eq!(
+        manager
+            .submit_heal_request_with_receipt(other_scope)
+            .await
+            .expect("different pool")
+            .result,
+        HealAdmissionResult::Accepted
+    );
+    let mut same_id_other_incarnation = incarnation_bound_mrf_fallback(Uuid::new_v4());
+    same_id_other_incarnation.id = original_id;
+    assert_eq!(
+        manager
+            .submit_heal_request_with_receipt(same_id_other_incarnation)
+            .await
+            .expect("same-ID conflict must be decided")
+            .result,
+        HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+    );
+}
+
+#[test]
+fn mrf_bound_dedup_keys_separate_incarnation_source_and_untrusted_components() {
+    let incarnation = Uuid::new_v4();
+    let bound = incarnation_bound_mrf_fallback(incarnation);
+    let mut unbound = bound.clone();
+    unbound.expected_mrf_bucket_incarnation_id = None;
+    let mut non_mrf = bound.clone();
+    non_mrf.source = HealRequestSource::ReadRepair;
+    assert_ne!(PriorityHealQueue::make_dedup_key(&bound), PriorityHealQueue::make_dedup_key(&unbound));
+    assert_eq!(
+        PriorityHealQueue::make_dedup_key(&non_mrf),
+        PriorityHealQueue::make_dedup_key_for_scope(&non_mrf.heal_type, &non_mrf.options),
+        "an unrelated source retains its previous scope key even if the MRF-only field is set"
+    );
+    assert_eq!(PriorityHealQueue::make_dedup_key(&non_mrf), PriorityHealQueue::make_dedup_key(&unbound));
+
+    let mut left = bound.clone();
+    left.heal_type = HealType::Object {
+        bucket: "mrf-recreated-bucket".to_owned(),
+        object: "part:a".to_owned(),
+        version_id: Some("b".to_owned()),
+    };
+    let mut right = bound.clone();
+    right.heal_type = HealType::Object {
+        bucket: "mrf-recreated-bucket".to_owned(),
+        object: "part".to_owned(),
+        version_id: Some("a:b".to_owned()),
+    };
+    assert_eq!(
+        PriorityHealQueue::make_dedup_key_for_scope(&left.heal_type, &left.options),
+        PriorityHealQueue::make_dedup_key_for_scope(&right.heal_type, &right.options),
+        "the legacy delimiter-only key demonstrates the collision"
+    );
+    assert_ne!(
+        PriorityHealQueue::make_dedup_key(&left),
+        PriorityHealQueue::make_dedup_key(&right),
+        "length-prefixed object and version components must not alias"
+    );
+    let mut forged = unbound.clone();
+    forged.heal_type = HealType::Object {
+        bucket: "mrf-recreated-bucket".to_owned(),
+        object: PriorityHealQueue::make_dedup_key(&bound),
+        version_id: None,
+    };
+    assert_ne!(PriorityHealQueue::make_dedup_key(&bound), PriorityHealQueue::make_dedup_key(&forged));
+}
+
+#[test]
+fn mrf_bound_optional_versions_do_not_alias_empty_or_forged_components() {
+    let incarnation = Uuid::new_v4();
+    for decode in [false, true] {
+        let make_request = |object: &str, version_id: Option<&str>| {
+            let mut request = incarnation_bound_mrf_fallback(incarnation);
+            let bucket = "mrf-recreated-bucket".to_owned();
+            let object = object.to_owned();
+            let version_id = version_id.map(str::to_owned);
+            request.heal_type = if decode {
+                HealType::ECDecode {
+                    bucket,
+                    object,
+                    version_id,
+                }
+            } else {
+                HealType::Object {
+                    bucket,
+                    object,
+                    version_id,
+                }
+            };
+            request
+        };
+        let absent = make_request("object", None);
+        let empty = make_request("object", Some(""));
+        let forged_object = make_request("object4:some0:", None);
+        let forged_version = make_request("object", Some("4:none"));
+        let keys = [&absent, &empty, &forged_object, &forged_version].map(PriorityHealQueue::make_dedup_key);
+        for (index, key) in keys.iter().enumerate() {
+            assert!(keys.iter().skip(index + 1).all(|other| key != other));
+        }
+
+        let mut queue = PriorityHealQueue::new();
+        assert_eq!(queue.push(absent.clone()), QueuePushOutcome::Accepted);
+        assert_eq!(queue.push(empty.clone()), QueuePushOutcome::Accepted);
+        assert_eq!(queue.push(make_request("object", None)), QueuePushOutcome::Merged);
+        assert_eq!(queue.push(make_request("object", Some(""))), QueuePushOutcome::Merged);
+        assert_eq!(queue.len(), 2);
+
+        for source in [HealRequestSource::Mrf, HealRequestSource::ReadRepair] {
+            let mut legacy_absent = absent.clone();
+            let mut legacy_empty = empty.clone();
+            legacy_absent.source = source;
+            legacy_empty.source = source;
+            if source == HealRequestSource::Mrf {
+                legacy_absent.expected_mrf_bucket_incarnation_id = None;
+                legacy_empty.expected_mrf_bucket_incarnation_id = None;
+            }
+            assert_eq!(
+                PriorityHealQueue::make_dedup_key(&legacy_absent),
+                PriorityHealQueue::make_dedup_key(&legacy_empty),
+                "unbound and non-MRF work retain the legacy empty-version key"
+            );
+        }
+    }
+}
+
+#[test]
+fn mrf_bound_queue_releases_each_incarnation_representative_independently() {
+    let mut queue = PriorityHealQueue::new();
+    let old = incarnation_bound_mrf_fallback(Uuid::new_v4());
+    let old_id = old.id.clone();
+    let old_key = PriorityHealQueue::make_dedup_key(&old);
+    let new = incarnation_bound_mrf_fallback(Uuid::new_v4());
+    let new_id = new.id.clone();
+    let new_key = PriorityHealQueue::make_dedup_key(&new);
+    let mut forced_old = old.clone();
+    forced_old.id = Uuid::new_v4().to_string();
+    forced_old.force_start = true;
+    let forced_id = forced_old.id.clone();
+
+    assert_eq!(queue.push(old), QueuePushOutcome::Accepted);
+    assert_eq!(queue.push(new), QueuePushOutcome::Accepted);
+    assert_eq!(queue.push(forced_old), QueuePushOutcome::Accepted);
+    assert_eq!(queue.queued_request_id_for_dedup_key(&old_key), Some(old_id.as_str()));
+    assert_eq!(queue.queued_request_id_for_dedup_key(&new_key), Some(new_id.as_str()));
+    queue
+        .remove_request_id(&old_id)
+        .expect("remove the old incarnation representative");
+    assert_eq!(queue.queued_request_id_for_dedup_key(&old_key), Some(forced_id.as_str()));
+    assert_eq!(queue.queued_request_id_for_dedup_key(&new_key), Some(new_id.as_str()));
+    assert_eq!(queue.pop_next().expect("new incarnation stays queued").id, new_id);
+    assert!(queue.queued_request_id_for_dedup_key(&new_key).is_none());
+    assert_eq!(queue.pop_next().expect("forced old owner stays queued").id, forced_id);
+    assert!(queue.queued_request_id_for_dedup_key(&old_key).is_none());
+}
+
+#[tokio::test]
+async fn durable_mrf_recreated_bucket_is_not_merged_into_queued_old_fallback() {
+    use rustfs_common::mrf_channel::{MrfIngressResult, MrfIntent, MrfKind, MrfScope, try_rearm_mrf_replay_intent};
+
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let old = incarnation_bound_mrf_fallback(Uuid::new_v4());
+    assert_eq!(
+        manager
+            .submit_heal_request_with_receipt(old)
+            .await
+            .expect("old fallback")
+            .result,
+        HealAdmissionResult::Accepted
+    );
+    let incarnation = Uuid::new_v4();
+    let request = incarnation_bound_mrf_fallback(incarnation);
+    let request_id = request.id.clone();
+    let mut intent = MrfIntent {
+        bucket: Arc::from("mrf-recreated-bucket"),
+        object: Arc::from("object"),
+        version_id: None,
+        kind: MrfKind::PartialWrite,
+        delete_marker_purge: None,
+        scope: Some(MrfScope {
+            pool_index: 0,
+            set_index: 0,
+        }),
+        lease: None,
+        enqueued_at_ms: 1,
+        attempts: 0,
+    };
+    assert_eq!(try_rearm_mrf_replay_intent(&mut intent), MrfIngressResult::Enqueued);
+    let anchor = rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(&intent, incarnation).expect("durable identity");
+    let target = MrfRepairNoticeTarget {
+        bucket: Arc::clone(&intent.bucket),
+        object: Arc::clone(&intent.object),
+        version_id: intent.version_id,
+        kind: intent.kind,
+        scope: intent.scope,
+        delete_marker_purge: None,
+        lease: intent.lease,
+        durable_anchor: Some(anchor.clone()),
+    };
+    let receipt = manager
+        .submit_mrf_heal_request_with_receipt_and_identity(request, target)
+        .await
+        .expect("durable admission must be decided");
+    assert_eq!(
+        receipt.result,
+        HealAdmissionResult::Accepted,
+        "a durable new incarnation needs its own owner"
+    );
+    assert_eq!(receipt.task_id, request_id);
+    assert!(
+        manager
+            .heal_queue
+            .lock()
+            .await
+            .requests()
+            .any(|queued| { queued.id == request_id && queued.expected_mrf_bucket_incarnation_id == Some(incarnation) })
+    );
+    assert!(
+        manager
+            .try_in_flight_durable_mrf_anchors()
+            .expect("stable owner snapshot")
+            .contains(&anchor)
+    );
+}
+
+#[tokio::test]
+async fn durable_mrf_same_incarnation_merge_retains_the_durable_anchor() {
+    use rustfs_common::mrf_channel::{MrfIngressResult, MrfIntent, MrfKind, MrfScope, try_rearm_mrf_replay_intent};
+
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let incarnation = Uuid::new_v4();
+    let object = format!("same-generation-{}", Uuid::new_v4());
+    let make_request = || {
+        let mut request = incarnation_bound_mrf_fallback(incarnation);
+        request.heal_type = HealType::Object {
+            bucket: "mrf-recreated-bucket".to_owned(),
+            object: object.clone(),
+            version_id: None,
+        };
+        request
+    };
+    let old = make_request();
+    let old_id = old.id.clone();
+    assert_eq!(
+        manager
+            .submit_heal_request_with_receipt(old)
+            .await
+            .expect("old fallback")
+            .result,
+        HealAdmissionResult::Accepted
+    );
+    let mut intent = MrfIntent {
+        bucket: Arc::from("mrf-recreated-bucket"),
+        object: Arc::from(object.as_str()),
+        version_id: None,
+        kind: MrfKind::PartialWrite,
+        delete_marker_purge: None,
+        scope: Some(MrfScope {
+            pool_index: 0,
+            set_index: 0,
+        }),
+        lease: None,
+        enqueued_at_ms: 1,
+        attempts: 0,
+    };
+    assert_eq!(try_rearm_mrf_replay_intent(&mut intent), MrfIngressResult::Enqueued);
+    let anchor = rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(&intent, incarnation).expect("durable identity");
+    let target = MrfRepairNoticeTarget {
+        bucket: Arc::clone(&intent.bucket),
+        object: Arc::clone(&intent.object),
+        version_id: intent.version_id,
+        kind: intent.kind,
+        scope: intent.scope,
+        delete_marker_purge: None,
+        lease: intent.lease,
+        durable_anchor: Some(anchor.clone()),
+    };
+    let merged = manager
+        .submit_mrf_heal_request_with_receipt_and_identity(make_request(), target)
+        .await
+        .expect("durable same-incarnation admission");
+    assert_eq!(merged.result, HealAdmissionResult::Merged);
+    assert_eq!(merged.task_id, old_id);
+    assert!(
+        manager
+            .try_in_flight_durable_mrf_anchors()
+            .expect("stable owner snapshot")
+            .contains(&anchor)
+    );
+}
+
 fn completed_retention_fixture(completed_at: SystemTime) -> CompletedHealStatus {
     CompletedHealStatus {
         task_id: "fixture-token".to_string(),

crates/heal/src/heal/task/heal_object.rs

@@ -480,8 +480,9 @@ impl HealTask {
                 message: format!("Missing source bucket incarnation for durable MRF repair {bucket}/{object}"),
             })?;
         if current_bucket_incarnation_id != bucket_incarnation_id {
-            return Err(Error::TaskExecutionFailed {
-                message: format!("Bucket incarnation changed before durable MRF repair {bucket}/{object}"),
+            return Err(Error::StaleBucketIncarnation {
+                bucket: bucket.to_owned(),
+                expected: Some(bucket_incarnation_id),
             });
         }
         expected.bucket_incarnation_id = Some(bucket_incarnation_id);

crates/heal/src/heal/task/tests.rs

@@ -3375,10 +3375,67 @@ async fn durable_mrf_heal_rejects_a_recreated_bucket_before_storage_heal() {
 
     let result = task.heal_object("bucket", "object", None).await;
 
-    assert!(matches!(result, Err(Error::TaskExecutionFailed { .. })));
+    assert!(matches!(
+        result,
+        Err(Error::StaleBucketIncarnation { expected: Some(id), .. }) if id == original_incarnation
+    ));
     assert!(storage.heal_object_calls.lock().unwrap().is_empty());
 }
 
+#[tokio::test]
+async fn durable_mrf_stale_incarnation_object_name_cannot_change_retry_classification() {
+    for object in ["timed out", "network error", "ordinary"] {
+        let original_incarnation = Uuid::new_v4();
+        let recreated_incarnation = Uuid::new_v4();
+        let storage = Arc::new(MockStorage {
+            bucket_incarnation_id: Mutex::new(Some(recreated_incarnation)),
+            ..Default::default()
+        });
+        let mut request = HealRequest::object("bucket".to_owned(), object.to_owned(), None);
+        request.source = HealRequestSource::Mrf;
+        request.expected_mrf_bucket_incarnation_id = Some(original_incarnation);
+        request.options.pool_index = Some(0);
+        request.options.set_index = Some(1);
+        let task = HealTask::from_request(request, storage.clone());
+
+        let error = task
+            .execute()
+            .await
+            .expect_err("a recreated bucket must reject the old MRF responsibility");
+        assert!(
+            !error.is_recoverable_heal(),
+            "user-controlled object names must not make a stale incarnation retryable: {object}"
+        );
+        assert!(
+            matches!(
+                &error,
+                Error::StaleBucketIncarnation { bucket, expected }
+                    if bucket == "bucket" && *expected == Some(original_incarnation)
+            ),
+            "the failure must retain the original bucket incarnation: {error:?}"
+        );
+        assert!(
+            matches!(crate::error::classify_heal_error(&error), crate::error::HealErrorClass::Aborted),
+            "stale responsibility must be classified independently of the object name"
+        );
+        assert!(
+            storage.heal_object_calls.lock().expect("object heal calls").is_empty(),
+            "an old responsibility must not reach storage repair in the recreated bucket"
+        );
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.counters.healed, 0, "a stale task cannot claim a repair");
+        assert!(
+            outcome.objects.iter().all(|item| !matches!(
+                &item.disposition,
+                HealObjectDisposition::Repaired
+                    | HealObjectDisposition::VerifiedHealthy
+                    | HealObjectDisposition::AuthoritativelyAbsent
+            )),
+            "the stale task must not produce a positive repair proof"
+        );
+    }
+}
+
 #[tokio::test(start_paused = true)]
 async fn read_repair_object_heal_is_not_failed_by_flat_task_timeout() {
     let storage = Arc::new(MockStorage {