fix(catalog): enumerate table owners without skipping namespaces

Health inferred pool metadata write authority from acquiring the local
authority gate within 100 ms. Sharing the gate among read-only
admissions still left held writers, and writers queued behind a reader
parked on the pool.bin fence, turning into cluster-write 503s.

Each writer now publishes the authority it leaves before releasing the
gate, and health reads that published state without acquiring it.
Transaction latches stay shared with the published copy; recovery
clears the latch and its failure under one lock so inspection never
observes a half-cleared state. The inspection-timeout path that can no
longer occur (check_timed_out, the check_timeout admin state, the
pool_metadata_check_timeout reason, the node timeout cache and its
metric) is removed.

crates/ecstore/src/disk/local.rs

@@ -977,7 +977,6 @@ async fn restore_delete_rollback_with_namespace_owner(
             .insert(backup_path.clone());
         os::rename_all_with_owner(&backup_path, xl_path, object_dir, publication_root, namespace_owner.clone()).await
     };
-    let restored_backup = restored.is_ok();
     match restored {
         Ok(()) => {}
         Err(DiskError::FileNotFound) if remove_new_metadata => {
@@ -1007,16 +1006,6 @@ async fn restore_delete_rollback_with_namespace_owner(
         os::fsync_dir_with_owner(object_dir, namespace_owner.clone())
             .await
             .map_err(to_file_error)?;
-    }
-    if restored_backup {
-        // Renaming two hardlinks to the same inode can leave the backup in place.
-        match os::remove_file_with_owner(&backup_path, namespace_owner.clone()).await {
-            Ok(()) => {}
-            Err(err) if err.kind() == ErrorKind::NotFound => {}
-            Err(err) => return Err(to_file_error(err).into()),
-        }
-    }
-    if effective_durability(volume).syncs_commit_metadata() {
         os::fsync_dir_with_owner(&rollback_path, namespace_owner.clone())
             .await
             .map_err(to_file_error)?;

crates/ecstore/src/disk/local/delete_recovery.rs

@@ -7240,9 +7240,6 @@ mod tests {
     #[cfg(unix)]
     #[tokio::test]
     async fn delete_recovery_fused_prepare_checkpoint_reopens_before_head_publication() {
-        use std::os::unix::fs::MetadataExt;
-
-        let _mode = durability_mode_override::set(DurabilityMode::Strict);
         let (_temp, disk, mut first, _) = fixture().await;
         first.version_id = Some(Uuid::new_v4());
         let mut second = first.clone();
@@ -7266,25 +7263,12 @@ mod tests {
             let source = source_root.join(transaction.to_string());
             let destination = copy_root.join(transaction.to_string());
             std::fs::create_dir(&destination).expect("create checkpoint transaction");
-            let source_head = source_root.join(STORAGE_FORMAT_FILE);
-            let source_head_meta = std::fs::metadata(&source_head).expect("stat source head");
-            let copy_head = copy_root.join(STORAGE_FORMAT_FILE);
-            std::fs::copy(&source_head, &copy_head).expect("capture head before publication");
+            std::fs::copy(source_root.join(STORAGE_FORMAT_FILE), copy_root.join(STORAGE_FORMAT_FILE))
+                .expect("capture head before publication");
             for entry in std::fs::read_dir(source).expect("read published preparation") {
                 let entry = entry.expect("prepared file");
                 assert!(entry.file_type().expect("prepared file type").is_file());
-                let dst = destination.join(entry.file_name());
-                if entry.file_name() == STORAGE_FORMAT_FILE_BACKUP {
-                    let backup_meta = entry.metadata().expect("stat prepared backup");
-                    assert_eq!(
-                        (backup_meta.dev(), backup_meta.ino()),
-                        (source_head_meta.dev(), source_head_meta.ino()),
-                        "fused preparation must hardlink backup to live head"
-                    );
-                    std::fs::hard_link(&copy_head, &dst).expect("preserve hardlinked backup");
-                } else {
-                    std::fs::copy(entry.path(), dst).expect("capture prepared file");
-                }
+                std::fs::copy(entry.path(), destination.join(entry.file_name())).expect("capture prepared file");
             }
         });
         let opts = DeleteOptions {
@@ -7308,42 +7292,10 @@ mod tests {
             .await
             .expect("actual preparation checkpoint must remain recoverable");
         assert_eq!(inspection.state, DeleteRecoveryState::NewPrepared);
-        let reopened_tx = reopened
-            .io_get_object_path("recovery", &format!("object/{transaction}"))
-            .expect("reopened transaction path");
-        os::fsync_dir_recorder::set_failure(&reopened_tx, ErrorKind::Other);
-        assert!(
-            reopened
-                .settle_delete_transaction(
-                    "recovery",
-                    "object",
-                    transaction,
-                    inspection.fingerprint,
-                    DeleteRecoveryAction::Rollback
-                )
-                .await
-                .is_err(),
-            "failed transaction directory sync after backup removal must retain journal"
-        );
-        assert!(reopened_tx.join(PREPARED).is_file());
-        assert!(!reopened_tx.join(STORAGE_FORMAT_FILE_BACKUP).exists());
-        let retry_inspection = reopened
-            .delete_recovery_inspect("recovery", "object", transaction)
-            .await
-            .expect("inspect interrupted rollback");
-        assert_eq!(retry_inspection.state, DeleteRecoveryState::NewPrepared);
-        let settled = reopened
-            .settle_delete_transaction(
-                "recovery",
-                "object",
-                transaction,
-                retry_inspection.fingerprint,
-                DeleteRecoveryAction::Rollback,
-            )
+        reopened
+            .settle_delete_transaction("recovery", "object", transaction, inspection.fingerprint, DeleteRecoveryAction::Rollback)
             .await
             .expect("roll back interrupted preparation");
-        assert_eq!(settled.state, DeleteRecoveryState::Absent, "same-inode backup must not survive rollback");
-        assert!(!checkpoint_root.join(transaction.to_string()).exists());
         assert_eq!(
             fs::read(checkpoint_root.join(STORAGE_FORMAT_FILE))
                 .await

crates/ecstore/src/services/tier/tier.rs

@@ -93,7 +93,6 @@ use super::{
         delete_tier_mutation_intent_record_if_current, list_tier_coordinator_mutation_intent_records,
         list_tier_mutation_intent_records, load_tier_coordinator_mutation_intent_record_with_etag,
         load_tier_mutation_intent_record_with_etag, save_tier_coordinator_mutation_intent_record_if_absent,
-        save_tier_mutation_intent_record_if_current,
     },
     warm_backend::WarmBackendImpl,
 };
@@ -621,11 +620,7 @@ struct CommittedMutationRecoveryPlan {
 
 impl TierMutationRecoverySnapshot {
     fn requires_config_lock(&self) -> bool {
-        self.has_committed_mutation_blocks
-            || self.intents.iter().any(|recovered| {
-                recovered.has_coordinator_record
-                    || (recovered.has_peer_record && recovered.intent.state == TierMutationIntentState::Prepared)
-            })
+        self.has_committed_mutation_blocks || self.intents.iter().any(|recovered| recovered.has_coordinator_record)
     }
 
     fn allowance(&self) -> MutationBlockAllowance {
@@ -5107,9 +5102,6 @@ impl TierConfigMgr {
 
             Self::reconcile_terminal_dual_prefix_mutation_intents(api.clone(), &mut snapshot).await?;
             Self::recover_prepared_coordinator_mutation_intents(api.clone(), &mut snapshot.coordinator_intents).await?;
-            if config_lock.is_some() {
-                Self::recover_expired_peer_only_prepares(api.clone(), &snapshot).await?;
-            }
 
             snapshot = Self::load_and_reconcile_mutation_recovery_snapshot(handle, api.clone(), true).await?;
             if snapshot.requires_config_lock() && config_lock.is_none() {
@@ -5510,58 +5502,6 @@ impl TierConfigMgr {
         }
     }
 
-    async fn recover_expired_peer_only_prepares<S>(api: Arc<S>, snapshot: &TierMutationRecoverySnapshot) -> io::Result<()>
-    where
-        S: EcstoreObjectIO,
-    {
-        let now = i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos()).unwrap_or(i64::MAX);
-        let orphaned = snapshot
-            .intents
-            .iter()
-            .filter(|recovered| {
-                recovered.has_peer_record
-                    && !recovered.has_coordinator_record
-                    && recovered.intent.state == TierMutationIntentState::Prepared
-                    && Self::tier_mutation_tombstone_retention_elapsed(&recovered.intent, now)
-            })
-            .map(|recovered| recovered.intent.clone())
-            .collect::<Vec<_>>();
-        if orphaned.is_empty() {
-            return Ok(());
-        }
-
-        // The caller holds the tier-config namespace lock. Only the exact
-        // persisted candidate proves a commit; an old or different config
-        // leaves the peer fence intact.
-        let loaded = load_tier_config_for_recovery(api.clone()).await?;
-        let Some(config_etag) = loaded.etag else {
-            return Ok(());
-        };
-        for intent in orphaned {
-            if loaded.persisted_digest != Some(intent.candidate_digest) {
-                continue;
-            }
-            let _mutation_guard = acquire_tier_mutation_mutex(intent.mutation_id).await;
-            let (mut current, record_etag) =
-                match load_tier_mutation_intent_record_with_etag(api.clone(), intent.mutation_id).await {
-                    Ok(record) => record,
-                    Err(Error::ConfigNotFound) => continue,
-                    Err(err) => return Err(tier_mutation_replay_error(err)),
-                };
-            if current != intent {
-                continue;
-            }
-            current
-                .advance(TierMutationIntentState::Committed, Some(config_etag.clone()))
-                .map_err(io::Error::other)?;
-            match save_tier_mutation_intent_record_if_current(api.clone(), &current, &record_etag).await {
-                Ok(()) | Err(Error::PreconditionFailed) => {}
-                Err(err) => return Err(tier_mutation_replay_error(err)),
-            }
-        }
-        Ok(())
-    }
-
     async fn recover_prepared_coordinator_mutation_intents<S>(api: Arc<S>, intents: &mut [TierMutationIntent]) -> io::Result<()>
     where
         S: EcstoreObjectIO,
@@ -13439,208 +13379,6 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn reload_recovers_expired_peer_only_prepare_for_persisted_add() {
-        use crate::services::tier::tier_mutation_intent::save_tier_mutation_intent_record;
-
-        let store = Arc::new(CasConfigStore::default());
-        let previous = empty_mgr();
-        let mut candidate = empty_mgr();
-        candidate.tiers.insert("COLD-A".to_string(), build_rustfs_tier("COLD-A"));
-        let targets = build_tier_mutation_affected_targets(
-            TierMutationIntentKind::Add,
-            HashSet::from(["COLD-A".to_string()]),
-            &previous,
-            &candidate,
-        )
-        .expect("added tier should have a target");
-        let mut intent = build_coordinator_tier_mutation_intent(TierMutationIntentKind::Add, None, &candidate, targets)
-            .expect("intent should build")
-            .expect("add should require an intent");
-        intent.expires_at_unix_nanos = 1;
-        candidate
-            .save_tiering_config_if_current(store.clone(), None)
-            .await
-            .expect("candidate should persist");
-        save_tier_mutation_intent_record(store.clone(), &intent)
-            .await
-            .expect("peer prepare should persist without a coordinator record");
-
-        let manager = TierConfigMgr::new();
-        TIER_MUTATION_TEST_PEERS
-            .scope(Vec::new(), async {
-                TIER_DRIVER_TEST_FACTORY
-                    .scope(healthy_driver_factory(), async {
-                        TierConfigMgr::reload_handle_with(&manager, store.clone())
-                            .await
-                            .expect("matching saved candidate should recover");
-                        TierConfigMgr::acquire_operation_lease(&manager, "COLD-A")
-                            .await
-                            .expect("recovered tier should be readable");
-                    })
-                    .await;
-            })
-            .await;
-        assert!(
-            TierConfigMgr::load_tier_mutation_intents(store)
-                .await
-                .expect("peer scan should succeed")
-                .iter()
-                .all(|record| record.state != TierMutationIntentState::Prepared),
-            "the orphaned prepare must no longer block later reloads"
-        );
-    }
-
-    #[tokio::test]
-    async fn reload_keeps_peer_only_prepare_without_commit_proof() {
-        use crate::services::tier::tier_mutation_intent::{
-            save_tier_mutation_intent_record, tier_mutation_intent_record_object_name,
-        };
-
-        for case in [
-            "unexpired",
-            "recently_expired",
-            "different_candidate",
-            "missing_config",
-            "concurrent_change",
-        ] {
-            let store = Arc::new(CasConfigStore::default());
-            let previous = empty_mgr();
-            let mut candidate = empty_mgr();
-            candidate.tiers.insert("COLD-A".to_string(), build_rustfs_tier("COLD-A"));
-            let targets = build_tier_mutation_affected_targets(
-                TierMutationIntentKind::Add,
-                HashSet::from(["COLD-A".to_string()]),
-                &previous,
-                &candidate,
-            )
-            .expect("added tier should have a target");
-            let mut intent = build_coordinator_tier_mutation_intent(TierMutationIntentKind::Add, None, &candidate, targets)
-                .expect("intent should build")
-                .expect("add should require an intent");
-            intent.expires_at_unix_nanos = match case {
-                "unexpired" => tier_mutation_intent_expiry_unix_nanos(),
-                "recently_expired" => {
-                    i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos()).expect("test clock should fit")
-                        - 1_000_000_000
-                }
-                _ => 1,
-            };
-            if case == "different_candidate" {
-                intent.candidate_digest = [0x51; 32];
-            }
-            if case != "missing_config" {
-                candidate
-                    .save_tiering_config_if_current(store.clone(), None)
-                    .await
-                    .expect("candidate should persist");
-            }
-            save_tier_mutation_intent_record(store.clone(), &intent)
-                .await
-                .expect("peer prepare should persist");
-            let expected = if case == "concurrent_change" {
-                let mut changed = intent.clone();
-                changed.candidate_digest = [0x52; 32];
-                let object = tier_mutation_intent_record_object_name(intent.mutation_id).expect("record path should build");
-                store
-                    .rewrite_on_next_if_match(object, changed.encode().expect("changed intent should encode"))
-                    .await;
-                changed
-            } else {
-                intent
-            };
-
-            let manager = TierConfigMgr::new();
-            TIER_MUTATION_TEST_PEERS
-                .scope(Vec::new(), async {
-                    TierConfigMgr::reload_handle_with(&manager, store.clone())
-                        .await
-                        .expect("unproven prepare should remain fenced");
-                })
-                .await;
-            let error = match TierConfigMgr::acquire_operation_lease(&manager, "COLD-A").await {
-                Ok(_) => panic!("{case}: unproven prepare must block reads"),
-                Err(error) => error,
-            };
-            assert!(error.message.contains("being replaced"), "{case}: {error}");
-            let records = TierConfigMgr::load_tier_mutation_intents(store)
-                .await
-                .expect("peer scan should succeed");
-            assert_eq!(records, vec![expected], "{case}: prepared evidence must remain unchanged");
-        }
-    }
-
-    #[tokio::test]
-    async fn peer_only_prepare_recovery_retains_fence_until_peer_commit_succeeds() {
-        use crate::services::tier::tier_mutation_intent::save_tier_mutation_intent_record;
-
-        let store = Arc::new(CasConfigStore::default());
-        let previous = empty_mgr();
-        let mut candidate = empty_mgr();
-        candidate.tiers.insert("COLD-A".to_string(), build_rustfs_tier("COLD-A"));
-        let targets = build_tier_mutation_affected_targets(
-            TierMutationIntentKind::Add,
-            HashSet::from(["COLD-A".to_string()]),
-            &previous,
-            &candidate,
-        )
-        .expect("added tier should have a target");
-        let mut intent = build_coordinator_tier_mutation_intent(TierMutationIntentKind::Add, None, &candidate, targets)
-            .expect("intent should build")
-            .expect("add should require an intent");
-        intent.expires_at_unix_nanos = 1;
-        candidate
-            .save_tiering_config_if_current(store.clone(), None)
-            .await
-            .expect("candidate should persist");
-        save_tier_mutation_intent_record(store.clone(), &intent)
-            .await
-            .expect("peer prepare should persist");
-
-        let manager = TierConfigMgr::new();
-        let calls = Arc::new(Mutex::new(Vec::new()));
-        TIER_MUTATION_TEST_PEERS
-            .scope(vec![FakeTierMutationPeer::boxed("peer-a", calls.clone(), Err("offline"))], async {
-                TierConfigMgr::reload_handle_with(&manager, store.clone())
-                    .await
-                    .expect_err("failed peer commit must leave recovery incomplete");
-            })
-            .await;
-        let retained = TierConfigMgr::load_tier_mutation_intents(store.clone())
-            .await
-            .expect("peer scan should succeed");
-        assert_eq!(retained.len(), 1);
-        assert_eq!(retained[0].state, TierMutationIntentState::Committed);
-        let error = match TierConfigMgr::acquire_operation_lease(&manager, "COLD-A").await {
-            Ok(_) => panic!("failed peer commit must retain the fence"),
-            Err(error) => error,
-        };
-        assert!(error.message.contains("being replaced"), "{error}");
-
-        TIER_MUTATION_TEST_PEERS
-            .scope(
-                vec![FakeTierMutationPeer::boxed(
-                    "peer-a",
-                    calls.clone(),
-                    Ok(PeerTierMutationState::Committed),
-                )],
-                async {
-                    TIER_DRIVER_TEST_FACTORY
-                        .scope(healthy_driver_factory(), async {
-                            TierConfigMgr::reload_handle_with(&manager, store.clone())
-                                .await
-                                .expect("peer commit retry should recover");
-                            TierConfigMgr::acquire_operation_lease(&manager, "COLD-A")
-                                .await
-                                .expect("tier should be readable after the retry");
-                        })
-                        .await;
-                },
-            )
-            .await;
-        assert_eq!(lock_unpoisoned(&calls).len(), 2, "recovery should retry the peer commit");
-    }
-
     #[tokio::test]
     async fn coordinator_mutation_recovery_scan_blocks_new_operation_lease() {
         let store = Arc::new(CasConfigStore::default());