From 1ac1d883f4b126198024a0f690f169cd8b48fdfc Mon Sep 17 00:00:00 2001
From: Ramakrishna Chilaka
 <49393831+RamakrishnaChilaka@users.noreply.github.com>
Date: Mon, 2 Nov 2026 20:30:48 +0530
Subject: [PATCH 1/5] perf(ecstore): shard disk path lock registries with SWAR
 byte keys (#9127)

---
 crates/ecstore/src/disk/os.rs | 605 +++++++++++++++++++++++++++++-----
 1 file changed, 525 insertions(+), 80 deletions(-)

diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs
index 2b6839e38..15d03a1a9 100644
--- a/crates/ecstore/src/disk/os.rs
+++ b/crates/ecstore/src/disk/os.rs
@@ -1677,8 +1677,230 @@ const MAX_GLOBAL_FILE_SYNCS: usize = 512;
 #[cfg(test)]
 const TEST_GLOBAL_FILE_SYNCS: usize = 64;
 
+const PATH_LOCK_SHARDS: usize = 64;
+const PATH_LOCK_INITIAL_GC_THRESHOLD: usize = 64;
+
+#[cfg(unix)]
+#[derive(Clone, Copy, Default)]
+struct Xxh3SliceHasher(u64);
+
+#[cfg(unix)]
+impl std::hash::Hasher for Xxh3SliceHasher {
+    #[inline]
+    fn write(&mut self, bytes: &[u8]) {
+        self.0 = xxhash_rust::xxh3::xxh3_64(bytes);
+    }
+
+    #[inline]
+    fn write_usize(&mut self, _i: usize) {}
+
+    #[inline]
+    fn finish(&self) -> u64 {
+        self.0
+    }
+}
+
+#[cfg(unix)]
+type Xxh3SliceBuildHasher = std::hash::BuildHasherDefault<Xxh3SliceHasher>;
+
+#[cfg(unix)]
+#[inline]
+fn is_canonical_unix_key_bytes(bytes: &[u8]) -> bool {
+    if bytes.starts_with(b"./") {
+        return !bytes[1..].windows(2).any(|w| w == b"//") && !bytes[1..].windows(3).any(|w| w == b"/./");
+    }
+    const SLASH_MASK: u64 = 0x2f2f_2f2f_2f2f_2f2f;
+    const LOW_BITS: u64 = 0x0101_0101_0101_0101;
+    const HIGH_BITS: u64 = 0x8080_8080_8080_8080;
+
+    let mut i = 0usize;
+    let len = bytes.len();
+    let mut prev_s = 0u64;
+    while i + 8 <= len {
+        let mut buf = [0u8; 8];
+        buf.copy_from_slice(&bytes[i..i + 8]);
+        let w = u64::from_le_bytes(buf);
+        let x = w ^ SLASH_MASK;
+        let s = (x.wrapping_sub(LOW_BITS)) & !x & HIGH_BITS;
+        let s_prev1 = (s << 8) | (prev_s >> 56);
+        let s_prev2 = (s << 16) | (prev_s >> 48);
+        if (s & (s_prev1 | s_prev2)) != 0 {
+            let start = i.saturating_sub(2);
+            let window = &bytes[start..i + 8];
+            if window.windows(2).any(|pair| pair == b"//") || window.windows(3).any(|triple| triple == b"/./") {
+                return false;
+            }
+        }
+        prev_s = s;
+        i += 8;
+    }
+    let tail_start = i.saturating_sub(2);
+    let tail = &bytes[tail_start..];
+    !tail.windows(2).any(|pair| pair == b"//") && !tail.windows(3).any(|triple| triple == b"/./")
+}
+
+#[cfg(unix)]
+#[cold]
+fn canonical_unix_path_key_slow(path: &Path) -> std::borrow::Cow<'static, [u8]> {
+    use std::os::unix::ffi::OsStrExt;
+    let mut out = Vec::with_capacity(path.as_os_str().len());
+    for comp in path.components() {
+        match comp {
+            Component::RootDir => {
+                out.clear();
+                out.push(b'/');
+            }
+            Component::CurDir => {
+                if out.is_empty() {
+                    out.push(b'.');
+                }
+            }
+            Component::ParentDir => {
+                if out.len() > 1 || (out.len() == 1 && out[0] != b'/') {
+                    out.push(b'/');
+                }
+                out.extend_from_slice(b"..");
+            }
+            Component::Normal(part) => {
+                if out.len() > 1 || (out.len() == 1 && out[0] != b'/') {
+                    out.push(b'/');
+                }
+                out.extend_from_slice(part.as_bytes());
+            }
+            Component::Prefix(_) => {}
+        }
+    }
+    std::borrow::Cow::Owned(out)
+}
+
+#[cfg(unix)]
+#[inline]
+fn canonical_unix_path_key(path: &Path) -> std::borrow::Cow<'_, [u8]> {
+    use std::os::unix::ffi::OsStrExt;
+    let mut bytes = path.as_os_str().as_bytes();
+    while bytes.len() > 1 {
+        if bytes[bytes.len() - 1] == b'/' {
+            bytes = &bytes[..bytes.len() - 1];
+        } else if bytes.len() >= 2 && bytes[bytes.len() - 2] == b'/' && bytes[bytes.len() - 1] == b'.' {
+            bytes = if bytes.len() == 2 { b"/" } else { &bytes[..bytes.len() - 2] };
+        } else {
+            break;
+        }
+    }
+    if is_canonical_unix_key_bytes(bytes) {
+        return std::borrow::Cow::Borrowed(bytes);
+    }
+    canonical_unix_path_key_slow(path)
+}
+
+#[cfg(unix)]
+#[inline]
+fn unix_canonical_key_starts_with(path_key: &[u8], root_key: &[u8]) -> bool {
+    if root_key.is_empty() {
+        return true;
+    }
+    if root_key == b"/" {
+        return path_key.starts_with(b"/");
+    }
+    path_key.starts_with(root_key) && (path_key.len() == root_key.len() || path_key.get(root_key.len()) == Some(&b'/'))
+}
+
+struct PathLockShard<L> {
+    #[cfg(unix)]
+    entries: HashMap<Box<[u8]>, Weak<L>, Xxh3SliceBuildHasher>,
+    #[cfg(not(unix))]
+    entries: HashMap<PathBuf, Weak<L>>,
+    next_gc_threshold: usize,
+}
+
+impl<L> Default for PathLockShard<L> {
+    fn default() -> Self {
+        Self {
+            entries: HashMap::default(),
+            next_gc_threshold: PATH_LOCK_INITIAL_GC_THRESHOLD,
+        }
+    }
+}
+
+struct ShardedPathLockRegistry<L> {
+    shards: [Mutex<PathLockShard<L>>; PATH_LOCK_SHARDS],
+}
+
+impl<L> ShardedPathLockRegistry<L> {
+    fn new() -> Self {
+        Self {
+            shards: std::array::from_fn(|_| Mutex::new(PathLockShard::default())),
+        }
+    }
+
+    #[cfg(unix)]
+    #[inline]
+    fn get_or_create_by_key(&self, key: &[u8], init: impl FnOnce() -> L) -> Arc<L> {
+        let hash = xxhash_rust::xxh3::xxh3_64(key);
+        let shard_idx = ((hash >> 32) as usize) & (PATH_LOCK_SHARDS - 1);
+        let mut shard = self.shards[shard_idx].lock();
+        if let Some(existing) = shard.entries.get(key).and_then(Weak::upgrade) {
+            return existing;
+        }
+        let created = Arc::new(init());
+        let weak = Arc::downgrade(&created);
+        if let Some(slot) = shard.entries.get_mut(key) {
+            *slot = weak;
+            return created;
+        }
+        if shard.entries.len() >= shard.next_gc_threshold {
+            shard.entries.retain(|_, entry| entry.strong_count() > 0);
+            shard.next_gc_threshold = shard
+                .entries
+                .len()
+                .saturating_add(1)
+                .saturating_mul(2)
+                .max(PATH_LOCK_INITIAL_GC_THRESHOLD);
+        }
+        shard.entries.insert(key.into(), weak);
+        created
+    }
+
+    #[inline]
+    fn get_or_create(&self, path: &Path, init: impl FnOnce() -> L) -> Arc<L> {
+        #[cfg(unix)]
+        {
+            let key = canonical_unix_path_key(path);
+            self.get_or_create_by_key(key.as_ref(), init)
+        }
+        #[cfg(not(unix))]
+        {
+            use std::hash::{Hash, Hasher};
+            let mut hasher = std::collections::hash_map::DefaultHasher::new();
+            path.hash(&mut hasher);
+            let shard_idx = ((hasher.finish() >> 32) as usize) & (PATH_LOCK_SHARDS - 1);
+            let mut shard = self.shards[shard_idx].lock();
+            if let Some(existing) = shard.entries.get(path).and_then(Weak::upgrade) {
+                return existing;
+            }
+            let created = Arc::new(init());
+            let weak = Arc::downgrade(&created);
+            if let Some(slot) = shard.entries.get_mut(path) {
+                *slot = weak;
+                return created;
+            }
+            if shard.entries.len() >= shard.next_gc_threshold {
+                shard.entries.retain(|_, entry| entry.strong_count() > 0);
+                shard.next_gc_threshold = shard
+                    .entries
+                    .len()
+                    .saturating_add(1)
+                    .saturating_mul(2)
+                    .max(PATH_LOCK_INITIAL_GC_THRESHOLD);
+            }
+            shard.entries.insert(path.to_path_buf(), weak);
+            created
+        }
+    }
+}
+
 static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
-static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+static DISK_FILE_SYNC_LIMITERS: LazyLock<ShardedPathLockRegistry<Semaphore>> = LazyLock::new(ShardedPathLockRegistry::new);
 
 /// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
 /// configured with >1 threads, isolates device-bound fsync from the main
@@ -1742,12 +1964,10 @@ fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'stati
         f()
     })
 }
-static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
-    LazyLock::new(|| Mutex::new(HashMap::new()));
+static DISK_VOLUME_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<RwLock<()>>> = LazyLock::new(ShardedPathLockRegistry::new);
 type NamespaceMutationLock = AsyncMutex<()>;
-type NamespaceMutationLockRegistry = HashMap<PathBuf, Weak<NamespaceMutationLock>>;
-static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock<Mutex<NamespaceMutationLockRegistry>> =
-    LazyLock::new(|| Mutex::new(HashMap::new()));
+static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<NamespaceMutationLock>> =
+    LazyLock::new(ShardedPathLockRegistry::new);
 
 fn default_global_file_sync_limit(cpu_count: usize, max_blocking_threads: usize) -> usize {
     let cpu_scaled = cpu_count
@@ -1771,15 +1991,7 @@ fn global_file_sync_limit() -> usize {
 
 /// Reuse a disk's limiter across reconnects while detached sync calls still hold it.
 pub(crate) fn disk_file_sync_limiter(root: &Path) -> Arc<Semaphore> {
-    let mut limiters = DISK_FILE_SYNC_LIMITERS.lock();
-    limiters.retain(|_, limiter| limiter.strong_count() > 0);
-    if let Some(limiter) = limiters.get(root).and_then(Weak::upgrade) {
-        return limiter;
-    }
-
-    let limiter = Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS));
-    limiters.insert(root.to_path_buf(), Arc::downgrade(&limiter));
-    limiter
+    DISK_FILE_SYNC_LIMITERS.get_or_create(root, || Semaphore::new(MAX_PARALLEL_FILE_SYNCS))
 }
 
 // Keep warm draining workers per disk/lane and spin briefly in user space
@@ -2147,45 +2359,131 @@ pub(crate) fn enter_blocking_test_hook() -> BlockingTestHookGuard {
     BlockingTestHookGuard { queue }
 }
 
-static DISK_BLOCKING_QUEUES: LazyLock<Mutex<HashMap<PathBuf, Weak<DiskBlockingQueue>>>> =
-    LazyLock::new(|| Mutex::new(HashMap::new()));
+struct RegisteredDiskBlockingQueue {
+    #[cfg(not(unix))]
+    root: PathBuf,
+    #[cfg(unix)]
+    canonical_root: Box<[u8]>,
+    queue: Weak<DiskBlockingQueue>,
+}
+
+static DISK_BLOCKING_QUEUES: LazyLock<parking_lot::RwLock<Vec<RegisteredDiskBlockingQueue>>> =
+    LazyLock::new(|| parking_lot::RwLock::new(Vec::new()));
 static FALLBACK_BLOCKING_QUEUES: LazyLock<[Arc<DiskBlockingQueue>; FALLBACK_NAMESPACE_QUEUE_SHARDS]> =
     LazyLock::new(|| std::array::from_fn(|_| DiskBlockingQueue::new()));
 
 pub(crate) fn register_disk_blocking_queue(root: &Path, io_root: &Path) -> Arc<DiskBlockingQueue> {
-    let mut queues = DISK_BLOCKING_QUEUES.lock();
-    queues.retain(|_, q| q.strong_count() > 0);
-    let queue = queues.get(root).and_then(Weak::upgrade).unwrap_or_else(|| {
-        let queue = DiskBlockingQueue::new();
-        queues.insert(root.to_path_buf(), Arc::downgrade(&queue));
-        queue
-    });
-    if io_root != root {
-        queues.insert(io_root.to_path_buf(), Arc::downgrade(&queue));
+    #[cfg(unix)]
+    let root_key = canonical_unix_path_key(root);
+    #[cfg(unix)]
+    let io_root_key = canonical_unix_path_key(io_root);
+
+    let mut queues = DISK_BLOCKING_QUEUES.write();
+    queues.retain(|entry| entry.queue.strong_count() > 0);
+    let queue = queues
+        .iter()
+        .find_map(|entry| {
+            #[cfg(unix)]
+            let matches = entry.canonical_root.as_ref() == root_key.as_ref();
+            #[cfg(not(unix))]
+            let matches = entry.root == root;
+            matches.then(|| entry.queue.upgrade()).flatten()
+        })
+        .unwrap_or_else(|| {
+            let queue = DiskBlockingQueue::new();
+            queues.push(RegisteredDiskBlockingQueue {
+                #[cfg(not(unix))]
+                root: root.to_path_buf(),
+                #[cfg(unix)]
+                canonical_root: root_key.as_ref().into(),
+                queue: Arc::downgrade(&queue),
+            });
+            queue
+        });
+    #[cfg(unix)]
+    let io_matches_root = io_root_key.as_ref() == root_key.as_ref();
+    #[cfg(not(unix))]
+    let io_matches_root = io_root == root;
+    if !io_matches_root {
+        let weak = Arc::downgrade(&queue);
+        if let Some(existing) = queues.iter_mut().find(|entry| {
+            #[cfg(unix)]
+            {
+                entry.canonical_root.as_ref() == io_root_key.as_ref()
+            }
+            #[cfg(not(unix))]
+            {
+                entry.root == io_root
+            }
+        }) {
+            existing.queue = weak;
+        } else {
+            queues.push(RegisteredDiskBlockingQueue {
+                #[cfg(not(unix))]
+                root: io_root.to_path_buf(),
+                #[cfg(unix)]
+                canonical_root: io_root_key.as_ref().into(),
+                queue: weak,
+            });
+        }
     }
     queue
 }
 
 fn disk_blocking_queue_for_root(root: &Path) -> Arc<DiskBlockingQueue> {
-    register_disk_blocking_queue(root, root)
-}
-
-fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
+    #[cfg(unix)]
+    let root_key = canonical_unix_path_key(root);
     {
-        let queues = DISK_BLOCKING_QUEUES.lock();
-        for (root, weak) in queues.iter() {
-            if path.starts_with(root)
-                && let Some(queue) = weak.upgrade()
-            {
+        let queues = DISK_BLOCKING_QUEUES.read();
+        for entry in queues.iter() {
+            #[cfg(unix)]
+            let matches = entry.canonical_root.as_ref() == root_key.as_ref();
+            #[cfg(not(unix))]
+            let matches = entry.root == root;
+            if matches && let Some(queue) = entry.queue.upgrade() {
                 return queue;
             }
         }
     }
-    use std::hash::{Hash, Hasher};
-    let mut hasher = std::collections::hash_map::DefaultHasher::new();
-    path.hash(&mut hasher);
-    let idx = usize::try_from(hasher.finish() % (FALLBACK_NAMESPACE_QUEUE_SHARDS as u64)).unwrap_or(0);
-    Arc::clone(&FALLBACK_BLOCKING_QUEUES[idx])
+    register_disk_blocking_queue(root, root)
+}
+
+fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
+    #[cfg(unix)]
+    {
+        let path_key = canonical_unix_path_key(path);
+        {
+            let queues = DISK_BLOCKING_QUEUES.read();
+            for entry in queues.iter() {
+                if unix_canonical_key_starts_with(path_key.as_ref(), entry.canonical_root.as_ref())
+                    && let Some(queue) = entry.queue.upgrade()
+                {
+                    return queue;
+                }
+            }
+        }
+        let hash = xxhash_rust::xxh3::xxh3_64(path_key.as_ref());
+        let idx = usize::try_from(hash % (FALLBACK_NAMESPACE_QUEUE_SHARDS as u64)).unwrap_or(0);
+        Arc::clone(&FALLBACK_BLOCKING_QUEUES[idx])
+    }
+    #[cfg(not(unix))]
+    {
+        {
+            let queues = DISK_BLOCKING_QUEUES.read();
+            for entry in queues.iter() {
+                if path.starts_with(&entry.root)
+                    && let Some(queue) = entry.queue.upgrade()
+                {
+                    return queue;
+                }
+            }
+        }
+        use std::hash::{Hash, Hasher};
+        let mut hasher = std::collections::hash_map::DefaultHasher::new();
+        path.hash(&mut hasher);
+        let idx = usize::try_from(hasher.finish() % (FALLBACK_NAMESPACE_QUEUE_SHARDS as u64)).unwrap_or(0);
+        Arc::clone(&FALLBACK_BLOCKING_QUEUES[idx])
+    }
 }
 
 /// Serialize a bucket's local metadata commits with physical bucket removal.
@@ -2194,32 +2492,46 @@ fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
 /// [`LocalDisk`](super::local::LocalDisk) instances share the same lock while
 /// disconnected disks do not keep the registry alive.
 pub(crate) fn disk_volume_mutation_lock(root: &Path, volume: &str) -> Arc<RwLock<()>> {
-    let key = root.join(volume);
-    let mut locks = DISK_VOLUME_MUTATION_LOCKS.lock();
-    locks.retain(|_, lock| lock.strong_count() > 0);
-    if let Some(lock) = locks.get(&key).and_then(Weak::upgrade) {
-        return lock;
+    #[cfg(unix)]
+    {
+        let vol_bytes = volume.as_bytes();
+        if !vol_bytes.is_empty()
+            && vol_bytes[0] != b'/'
+            && vol_bytes != b"."
+            && !vol_bytes.starts_with(b"./")
+            && !vol_bytes.ends_with(b"/")
+            && !vol_bytes.ends_with(b"/.")
+            && is_canonical_unix_key_bytes(vol_bytes)
+            && let std::borrow::Cow::Borrowed(root_bytes) = canonical_unix_path_key(root)
+            && !root_bytes.is_empty()
+            && root_bytes != b"/"
+        {
+            let total_len = root_bytes.len() + 1 + vol_bytes.len();
+            if total_len <= 512 {
+                let mut stack_key = [0u8; 512];
+                stack_key[..root_bytes.len()].copy_from_slice(root_bytes);
+                stack_key[root_bytes.len()] = b'/';
+                stack_key[root_bytes.len() + 1..total_len].copy_from_slice(vol_bytes);
+                return DISK_VOLUME_MUTATION_LOCKS.get_or_create_by_key(&stack_key[..total_len], || RwLock::new(()));
+            }
+        }
     }
-
-    let lock = Arc::new(RwLock::new(()));
-    locks.insert(key, Arc::downgrade(&lock));
-    lock
+    let key = root.join(volume);
+    DISK_VOLUME_MUTATION_LOCKS.get_or_create(&key, || RwLock::new(()))
 }
 
 fn disk_namespace_mutation_lock(path: &Path) -> Arc<NamespaceMutationLock> {
-    let mut locks = DISK_NAMESPACE_MUTATION_LOCKS.lock();
-    locks.retain(|_, lock| lock.strong_count() > 0);
-    if let Some(lock) = locks.get(path).and_then(Weak::upgrade) {
-        return lock;
-    }
-
-    let lock = Arc::new(AsyncMutex::new(()));
-    locks.insert(path.to_path_buf(), Arc::downgrade(&lock));
-    lock
+    DISK_NAMESPACE_MUTATION_LOCKS.get_or_create(path, || AsyncMutex::new(()))
 }
 
-static DISK_METADATA_MUTATION_LOCKS: LazyLock<Mutex<NamespaceMutationLockRegistry>> =
-    LazyLock::new(|| Mutex::new(HashMap::new()));
+#[cfg(unix)]
+#[inline]
+fn disk_namespace_mutation_lock_by_key(key: &[u8]) -> Arc<NamespaceMutationLock> {
+    DISK_NAMESPACE_MUTATION_LOCKS.get_or_create_by_key(key, || AsyncMutex::new(()))
+}
+
+static DISK_METADATA_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<NamespaceMutationLock>> =
+    LazyLock::new(ShardedPathLockRegistry::new);
 
 /// Serializes the complete xl.meta read/modify/commit transaction. This domain
 /// precedes namespace/volume publication locks, whose narrower syscall leases
@@ -2233,17 +2545,7 @@ pub(crate) async fn acquire_metadata_mutation_lease(
     object: &Path,
     owner: Option<Arc<dyn Send + Sync>>,
 ) -> Arc<MetadataMutationLease> {
-    let lock = {
-        let mut locks = DISK_METADATA_MUTATION_LOCKS.lock();
-        locks.retain(|_, lock| lock.strong_count() > 0);
-        if let Some(lock) = locks.get(object).and_then(Weak::upgrade) {
-            lock
-        } else {
-            let lock = Arc::new(AsyncMutex::new(()));
-            locks.insert(object.to_path_buf(), Arc::downgrade(&lock));
-            lock
-        }
-    };
+    let lock = DISK_METADATA_MUTATION_LOCKS.get_or_create(object, || AsyncMutex::new(()));
     Arc::new(MetadataMutationLease {
         _guard: lock.lock_owned().await,
         _owner: owner,
@@ -2306,17 +2608,45 @@ pub(crate) async fn acquire_rename_data_mutation_lease_with_owner(
 ) -> Arc<NamespaceMutationLease> {
     // Metadata ownership precedes namespace paths in lexical order, then volume
     // ownership. Deduplicate identical keys so TMP-to-TMP publication cannot self-lock.
-    let destination_lock = disk_namespace_mutation_lock(destination_object);
-    let (namespace_guard, staging_guard) = match staging_prefix {
-        Some(source) if source < destination_object => {
-            let staging = disk_namespace_mutation_lock(source).lock_owned().await;
-            (destination_lock.lock_owned().await, Some(staging))
+    #[cfg(unix)]
+    let (namespace_guard, staging_guard) = {
+        let dst_key = canonical_unix_path_key(destination_object);
+        let destination_lock = disk_namespace_mutation_lock_by_key(dst_key.as_ref());
+        match staging_prefix {
+            Some(source) => {
+                let src_key = canonical_unix_path_key(source);
+                match src_key.as_ref().cmp(dst_key.as_ref()) {
+                    std::cmp::Ordering::Less => {
+                        let staging = disk_namespace_mutation_lock_by_key(src_key.as_ref()).lock_owned().await;
+                        (destination_lock.lock_owned().await, Some(staging))
+                    }
+                    std::cmp::Ordering::Greater => {
+                        let destination = destination_lock.lock_owned().await;
+                        (
+                            destination,
+                            Some(disk_namespace_mutation_lock_by_key(src_key.as_ref()).lock_owned().await),
+                        )
+                    }
+                    std::cmp::Ordering::Equal => (destination_lock.lock_owned().await, None),
+                }
+            }
+            None => (destination_lock.lock_owned().await, None),
         }
-        Some(source) if source > destination_object => {
-            let destination = destination_lock.lock_owned().await;
-            (destination, Some(disk_namespace_mutation_lock(source).lock_owned().await))
+    };
+    #[cfg(not(unix))]
+    let (namespace_guard, staging_guard) = {
+        let destination_lock = disk_namespace_mutation_lock(destination_object);
+        match staging_prefix {
+            Some(source) if source < destination_object => {
+                let staging = disk_namespace_mutation_lock(source).lock_owned().await;
+                (destination_lock.lock_owned().await, Some(staging))
+            }
+            Some(source) if source > destination_object => {
+                let destination = destination_lock.lock_owned().await;
+                (destination, Some(disk_namespace_mutation_lock(source).lock_owned().await))
+            }
+            _ => (destination_lock.lock_owned().await, None),
         }
-        _ => (destination_lock.lock_owned().await, None),
     };
     let volume_guard = disk_volume_mutation_lock(root, volume).read_owned().await;
     let blocking_queue = disk_blocking_queue_for_root(root);
@@ -11992,6 +12322,121 @@ mod tests {
         );
         assert!(preserved_final.exists(), "final file must remain when intermediate removal fails");
     }
+
+    #[cfg(unix)]
+    #[test]
+    fn canonical_unix_path_key_preserves_std_path_equality_and_prefix_semantics() {
+        let samples = [
+            "",
+            "/",
+            "///",
+            "/.",
+            "/./.",
+            ".",
+            "./",
+            "./.",
+            "./a",
+            "././a/",
+            ".//a/.",
+            "a",
+            "a/",
+            "a/.",
+            "a/b/c",
+            "a/b/c/",
+            "a//b/./c/.",
+            "/tmp/d0/bucket/obj",
+            "/tmp/d0/bucket/obj/",
+            "/tmp/d0//bucket/./obj/.",
+            "/tmp/d0/.rustfs.sys/multipart/sha/upload",
+            "/proc/self/fd/12/.",
+            "/proc/self/fd/12",
+            "/proc/self/fd/12/./bucket/obj",
+            "/proc/self/fd/12/bucket/obj",
+            "/tmp/d1",
+            "/tmp/d10",
+            "a/../b",
+            "a/../b/",
+            "/utf8/\u{00af}/segment/",
+        ];
+        for s in samples {
+            let p = Path::new(s);
+            let fast = canonical_unix_path_key(p);
+            let slow = canonical_unix_path_key_slow(p);
+            assert_eq!(fast.as_ref(), slow.as_ref(), "fast and slow key mismatch for {s:?}");
+        }
+        for a in samples {
+            for b in samples {
+                let pa = Path::new(a);
+                let pb = Path::new(b);
+                let ka = canonical_unix_path_key(pa);
+                let kb = canonical_unix_path_key(pb);
+                assert_eq!(
+                    pa == pb,
+                    ka.as_ref() == kb.as_ref(),
+                    "Path::eq vs canonical key mismatch for {a:?} and {b:?}"
+                );
+                assert_eq!(
+                    pa.starts_with(pb),
+                    unix_canonical_key_starts_with(ka.as_ref(), kb.as_ref()),
+                    "Path::starts_with vs canonical key prefix mismatch for {a:?} and {b:?}"
+                );
+            }
+        }
+
+        let dir = tempdir().expect("tempdir");
+        let p_clean = dir.path().join("bucket/object");
+        let p_trailing = PathBuf::from(format!("{}/", p_clean.display()));
+        let p_redundant = PathBuf::from(format!("{}//bucket/./object/.", dir.path().display()));
+        let lock1 = disk_namespace_mutation_lock(&p_clean);
+        let lock2 = disk_namespace_mutation_lock(&p_trailing);
+        let lock3 = disk_namespace_mutation_lock(&p_redundant);
+        assert!(Arc::ptr_eq(&lock1, &lock2));
+        assert!(Arc::ptr_eq(&lock1, &lock3));
+
+        let vlock1 = disk_volume_mutation_lock(dir.path(), "bucket");
+        let vlock2 = disk_volume_mutation_lock(&PathBuf::from(format!("{}/.", dir.path().display())), "bucket/");
+        assert!(Arc::ptr_eq(&vlock1, &vlock2));
+
+        let io_root = dir.path().join("io_root/.");
+        let q_reg = register_disk_blocking_queue(dir.path(), &io_root);
+        let q_root = disk_blocking_queue_for_root(dir.path());
+        let q_path = disk_blocking_queue_for_path(&dir.path().join("io_root/bucket/key"));
+        assert!(Arc::ptr_eq(&q_reg, &q_root));
+        assert!(Arc::ptr_eq(&q_reg, &q_path));
+    }
+
+    #[test]
+    fn sharded_path_lock_registry_reuses_slots_and_bounds_dead_weak_entries() {
+        let registry: ShardedPathLockRegistry<AsyncMutex<()>> = ShardedPathLockRegistry::new();
+        let target = Path::new("/tmp/sharded-lock-test/persistent-key");
+        let held = registry.get_or_create(target, || AsyncMutex::new(()));
+        let held_again = registry.get_or_create(target, || AsyncMutex::new(()));
+        assert!(Arc::ptr_eq(&held, &held_again), "live lock must be shared");
+
+        // Re-acquiring a key after drop must reuse the existing slot in-place.
+        let transient = Path::new("/tmp/sharded-lock-test/reused-slot");
+        drop(registry.get_or_create(transient, || AsyncMutex::new(())));
+        let reused = registry.get_or_create(transient, || AsyncMutex::new(()));
+        drop(reused);
+
+        // Flood transient keys across shards and verify `held` is never evicted
+        // and total dead entries remain bounded by the amortized GC threshold.
+        for i in 0..(PATH_LOCK_SHARDS * PATH_LOCK_INITIAL_GC_THRESHOLD * 3) {
+            let p = PathBuf::from(format!("/tmp/sharded-lock-test/ephemeral-{i}"));
+            drop(registry.get_or_create(&p, || AsyncMutex::new(())));
+        }
+        let still_held = registry.get_or_create(target, || AsyncMutex::new(()));
+        assert!(Arc::ptr_eq(&held, &still_held), "GC must never evict a live lock");
+        for shard in &registry.shards {
+            let guard = shard.lock();
+            assert!(
+                guard.entries.len() <= guard.next_gc_threshold,
+                "shard entries ({}) exceeded GC threshold ({})",
+                guard.entries.len(),
+                guard.next_gc_threshold
+            );
+        }
+    }
 }
 
 #[cfg(all(test, target_os = "linux"))]

From 12c58d25b58cdeb9b98ea63c82d13921653c7132 Mon Sep 17 00:00:00 2001
From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com>
Date: Mon, 2 Nov 2026 23:01:55 +0800
Subject: [PATCH 2/5] chore(deps): bump the dependencies group across 1
 directory with 5 updates (#9113)

* chore(deps): bump the dependencies group across 1 directory with 5 updates

Bumps the dependencies group with 5 updates in the / directory:

| Package | From | To |
| --- | --- | --- |
| [pulsar](https://github.com/streamnative/pulsar-rs) | `6.9.0` | `6.9.1` |
| [arrow-array](https://github.com/apache/arrow-rs) | `59.3.2` | `60.1.0` |
| [arrow-ipc](https://github.com/apache/arrow-rs) | `59.3.2` | `60.1.0` |
| [arrow-schema](https://github.com/apache/arrow-rs) | `59.3.2` | `60.1.0` |
| [parquet](https://github.com/apache/arrow-rs) | `59.3.2` | `60.1.0` |



Updates `pulsar` from 6.9.0 to 6.9.1
- [Release notes](https://github.com/streamnative/pulsar-rs/releases)
- [Commits](https://github.com/streamnative/pulsar-rs/compare/v6.9.0...v6.9.1)

Updates `arrow-array` from 59.3.2 to 60.1.0
- [Release notes](https://github.com/apache/arrow-rs/releases)
- [Changelog](https://github.com/apache/arrow-rs/blob/main/CHANGELOG.md)
- [Commits](https://github.com/apache/arrow-rs/compare/59.3.2...60.1.0)

Updates `arrow-ipc` from 59.3.2 to 60.1.0
- [Release notes](https://github.com/apache/arrow-rs/releases)
- [Changelog](https://github.com/apache/arrow-rs/blob/main/CHANGELOG.md)
- [Commits](https://github.com/apache/arrow-rs/compare/59.3.2...60.1.0)

Updates `arrow-schema` from 59.3.2 to 60.1.0
- [Release notes](https://github.com/apache/arrow-rs/releases)
- [Changelog](https://github.com/apache/arrow-rs/blob/main/CHANGELOG.md)
- [Commits](https://github.com/apache/arrow-rs/compare/59.3.2...60.1.0)

Updates `parquet` from 59.3.2 to 60.1.0
- [Release notes](https://github.com/apache/arrow-rs/releases)
- [Changelog](https://github.com/apache/arrow-rs/blob/main/CHANGELOG.md)
- [Commits](https://github.com/apache/arrow-rs/compare/59.3.2...60.1.0)

---
updated-dependencies:
- dependency-name: arrow-array
  dependency-version: 60.1.0
  dependency-type: direct:production
  update-type: version-update:semver-major
  dependency-group: dependencies
- dependency-name: arrow-ipc
  dependency-version: 60.1.0
  dependency-type: direct:production
  update-type: version-update:semver-major
  dependency-group: dependencies
- dependency-name: arrow-schema
  dependency-version: 60.1.0
  dependency-type: direct:production
  update-type: version-update:semver-major
  dependency-group: dependencies
- dependency-name: parquet
  dependency-version: 60.1.0
  dependency-type: direct:production
  update-type: version-update:semver-major
  dependency-group: dependencies
- dependency-name: pulsar
  dependency-version: 6.9.1
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: dependencies
...

Signed-off-by: dependabot[bot] <support@github.com>

* chore(deps): update dependencies and pin pulsar 6.9.0

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

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

* chore(deps): ignore pulsar 6.9 updates

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

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

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Hauser <housemecn@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
---
 .github/dependabot.yml        |   2 +
 Cargo.lock                    | 311 +++++++++++++++++++++++-----------
 Cargo.toml                    |   8 +-
 crates/zstd-compat/Cargo.toml |   8 +-
 4 files changed, 224 insertions(+), 105 deletions(-)

diff --git a/.github/dependabot.yml b/.github/dependabot.yml
index 24c57eba9..7827ad1f3 100644
--- a/.github/dependabot.yml
+++ b/.github/dependabot.yml
@@ -32,6 +32,8 @@ updates:
       - "overtrue"
       - "majinghe"
     ignore:
+      - dependency-name: "pulsar"
+        versions: [ "6.9.x" ]
       - dependency-name: "object_store"
         versions: [ "0.13.x" ]
       - dependency-name: "libunftp"
diff --git a/Cargo.lock b/Cargo.lock
index badc6a684..8b4273870 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -370,17 +370,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f04e0a89cc175eaf3740282537851e6fd9d7276f97730b9d0ea2859c11a85df"
 dependencies = [
  "arrow-arith",
- "arrow-array",
- "arrow-buffer",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
  "arrow-cast",
  "arrow-csv",
- "arrow-data",
- "arrow-ipc",
+ "arrow-data 59.3.2",
+ "arrow-ipc 59.3.2",
  "arrow-json",
  "arrow-ord",
  "arrow-row",
- "arrow-schema",
- "arrow-select",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
  "arrow-string",
 ]
 
@@ -390,10 +390,10 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "157943c720919e8f5b6731407ce5cd4c9a5ef47a75bf10edaf24b7081b5763ef"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
  "chrono",
  "num-traits",
 ]
@@ -405,9 +405,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72b5b1682924f11ecf6929689c16cbc69efbb6dd141b06ad98ee1cb10d858c43"
 dependencies = [
  "ahash",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
  "chrono",
  "chrono-tz",
  "half",
@@ -417,6 +417,24 @@ dependencies = [
  "num-traits",
 ]
 
+[[package]]
+name = "arrow-array"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f3d22a41d3df78b71df0b685af219ae3a94188ab98d6fb29c958eb55c3247f4"
+dependencies = [
+ "ahash",
+ "arrow-buffer 60.1.0",
+ "arrow-data 60.1.0",
+ "arrow-schema 60.1.0",
+ "chrono",
+ "half",
+ "hashbrown 0.17.1",
+ "num-complex",
+ "num-integer",
+ "num-traits",
+]
+
 [[package]]
 name = "arrow-buffer"
 version = "59.3.2"
@@ -429,18 +447,30 @@ dependencies = [
  "num-traits",
 ]
 
+[[package]]
+name = "arrow-buffer"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7bd46a530bc1a8e763102fc36c1dc0f51f3ffa150c78e7f5658ccb54b1458ade"
+dependencies = [
+ "bytes",
+ "half",
+ "num-bigint 0.5.1",
+ "num-traits",
+]
+
 [[package]]
 name = "arrow-cast"
 version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f23c3379feede42135ef27ca60e67f6e643cd4343ae74e4c8b16c83f55afd85"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
  "arrow-ord",
- "arrow-schema",
- "arrow-select",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
  "atoi 2.0.0",
  "base64 0.23.1",
  "chrono",
@@ -451,15 +481,26 @@ dependencies = [
  "ryu",
 ]
 
+[[package]]
+name = "arrow-cmp"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1913e57a64fbf2c4502fa6d54b547d6a1ae59e3146cbc15321d47468c937abd1"
+dependencies = [
+ "arrow-array 60.1.0",
+ "arrow-buffer 60.1.0",
+ "arrow-schema 60.1.0",
+]
+
 [[package]]
 name = "arrow-csv"
 version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "85b1984628b091b8319b3c305fce21d04f9e9b2578bb733b0d0108462ae1d698"
 dependencies = [
- "arrow-array",
+ "arrow-array 59.3.2",
  "arrow-cast",
- "arrow-schema",
+ "arrow-schema 59.3.2",
  "chrono",
  "csv",
  "csv-core",
@@ -472,8 +513,21 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bb0a031c7ff5e2bdb443117054fb466af1d4df5b7f0ecbe197713dc8b113b1ba"
 dependencies = [
- "arrow-buffer",
- "arrow-schema",
+ "arrow-buffer 59.3.2",
+ "arrow-schema 59.3.2",
+ "half",
+ "num-integer",
+ "num-traits",
+]
+
+[[package]]
+name = "arrow-data"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "415b7fbc4cf56b080bb5677a4ae74211b5c79e571ed37a1d547633ec07f3b43b"
+dependencies = [
+ "arrow-buffer 60.1.0",
+ "arrow-schema 60.1.0",
  "half",
  "num-integer",
  "num-traits",
@@ -485,28 +539,43 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f5dd1afadda2eb821aaf851bb4c0adef60c30cc0f1a29bd02fb99dd07fb9f58d"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
- "arrow-select",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
  "flatbuffers",
  "lz4_flex",
  "zstd 0.13.4",
 ]
 
+[[package]]
+name = "arrow-ipc"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "45f3e766bccf5a0cb57a5de52a4cba450dea5171547c2cb5939b486d08887cba"
+dependencies = [
+ "arrow-array 60.1.0",
+ "arrow-buffer 60.1.0",
+ "arrow-data 60.1.0",
+ "arrow-schema 60.1.0",
+ "arrow-select 60.1.0",
+ "flatbuffers",
+ "zstd 0.14.3",
+]
+
 [[package]]
 name = "arrow-json"
 version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e2bc4d56a0dad84cc51c5812ccff48b642047a1f3be80349cd84318983e79edb"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
  "arrow-cast",
  "arrow-ord",
- "arrow-schema",
- "arrow-select",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
  "chrono",
  "half",
  "indexmap 2.14.2",
@@ -526,11 +595,11 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "561b8209faa2cd489764b7fd07d7cb9e8d274d78048c38de59f77c41140cb834"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
- "arrow-select",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
 ]
 
 [[package]]
@@ -539,10 +608,10 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0d8c68b87a569f92b90c46cb0a09d86bd9609ddc9b1bf1fc5a503c8c6907c14e"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
  "half",
 ]
 
@@ -556,6 +625,12 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "arrow-schema"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "de2c8e714b466404cdc1189e89f892426efea614d938e16c15518c340cd43ba3"
+
 [[package]]
 name = "arrow-select"
 version = "59.3.2"
@@ -563,10 +638,25 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c844141958d805b47d1208e0d12aac53fcf388ce8011639fe82aa492adde0bb4"
 dependencies = [
  "ahash",
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
+ "num-traits",
+]
+
+[[package]]
+name = "arrow-select"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a5af3f6656c5f2c6944f93d9a357a7f204c93d5f349385a44cf46c4a26fbcf30"
+dependencies = [
+ "ahash",
+ "arrow-array 60.1.0",
+ "arrow-buffer 60.1.0",
+ "arrow-cmp",
+ "arrow-data 60.1.0",
+ "arrow-schema 60.1.0",
  "num-traits",
 ]
 
@@ -576,11 +666,11 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bc130892b1a9552da77971924b390e06cf375d1f7e4344a7b7b40ca4426fa470"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
- "arrow-select",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
  "memchr",
  "num-traits",
  "regex",
@@ -669,9 +759,9 @@ dependencies = [
 
 [[package]]
 name = "async-compression"
-version = "0.4.51"
+version = "0.4.52"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9d5755640120b24961f8231e05892e4a4f7e0d04e8aeca161508c9c397946970"
+checksum = "6ad47a740fb906fffdd891e8b6d5fde2511600f7bfaebd68939fa08a14999459"
 dependencies = [
  "compression-codecs",
  "compression-core",
@@ -1993,9 +2083,9 @@ dependencies = [
 
 [[package]]
 name = "cc"
-version = "1.10.0"
+version = "1.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "86e055fa6e3e40ad6187ba25b6855eeb635d5271f8e0f0e1e9a5826b8d75bea0"
+checksum = "375989decec1a0fa424442bed520fa12ede43a48475ce8c7417d2c69b32b5f6b"
 dependencies = [
  "find-msvc-tools",
  "jobserver",
@@ -2251,9 +2341,9 @@ dependencies = [
 
 [[package]]
 name = "compression-codecs"
-version = "0.4.46"
+version = "0.4.47"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cae36683101b2899f48bd31f8db9839437356188e3b1ff23096e73dab09ba8fe"
+checksum = "84f840cff84d10085260e460023934d0af699e71212b09b2a41452e8bd9cda43"
 dependencies = [
  "brotli 9.1.1",
  "bzip2",
@@ -2499,9 +2589,9 @@ dependencies = [
 
 [[package]]
 name = "crc32fast"
-version = "1.5.3"
+version = "1.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ba106e844045ddc02c20768ac4b0b7cb45f2cc2ecbd833ac9f5dacb183c2055"
+checksum = "5f97d3416d5337d15c4b0d78c8c91df922ca143f57ec20e2d1c188333047ef04"
 dependencies = [
  "cfg-if",
 ]
@@ -2891,7 +2981,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4cd5317db77ba6bca25c761a8091b9d294a9f4188e4f9dcd0243e6a57de862ac"
 dependencies = [
  "arrow",
- "arrow-schema",
+ "arrow-schema 59.3.2",
  "async-trait",
  "chrono",
  "datafusion-catalog",
@@ -2925,7 +3015,7 @@ dependencies = [
  "log",
  "object_store",
  "parking_lot 0.12.5",
- "parquet",
+ "parquet 59.3.2",
  "sqlparser",
  "tempfile",
  "tokio",
@@ -2989,8 +3079,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d26626e12b224b83351239465dcd00f72d47b02fb5e5b65a643e2dc8ed37f687"
 dependencies = [
  "arrow",
- "arrow-ipc",
- "arrow-schema",
+ "arrow-ipc 59.3.2",
+ "arrow-schema 59.3.2",
  "chrono",
  "foldhash 0.2.0",
  "half",
@@ -3001,7 +3091,7 @@ dependencies = [
  "log",
  "num-traits",
  "object_store",
- "parquet",
+ "parquet 59.3.2",
  "recursive",
  "sqlparser",
  "tokio",
@@ -3057,7 +3147,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "65b1b041688cad707fa103c3e7f4c6d89b3134ebf67da0129070dad3381851eb"
 dependencies = [
  "arrow",
- "arrow-ipc",
+ "arrow-ipc 59.3.2",
  "async-trait",
  "bytes",
  "datafusion-common",
@@ -3127,7 +3217,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c521ae91b6d899d7190812881f45fe9463c222c11d90149f5332a2e56555904"
 dependencies = [
  "arrow",
- "arrow-schema",
+ "arrow-schema 59.3.2",
  "async-trait",
  "bytes",
  "datafusion-common",
@@ -3148,7 +3238,7 @@ dependencies = [
  "log",
  "object_store",
  "parking_lot 0.12.5",
- "parquet",
+ "parquet 59.3.2",
  "tokio",
 ]
 
@@ -3165,7 +3255,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cfdd5ac0769d1e74a885450e49fde8f63ca7ab41eb479a068564aaea80b837"
 dependencies = [
  "arrow",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
  "async-trait",
  "bytes",
  "dashmap",
@@ -3191,7 +3281,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "86560fd7df4dda924f5474b588c56e7ae8e96dcf158c7327c9cfe9daff448ec5"
 dependencies = [
  "arrow",
- "arrow-schema",
+ "arrow-schema 59.3.2",
  "async-trait",
  "chrono",
  "datafusion-common",
@@ -3227,7 +3317,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "939ae2771da8a3fdc23ab4de610ae8fef9515476626b49d96d0d5c782b105771"
 dependencies = [
  "arrow",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
  "base64 0.23.1",
  "chrono",
  "chrono-tz",
@@ -3387,7 +3477,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9f002cef849e10c0597a217ec7891c2796c0f8b42d739f5d24abc24496373406"
 dependencies = [
  "arrow",
- "arrow-schema",
+ "arrow-schema 59.3.2",
  "datafusion-common",
  "datafusion-expr",
  "datafusion-expr-common",
@@ -3462,10 +3552,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b1451d6817e5ac0dd21819b19949d1c420ab51d748f55dd9a96ecb5418a24c97"
 dependencies = [
  "arrow",
- "arrow-data",
- "arrow-ipc",
+ "arrow-data 59.3.2",
+ "arrow-ipc 59.3.2",
  "arrow-ord",
- "arrow-schema",
+ "arrow-schema 59.3.2",
  "async-trait",
  "bytes",
  "datafusion-common",
@@ -3513,7 +3603,7 @@ version = "55.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "62dc5ce37a5ce1ff5362949776829aded2529a11dbaa13d24b0ffa6a059e106a"
 dependencies = [
- "arrow-schema",
+ "arrow-schema 59.3.2",
  "async-trait",
  "datafusion-common",
  "datafusion-execution",
@@ -7591,12 +7681,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8eb260d566f2fbdabe356d26aaa5d13403aac29b5b4e44463b8f9bd7b2b12680"
 dependencies = [
  "ahash",
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-ipc",
- "arrow-schema",
- "arrow-select",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-ipc 59.3.2",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
  "base64 0.23.1",
  "brotli 8.0.4",
  "bytes",
@@ -7618,6 +7708,32 @@ dependencies = [
  "zstd 0.13.4",
 ]
 
+[[package]]
+name = "parquet"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "806eaa0f1c0436dcb43bee8e6a82bf2efff90521a60172f8281cb9ef577b43a4"
+dependencies = [
+ "ahash",
+ "arrow-array 60.1.0",
+ "arrow-buffer 60.1.0",
+ "arrow-data 60.1.0",
+ "arrow-ipc 60.1.0",
+ "arrow-schema 60.1.0",
+ "arrow-select 60.1.0",
+ "base64 0.23.1",
+ "bytes",
+ "chrono",
+ "half",
+ "hashbrown 0.17.1",
+ "num-bigint 0.5.1",
+ "num-integer",
+ "num-traits",
+ "seq-macro",
+ "twox-hash",
+ "zstd 0.14.3",
+]
+
 [[package]]
 name = "password-hash"
 version = "0.6.1"
@@ -8976,9 +9092,9 @@ dependencies = [
 
 [[package]]
 name = "reflink-copy"
-version = "0.1.31"
+version = "0.1.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3dd31f458a66eeda05a9e837fc5f0effbaacbe8565868ecae5b37bf0544c19f5"
+checksum = "1ad64f61357bcf16105a0f353f9783733409c8bd889bc326ad857959605ea927"
 dependencies = [
  "cfg-if",
  "libc",
@@ -10214,18 +10330,18 @@ dependencies = [
 
 [[package]]
 name = "rustfs-mimalloc"
-version = "0.7.2"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8301a578772fc9823d68e083e42c3b414613d588a0eac594c1ca6bc0bbe0653f"
+checksum = "216ecfc9341b5d6e2506d47ce57c9232ae1b8020bb10264ea05286fb234286e7"
 dependencies = [
  "rustfs-mimalloc-sys",
 ]
 
 [[package]]
 name = "rustfs-mimalloc-sys"
-version = "0.7.2"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6eb4082853043cb29556b4b776b05f28c3146706f43c53869dbf1ef262e00e91"
+checksum = "fb16ad68f96883407a75cdb88252b43dc2ea592428f0fc54ff7885756de4b814"
 dependencies = [
  "cc",
 ]
@@ -12420,18 +12536,18 @@ dependencies = [
 
 [[package]]
 name = "test-case"
-version = "3.4.0"
+version = "3.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "124953e7f67cb0b2fcfb87e899e4ae5a64fb68e2e160767933cc67b646fb0042"
+checksum = "ae6e80608d2037173d7d59a09089d249e43352c182f6dde6b00d138bcb7cfdbe"
 dependencies = [
  "test-case-macros",
 ]
 
 [[package]]
 name = "test-case-core"
-version = "3.4.0"
+version = "3.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bd097615b407247e102046ba34426bb594923ab3554cf0f3cdfed050f1a5a3e8"
+checksum = "29b8e1f65819001927b8cdcbb5560f2428212f2e5ab5684e3c6815a4d80e72db"
 dependencies = [
  "cfg-if",
  "proc-macro2",
@@ -12441,9 +12557,9 @@ dependencies = [
 
 [[package]]
 name = "test-case-macros"
-version = "3.4.0"
+version = "3.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f4ce35ece947bccac166e1ded639133827a953f69a48ba134d17f91108333082"
+checksum = "802d6165a14df1598b7290a8c546df402940d3cfbe82d15c0fd35de176771693"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -14082,9 +14198,9 @@ dependencies = [
 
 [[package]]
 name = "zip"
-version = "9.2.1"
+version = "9.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "848dad21639597378c3be94c59fd4de683f0f6d8a5387575896e1245737d4f37"
+checksum = "e408c6cc4699dd017d53dc080cd51c2cea2b5354af57a1479bd265ed42911487"
 dependencies = [
  "aes 0.9.3",
  "bzip2",
@@ -14095,6 +14211,7 @@ dependencies = [
  "getrandom 0.4.3",
  "hmac 0.13.0",
  "indexmap 2.14.2",
+ "libc",
  "lzma-rust2",
  "memchr",
  "nix 0.31.3",
@@ -14138,12 +14255,12 @@ name = "zstd"
 version = "0.13.4"
 dependencies = [
  "apache-avro",
- "arrow-array",
- "arrow-ipc",
- "arrow-schema",
+ "arrow-array 60.1.0",
+ "arrow-ipc 60.1.0",
+ "arrow-schema 60.1.0",
  "bytes",
  "kafka-protocol",
- "parquet",
+ "parquet 60.1.0",
  "semver",
  "toml",
  "zstd 0.14.3",
diff --git a/Cargo.toml b/Cargo.toml
index 2e67147c5..512ab5747 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -147,7 +147,7 @@ rustfs-zip = { path = "./crates/zip", version = "1.0.2" }
 async-channel = "2.5.0"
 async_zip = { default-features = false, version = "0.0.19" }
 mysql_async = { default-features = false, version = "0.37.1" }
-async-compression = { version = "0.4.51" }
+async-compression = { version = "0.4.52" }
 async-trait = "0.1.92"
 async-nats = { version = "0.51.0", default-features = false }
 axum = "0.8.9"
@@ -337,7 +337,7 @@ strum = { version = "0.28.0" }
 sysinfo = "0.40.1"
 temp-env = "0.3.6"
 tempfile = "3.27.0"
-test-case = "3.4.0"
+test-case = "3.4.1"
 thiserror = "2.0.21"
 tracing = { version = "0.1.44" }
 tracing-appender = "0.2.5"
@@ -356,7 +356,7 @@ winapi-util = "0.1.11"
 windows = { version = "0.62.2" }
 windows-sys = "0.61.2"
 xxhash-rust = { version = "0.8.19" }
-zip = "9.2.1"
+zip = "9.2.2"
 zstd = "0.14.3"
 
 # Observability and Metrics
@@ -384,7 +384,7 @@ russh-sftp = "3.0.1"
 dav-server = "0.11.0"
 
 # Performance Analysis and Memory Profiling
-rustfs-mimalloc = { version = "0.7.2" }
+rustfs-mimalloc = { version = "0.7.3" }
 # Preserve Unicode focus filters until rustfs/backlog#2302 is resolved.
 hotpath = { version = "0.30.3", default-features = false }
 # Snapshot testing for output format regression detection
diff --git a/crates/zstd-compat/Cargo.toml b/crates/zstd-compat/Cargo.toml
index b0f63dbf9..9244aff11 100644
--- a/crates/zstd-compat/Cargo.toml
+++ b/crates/zstd-compat/Cargo.toml
@@ -33,10 +33,10 @@ zstd-safe = { version = "8.2.0", features = ["seekable"] }
 toml = "1.1.8"
 semver = "1.0.28"
 apache-avro.workspace = true
-arrow-array = "59.3.2"
-arrow-ipc = { version = "59.3.2", features = ["zstd"] }
-arrow-schema = "59.3.2"
-parquet = { version = "59.3.2", default-features = false, features = ["arrow", "zstd"] }
+arrow-array = "60.1.0"
+arrow-ipc = { version = "60.1.0", features = ["zstd"] }
+arrow-schema = "60.1.0"
+parquet = { version = "60.1.0", default-features = false, features = ["arrow", "zstd"] }
 kafka-protocol = { version = "0.18.0", default-features = false, features = ["zstd"] }
 bytes.workspace = true
 

From 9d66e04ae37b7e95d749ac58d1fc6419a5e95288 Mon Sep 17 00:00:00 2001
From: Hauser Zhi <housemecn@gmail.com>
Date: Mon, 2 Nov 2026 23:02:22 +0800
Subject: [PATCH 3/5] test(e2e): honor namespace backpressure in mixed workload
 (#9122)

Give each mixed-workload request a bounded policy for the server's
explicit namespace lock timeout response and Retry-After advice. Keep
unrelated SDK failures and data mismatches immediately visible.

Preserve 24 workers, four rounds, cross-node checks and all retained
objects. Cover exhausted SDK retries, bodyless HEAD, fatal responses,
persistent contention and incorrect acknowledged bytes, and include the
new cases in the protected distributed profile inventory.

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
---
 .config/e2e-distributed-selection.txt         |   4 +-
 .../distributed/concurrency_stability_test.rs | 309 ++++++++++++++++--
 2 files changed, 288 insertions(+), 25 deletions(-)

diff --git a/.config/e2e-distributed-selection.txt b/.config/e2e-distributed-selection.txt
index bf308ff5a..39204eef8 100644
--- a/.config/e2e-distributed-selection.txt
+++ b/.config/e2e-distributed-selection.txt
@@ -1,2 +1,2 @@
-sha256-linux=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4
-sha256-darwin=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4
+sha256-linux=2f97cbbeb2dbf52919b704a3d07336c162a41f95079d941d74251f9279c494e6
+sha256-darwin=2f97cbbeb2dbf52919b704a3d07336c162a41f95079d941d74251f9279c494e6
diff --git a/crates/e2e_test/src/distributed/concurrency_stability_test.rs b/crates/e2e_test/src/distributed/concurrency_stability_test.rs
index 06a24e9a4..c32d8b51d 100644
--- a/crates/e2e_test/src/distributed/concurrency_stability_test.rs
+++ b/crates/e2e_test/src/distributed/concurrency_stability_test.rs
@@ -12,11 +12,249 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::harness::{DistCluster, DistLayout, TestResult, assert_object_bytes, payload_for, put_object, unique_bucket};
+use super::harness::{DistCluster, DistLayout, TestResult, payload_for, s3_probe_client, sha256_hex, unique_bucket};
 use crate::common::init_logging;
+use aws_sdk_s3::{
+    Client,
+    error::{ProvideErrorMetadata, SdkError},
+    primitives::ByteStream,
+};
+use std::any::TypeId;
 use std::collections::BTreeSet;
+use std::fmt::Debug;
+use std::future::Future;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::Barrier;
+use tokio::time::Instant;
+
+const LOCK_ADMISSION_BUDGET: Duration = Duration::from_secs(45);
+
+async fn after_lock_contention<T, E, F, Fut>(operation: &str, mut call: F) -> TestResult<Result<T, SdkError<E>>>
+where
+    E: Debug + ProvideErrorMetadata + 'static,
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T, SdkError<E>>>,
+{
+    let started = Instant::now();
+    let mut attempts = 0;
+    // The workload uses unique unversioned keys. Repeat only explicitly
+    // advertised lock backpressure; every step still needs its real result.
+    let outcome = tokio::time::timeout(LOCK_ADMISSION_BUDGET, async {
+        loop {
+            attempts += 1;
+            let result = call().await;
+            let retry = result.as_ref().err().is_some_and(|error| {
+                let Some(response) = error.raw_response() else {
+                    return false;
+                };
+                let Some(service_error) = error.as_service_error() else {
+                    return false;
+                };
+                // HEAD has no XML error body, so its SDK service error can
+                // lack a code even when the server supplies retry headers.
+                let namespace_code = service_error.code() == Some("ServiceUnavailable")
+                    || (service_error.code().is_none()
+                        && TypeId::of::<E>() == TypeId::of::<aws_sdk_s3::operation::head_object::HeadObjectError>());
+                response.status().as_u16() == 503
+                    && namespace_code
+                    && response.headers().get("x-rustfs-error-cause") == Some("namespace_lock_timeout")
+                    && response.headers().get("retry-after") == Some("1")
+            });
+            if !retry {
+                return result;
+            }
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+    })
+    .await;
+    if attempts > 1 || !matches!(&outcome, Ok(Ok(_))) {
+        eprintln!(
+            "mixed workload request: operation={operation}, attempts={attempts}, elapsed={:?}",
+            started.elapsed()
+        );
+    }
+    outcome.map_err(|_| {
+        format!("{operation} exceeded its {LOCK_ADMISSION_BUDGET:?} request budget after {attempts} attempts").into()
+    })
+}
+
+async fn assert_workload_bytes(client: &Client, bucket: &str, key: &str, expected: &[u8]) -> TestResult {
+    let output = after_lock_contention(&format!("GET {key}"), || client.get_object().bucket(bucket).key(key).send()).await??;
+    let actual = output.body.collect().await?.into_bytes();
+    if actual.as_ref() != expected {
+        return Err(format!(
+            "object {key} bytes mismatch: expected {} bytes sha256={} got {} bytes sha256={}",
+            expected.len(),
+            sha256_hex(expected),
+            actual.len(),
+            sha256_hex(&actual)
+        )
+        .into());
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aws_smithy_http_client::test_util::infallible_client_fn;
+    use aws_smithy_types::body::SdkBody;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    type Reply = (u16, &'static str, Option<&'static str>, Option<&'static str>, &'static str);
+
+    fn client(respond: impl Fn(usize) -> Reply + Send + Sync + 'static) -> (Client, Arc<AtomicUsize>) {
+        let requests = Arc::new(AtomicUsize::new(0));
+        let seen = Arc::clone(&requests);
+        let transport = infallible_client_fn(move |_| {
+            let (status, code, cause, retry_after, body) = respond(seen.fetch_add(1, Ordering::SeqCst));
+            let body = if code.is_empty() {
+                body.to_string()
+            } else {
+                format!("<Error><Code>{code}</Code></Error>")
+            };
+            let mut response = http::Response::builder()
+                .status(status)
+                .header("content-type", "application/xml")
+                .header("content-length", body.len());
+            if let Some(cause) = cause {
+                response = response.header("x-rustfs-error-cause", cause);
+            }
+            if let Some(retry_after) = retry_after {
+                response = response.header("retry-after", retry_after);
+            }
+            response.body(SdkBody::from(body)).expect("scripted S3 response")
+        });
+        let config = crate::common::build_test_s3_config(
+            "http://contention.example.invalid",
+            "synthetic-access",
+            "synthetic-secret",
+            None,
+            "contention-test",
+        );
+        let client = Client::from_conf(config.to_builder().http_client(transport).build());
+        (s3_probe_client(&client), requests)
+    }
+
+    fn contention() -> Reply {
+        (503, "ServiceUnavailable", Some("namespace_lock_timeout"), Some("1"), "")
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn advertised_contention_recovers_after_standard_sdk_attempts_are_exhausted() -> TestResult {
+        let (client, requests) = client(|attempt| {
+            if attempt < 3 {
+                contention()
+            } else {
+                (200, "", None, None, "")
+            }
+        });
+        let standard = Client::from_conf(
+            client
+                .config()
+                .to_builder()
+                .retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(3))
+                .build(),
+        );
+        standard
+            .put_object()
+            .bucket("bucket")
+            .key("key")
+            .body(ByteStream::from_static(b"body"))
+            .send()
+            .await
+            .expect_err("three advertised lock responses exhaust the standard SDK policy");
+        assert_eq!(requests.load(Ordering::SeqCst), 3);
+        requests.store(0, Ordering::SeqCst);
+        let started = Instant::now();
+        after_lock_contention("PUT recovery", || {
+            client
+                .put_object()
+                .bucket("bucket")
+                .key("key")
+                .body(ByteStream::from_static(b"body"))
+                .send()
+        })
+        .await??;
+        assert_eq!(requests.load(Ordering::SeqCst), 4);
+        assert!(started.elapsed() >= Duration::from_secs(3), "honor each Retry-After response");
+        Ok(())
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn bodyless_head_contention_uses_explicit_retry_headers() -> TestResult {
+        let (client, requests) = client(|attempt| {
+            if attempt == 0 {
+                (503, "", Some("namespace_lock_timeout"), Some("1"), "")
+            } else {
+                (200, "", None, None, "")
+            }
+        });
+        after_lock_contention("HEAD recovery", || client.head_object().bucket("bucket").key("key").send()).await??;
+        assert_eq!(requests.load(Ordering::SeqCst), 2);
+        Ok(())
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn unrelated_or_malformed_responses_are_not_retried() -> TestResult {
+        for reply in [
+            (500, "InternalError", Some("namespace_lock_timeout"), Some("1"), ""),
+            (403, "AccessDenied", None, None, ""),
+            (404, "NoSuchKey", None, None, ""),
+            (503, "ServiceUnavailable", Some("storage_write_quorum"), None, ""),
+            (503, "ServiceUnavailable", Some("namespace_lock_unavailable"), Some("1"), ""),
+            (503, "InternalError", Some("namespace_lock_timeout"), Some("1"), ""),
+            (503, "", Some("namespace_lock_timeout"), Some("1"), ""),
+            (503, "ServiceUnavailable", None, Some("1"), ""),
+            (503, "ServiceUnavailable", Some("namespace_lock_timeout"), None, ""),
+            (503, "ServiceUnavailable", Some("namespace_lock_timeout"), Some("2"), ""),
+        ] {
+            let (client, requests) = client(move |_| reply);
+            let result = after_lock_contention("PUT fatal response", || {
+                client
+                    .put_object()
+                    .bucket("bucket")
+                    .key("key")
+                    .body(ByteStream::from_static(b"body"))
+                    .send()
+            })
+            .await?;
+            assert!(result.is_err(), "fatal response cannot become success: {reply:?}");
+            assert_eq!(requests.load(Ordering::SeqCst), 1, "fatal response must remain visible: {reply:?}");
+        }
+        Ok(())
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn persistent_lock_contention_exhausts_the_request_budget() {
+        let (client, requests) = client(|_| contention());
+        let started = Instant::now();
+        let result = after_lock_contention("PUT unavailable", || {
+            client
+                .put_object()
+                .bucket("bucket")
+                .key("key")
+                .body(ByteStream::from_static(b"body"))
+                .send()
+        })
+        .await;
+        assert!(result.is_err(), "persistent contention cannot acknowledge an operation");
+        assert!(started.elapsed() >= LOCK_ADMISSION_BUDGET);
+        assert!(requests.load(Ordering::SeqCst) > 1);
+        assert!(requests.load(Ordering::SeqCst) <= LOCK_ADMISSION_BUDGET.as_secs() as usize + 1);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn acknowledged_wrong_bytes_fail_without_retry() {
+        let (client, requests) = client(|_| (200, "", None, None, "wrong bytes"));
+        let error = assert_workload_bytes(&client, "bucket", "key", b"expected bytes")
+            .await
+            .expect_err("a successful but incorrect read is a consistency failure");
+        assert!(error.to_string().contains("bytes mismatch"));
+        assert_eq!(requests.load(Ordering::SeqCst), 1);
+    }
+}
 
 #[tokio::test]
 async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node() -> TestResult {
@@ -24,7 +262,9 @@ async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node()
     let dist = DistCluster::start(DistLayout::FourByFour).await?;
     let bucket = unique_bucket("conc");
     dist.create_bucket(&bucket).await?;
-    let clients = Arc::new(dist.clients()?);
+    // This consistency probe owns its admission policy; unrelated responses
+    // must not be hidden by the SDK's general-purpose retries.
+    let clients = Arc::new(dist.clients()?.iter().map(s3_probe_client).collect::<Vec<_>>());
     let worker_count = 24;
     let rounds = 4;
     let barrier = Arc::new(Barrier::new(worker_count));
@@ -44,31 +284,46 @@ async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node()
                 let key = format!("source/worker-{idx:02}-round-{round}.bin");
                 let copy_key = format!("retained/worker-{idx:02}-round-{round}.bin");
                 let body = payload_for(&key, 64 * 1024);
-                put_object(writer, &bucket, &key, body.clone()).await?;
+                let context = format!("worker={idx} round={round}");
+                after_lock_contention(&format!("{context} PUT source"), || {
+                    writer
+                        .put_object()
+                        .bucket(&bucket)
+                        .key(&key)
+                        .body(ByteStream::from(body.clone()))
+                        .send()
+                })
+                .await??;
 
-                let head = reader.head_object().bucket(&bucket).key(&key).send().await?;
+                let head = after_lock_contention(&format!("{context} HEAD source"), || {
+                    reader.head_object().bucket(&bucket).key(&key).send()
+                })
+                .await??;
                 if head.content_length() != Some(body.len() as i64) {
                     return Err(format!("HEAD returned the wrong size for {key}: {head:?}").into());
                 }
-                assert_object_bytes(reader, &bucket, &key, &body).await?;
+                assert_workload_bytes(reader, &bucket, &key, &body).await?;
 
-                copier
-                    .copy_object()
-                    .bucket(&bucket)
-                    .key(&copy_key)
-                    .copy_source(format!("{bucket}/{key}"))
-                    .send()
-                    .await?;
-                assert_object_bytes(writer, &bucket, &copy_key, &body).await?;
+                after_lock_contention(&format!("{context} COPY retained"), || {
+                    copier
+                        .copy_object()
+                        .bucket(&bucket)
+                        .key(&copy_key)
+                        .copy_source(format!("{bucket}/{key}"))
+                        .send()
+                })
+                .await??;
+                assert_workload_bytes(writer, &bucket, &copy_key, &body).await?;
 
-                writer.delete_object().bucket(&bucket).key(&key).send().await?;
-                let missing = reader
-                    .head_object()
-                    .bucket(&bucket)
-                    .key(&key)
-                    .send()
-                    .await
-                    .expect_err("deleted source key must not remain visible");
+                after_lock_contention(&format!("{context} DELETE source"), || {
+                    writer.delete_object().bucket(&bucket).key(&key).send()
+                })
+                .await??;
+                let missing = after_lock_contention(&format!("{context} HEAD deleted source"), || {
+                    reader.head_object().bucket(&bucket).key(&key).send()
+                })
+                .await?
+                .expect_err("deleted source key must not remain visible");
                 if missing.raw_response().map(|response| response.status().as_u16()) != Some(404) {
                     return Err(format!("deleted source {key} returned an unexpected result: {missing:?}").into());
                 }
@@ -82,14 +337,22 @@ async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node()
     for handle in handles {
         inventory.extend(handle.await??);
     }
+    assert_eq!(
+        inventory.len(),
+        worker_count * rounds,
+        "every worker round must retain its acknowledged copy"
+    );
 
     let expected_keys: BTreeSet<_> = inventory.iter().map(|(key, _)| key.as_str()).collect();
     for (node_idx, client) in clients.iter().enumerate() {
-        let listed = client.list_objects_v2().bucket(&bucket).prefix("retained/").send().await?;
+        let listed = after_lock_contention(&format!("node={node_idx} LIST retained"), || {
+            client.list_objects_v2().bucket(&bucket).prefix("retained/").send()
+        })
+        .await??;
         let listed_keys: BTreeSet<_> = listed.contents().iter().filter_map(|object| object.key()).collect();
         assert_eq!(listed_keys, expected_keys, "node {node_idx} returned a divergent retained-key listing");
         for (key, body) in &inventory {
-            assert_object_bytes(client, &bucket, key, body)
+            assert_workload_bytes(client, &bucket, key, body)
                 .await
                 .map_err(|error| format!("node {node_idx} failed to read {key}: {error}"))?;
         }

From 965d805a4f1d1fdb08b9731a6f5073f7a154e5ea Mon Sep 17 00:00:00 2001
From: hector <42570491+majinghe@users.noreply.github.com>
Date: Mon, 2 Nov 2026 23:03:36 +0800
Subject: [PATCH 4/5] ci: run distributed and upgrade e2e on schedule only
 (#9126)

Remove pull_request from the distributed and upgrade workflows. Distributed
stays on the nightly cron, and upgrade stays on the Monday cron plus
version tags. Both remain manual-dispatch lanes and stay outside the
required merge checks.

Co-authored-by: majinghe <majinghe@users.noreply.github.com>
---
 .config/nextest.toml                  |  2 +-
 .github/scheduled-validations.json    |  7 ++-----
 .github/workflows/e2e-distributed.yml | 30 ++-------------------------
 .github/workflows/e2e-upgrade.yml     | 25 +++++++---------------
 crates/e2e_test/README.md             |  6 +++---
 docs/testing/README.md                |  4 ++--
 docs/testing/ci-gates.md              |  7 +++----
 7 files changed, 20 insertions(+), 61 deletions(-)

diff --git a/.config/nextest.toml b/.config/nextest.toml
index 6c083520e..0574fb73c 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -827,7 +827,7 @@ slow-timeout = { period = "120s", terminate-after = 12, grace-period = "10s" }
 # ---------------------------------------------------------------------------
 # e2e-distributed profile — 4-node 4-disk Actions suite
 # ---------------------------------------------------------------------------
-# Storage-sensitive PR / nightly / dispatch lane owned by
+# Nightly and manual-dispatch lane owned by
 # .github/workflows/e2e-distributed.yml.
 # Each case starts four rustfs processes (and for site replication, two
 # clusters). Upgrade cases also require RUSTFS_UPGRADE_SOURCE_BINARY.
diff --git a/.github/scheduled-validations.json b/.github/scheduled-validations.json
index ffc959b82..891daba94 100644
--- a/.github/scheduled-validations.json
+++ b/.github/scheduled-validations.json
@@ -769,12 +769,10 @@
       "workflow": ".github/workflows/e2e-distributed.yml",
       "name": "e2e-distributed",
       "roles": [
-        "pr",
         "periodic",
         "manual"
       ],
       "events": [
-        "pull_request",
         "workflow_dispatch",
         "schedule"
       ],
@@ -849,12 +847,11 @@
       "workflow": ".github/workflows/e2e-upgrade.yml",
       "name": "Upgrade Compatibility",
       "roles": [
-        "pr",
         "periodic",
-        "manual"
+        "manual",
+        "release"
       ],
       "events": [
-        "pull_request",
         "push",
         "schedule",
         "workflow_dispatch"
diff --git a/.github/workflows/e2e-distributed.yml b/.github/workflows/e2e-distributed.yml
index 7b955c30e..47806f998 100644
--- a/.github/workflows/e2e-distributed.yml
+++ b/.github/workflows/e2e-distributed.yml
@@ -17,8 +17,8 @@
 # Each selected test starts a real localhost cluster via
 # `RustFSTestClusterEnvironment` (4 processes; 4 drives per node unless the
 # case is a two-site 4-node 1-drive pair or a 4-node upgrade). Membership is
-# `[profile.e2e-distributed]` in `.config/nextest.toml`. Storage-sensitive PRs,
-# nightly runs, and manual dispatches all execute the same fail-closed suite.
+# `[profile.e2e-distributed]` in `.config/nextest.toml`. The nightly schedule
+# and manual dispatches execute the same fail-closed suite.
 # Upgrade cases download the same pinned previous release as e2e-upgrade.yml.
 #
 # Isolated pool filesystems: expand/decommission/rebalance cases require
@@ -33,32 +33,6 @@
 name: e2e-distributed
 
 on:
-  pull_request:
-    paths:
-      - "Cargo.lock"
-      - "Cargo.toml"
-      - ".config/nextest.toml"
-      - ".config/e2e-distributed-selection.txt"
-      - ".github/workflows/e2e-distributed.yml"
-      - "crates/audit/**"
-      - "crates/common/**"
-      - "crates/config/**"
-      - "crates/e2e_test/**"
-      - "crates/ecstore/**"
-      - "crates/filemeta/**"
-      - "crates/heal/**"
-      - "crates/iam/**"
-      - "crates/lock/**"
-      - "crates/madmin/**"
-      - "crates/notify/**"
-      - "crates/replication/**"
-      - "crates/s3-client/**"
-      - "crates/s3-ops/**"
-      - "crates/s3-types/**"
-      - "crates/scanner/**"
-      - "crates/storage-api/**"
-      - "crates/utils/**"
-      - "rustfs/**"
   workflow_dispatch:
     inputs:
       filter:
diff --git a/.github/workflows/e2e-upgrade.yml b/.github/workflows/e2e-upgrade.yml
index 215e16cba..92e8c2163 100644
--- a/.github/workflows/e2e-upgrade.yml
+++ b/.github/workflows/e2e-upgrade.yml
@@ -12,30 +12,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+# Single-node upgrade and rollback lane.
+#
+# The weekly Monday schedule, version tags, and workflow_dispatch run this
+# suite. Distributed upgrade cases stay on the nightly e2e-distributed lane.
+
 name: Upgrade Compatibility
 
 on:
-  pull_request:
-    paths:
-      - ".github/workflows/e2e-upgrade.yml"
-      - "crates/e2e_test/src/common.rs"
-      - "crates/e2e_test/src/distributed/**"
-      - "crates/e2e_test/src/fake_s3_target/**"
-      - "crates/e2e_test/src/lib.rs"
-      - "crates/e2e_test/src/replication_extension_test.rs"
-      - "crates/e2e_test/src/rpc_fault_proxy.rs"
-      - "crates/e2e_test/src/upgrade_compatibility_test.rs"
-      - "crates/ecstore/**"
-      - "crates/filemeta/**"
-      - "crates/heal/**"
-      - "crates/kms/**"
-      - "crates/storage-api/**"
-      - "rustfs/**"
-      - "Cargo.lock"
   push:
     tags:
       - "[0-9]*.[0-9]*.[0-9]*"
   schedule:
+    # 03:17 UTC every Monday.
     - cron: "17 3 * * 1"
   workflow_dispatch:
 
@@ -44,7 +33,7 @@ permissions:
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
-  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+  cancel-in-progress: false
 
 env:
   CARGO_TERM_COLOR: always
diff --git a/crates/e2e_test/README.md b/crates/e2e_test/README.md
index 8eef80ee2..ab49ffba5 100644
--- a/crates/e2e_test/README.md
+++ b/crates/e2e_test/README.md
@@ -26,7 +26,7 @@ Registered in [`src/lib.rs`](src/lib.rs). Grouped by concern:
 | **protocols** | [`src/protocols/`](src/protocols) | FTPS, WebDAV, SFTP compliance. Fixed ports, own guide: [`src/protocols/README.md`](src/protocols/README.md) |
 | **reliant** | [`src/reliant/`](src/reliant) | Tests that reuse an **externally started** server (SQL/select, conditional writes, lifecycle, deleted-object reads, node-interact). Run via [`scripts/run_e2e_tests.sh`](../../scripts/run_e2e_tests.sh); see [`src/reliant/README.md`](src/reliant/README.md) |
 | **cluster** | `cluster_concurrency_test`, `stale_multipart_cleanup_cluster_test`, `namespace_lock_quorum_test`, `admin_timeout_regression_test`, `object_lambda_test`, `replication_extension_test`, `tier_stats_cluster_test` | Multi-node scenarios via `RustFSTestClusterEnvironment` |
-| **distributed 4×4** | [`src/distributed/`](src/distributed) | Storage-sensitive PR and nightly `e2e-distributed` lane: S3, object lock/WORM, versioning, bucket/site replication, quota, expand/decommission/rebalance, concurrency, chaos, 4-node upgrade of historical data and IAM AK/SK. Map: [`docs/testing/distributed-e2e.md`](../../docs/testing/distributed-e2e.md) |
+| **distributed 4×4** | [`src/distributed/`](src/distributed) | Nightly `e2e-distributed` lane: S3, object lock/WORM, versioning, bucket/site replication, quota, expand/decommission/rebalance, concurrency, chaos, 4-node upgrade of historical data and IAM AK/SK. Map: [`docs/testing/distributed-e2e.md`](../../docs/testing/distributed-e2e.md) |
 | **chaos / reliability** | [`src/chaos.rs`](src/chaos.rs), `reliability_disk_fault_test`, `heal_erasure_disk_rebuild_test`, `server_startup_failfast_test` | Disk offline/replace/corrupt, EC rebuild, heal, fail-fast startup |
 | **upgrade compatibility** | `upgrade_compatibility_test` | Pinned previous-release writes followed by current-build reads on the same data directory |
 
@@ -194,9 +194,9 @@ the same profile for membership and execution with one nightly worker.
 | `s3s-e2e` black-box | `e2e-tests` + `e2e-tests-rio-v2` jobs | **Active** (external conformance tool) |
 | ILM / lifecycle (ignored) | `test-ilm-integration-serial` lane, `-j1` | **Active** (backlog#1148 ilm-1) |
 | KMS suite | `e2e-full` job, merge queue + main | **Active** |
-| Direct and mixed-version rolling upgrades from pinned previous release | `e2e-upgrade.yml`, storage-sensitive PRs + release tags + weekly | **Active** |
+| Direct and mixed-version rolling upgrades from pinned previous release | `e2e-upgrade.yml`, release tags + weekly schedule + manual dispatch | **Active** |
 | Cluster faults (`e2e-nightly` profile) | consolidated nightly workflow | **Active** (backlog#1149 ci-7) |
-| Distributed 4-node 4-disk (`e2e-distributed` profile) | `.github/workflows/e2e-distributed.yml` | **Active** (storage-sensitive PR / nightly / dispatch) |
+| Distributed 4-node 4-disk (`e2e-distributed` profile) | `.github/workflows/e2e-distributed.yml` | **Active** (nightly / dispatch) |
 | Protocols (FTPS/WebDAV/SFTP) | consolidated nightly workflow, serial | **Active** (backlog#1149 ci-7) |
 | Replication (fast subset) | `e2e-smoke` profile, `e2e-tests` job, every PR | **Active** (backlog#1147 repl-1) |
 | Replication (slow + multi-node) | `e2e-repl-nightly` profile, consolidated nightly workflow | **Active** (backlog#1147 repl-1) |
diff --git a/docs/testing/README.md b/docs/testing/README.md
index a47adaab1..2a2027bd8 100644
--- a/docs/testing/README.md
+++ b/docs/testing/README.md
@@ -14,7 +14,7 @@ Pick the lowest layer that can prove the change; add a higher-layer test only wh
 | Outbound target matrix | Replication of every object shape (empty, plain, retention, legal hold, multipart, checksummed, SSE-S3) against every remote-target failure mode the fake target models; an explicit expectation table pins known-red cells to an open issue | `python3 scripts/e2e_binary.py build`, then `python3 scripts/e2e_binary.py run -- cargo nextest run -p e2e_test -E 'test(/^replication_target_matrix_test::/)'` | With `e2e-repl-nightly`; required locally for any change to outbound client defaults (SOP: [`docs/postmortems/2026-09-03-replication-checksum-default-regression.md`](../postmortems/2026-09-03-replication-checksum-default-regression.md)) |
 | s3s-e2e conformance | External S3 conformance tool against a live server | `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` | Non-documentation PRs, required through `Test and Lint` (second half of the `End-to-End Tests` job) |
 | S3 compatibility | `ceph/s3-tests` (boto3; allow-list `scripts/s3-tests/implemented_tests.txt`) and MinIO `mint` | `scripts/s3-tests/run.sh`; mint via `.github/workflows/mint.yml` | Non-documentation PRs: implemented and lifecycle S3 lanes required through `Test and Lint`; the separate s3-tests workflow provides additional PR reporting and a weekly full sweep; mint: weekly, report-only |
-| Chaos / fault-injection | Single-node disk fault injection (`crates/e2e_test/src/chaos.rs`, `crates/e2e_test/src/fault_proxy.rs`) plus the 4-node kill/fresh-drive/blackhole cases in `crates/e2e_test/src/distributed/chaos_test.rs` | Part of the e2e crate (`e2e-reliability` and `e2e-distributed`) | Reliability cases with `e2e-full`; 4-node chaos on storage-sensitive PRs and nightly via `e2e-distributed` |
+| Chaos / fault-injection | Single-node disk fault injection (`crates/e2e_test/src/chaos.rs`, `crates/e2e_test/src/fault_proxy.rs`) plus the 4-node kill/fresh-drive/blackhole cases in `crates/e2e_test/src/distributed/chaos_test.rs` | Part of the e2e crate (`e2e-reliability` and `e2e-distributed`) | Reliability cases with `e2e-full`; 4-node chaos nightly via `e2e-distributed` |
 | Fuzz | `cargo-fuzz` targets over untrusted parsing surfaces; isolated sub-workspace under `fuzz/` | `./scripts/fuzz/run.sh` (see [`fuzz/README.md`](../../fuzz/README.md)) | PR smoke on the paths listed in `.github/workflows/fuzz.yml`, plus nightly corpus |
 | Benchmarks | Criterion benches under each crate's `benches/` | `cargo bench -p <crate>` | On demand; never a gate |
 | Conditional reads | Paired ordinary/hit/miss GET latency, throughput and read-lock measurements | `scripts/run_conditional_read_abba.py` | Local conditional-read acceptance; [guide and scope](conditional-read-benchmark.md) |
@@ -88,7 +88,7 @@ All profiles are defined in `.config/nextest.toml`; its block comments hold the
 | `e2e-full` | Merge-queue / main-push single-node e2e lane |
 | `e2e-repl-nightly` | Nightly slow / cross-process replication lane |
 | `e2e-nightly` | Nightly serial multi-process cluster fault lane |
-| `e2e-distributed` | Storage-sensitive PR and nightly 4-node 4-disk S3 / lock / versioning / replication / quota / expand / decommission / rebalance / site-replication / chaos / upgrade (history + IAM AK/SK) lane |
+| `e2e-distributed` | Nightly 4-node 4-disk S3 / lock / versioning / replication / quota / expand / decommission / rebalance / site-replication / chaos / upgrade (history + IAM AK/SK) lane |
 | `e2e-protocols` | Nightly fixed-port FTPS/SFTP/WebDAV lane, run with `-j 1` |
 
 Membership of each e2e profile is pinned by a digest in `.config/e2e-<profile>-selection.txt` and checked by `scripts/check_test_wiring.py --check-profile <profile>` before the lane runs. To list what a profile selects on your platform (the result is platform-dependent because some modules are linux-only):
diff --git a/docs/testing/ci-gates.md b/docs/testing/ci-gates.md
index e66481f3d..caa67fc21 100644
--- a/docs/testing/ci-gates.md
+++ b/docs/testing/ci-gates.md
@@ -35,7 +35,7 @@ gh api repos/rustfs/rustfs/rulesets/6436880 \
   --jq '.rules[] | select(.type == "required_status_checks") | .parameters'
 ```
 
-The aggregate requires the validation lanes already selected by `ci.yml`; this closes the gap where a failing critical lane left the required workspace check green. Independent workflows remain report-only unless separately required. Before adding a new expensive lane or moving existing PR coverage to a schedule, collect representative execution and regression evidence, establish ownership and a working scheduled replacement, and update this reference with the resulting policy.
+The aggregate requires the validation lanes already selected by `ci.yml`; this closes the gap where a failing critical lane left the required workspace check green. Independent workflows remain report-only unless separately required. Before adding a new expensive lane or moving existing PR coverage to a schedule, collect representative execution and regression evidence, establish ownership and a working scheduled replacement, and update this reference with the resulting policy. `e2e-distributed.yml` and `e2e-upgrade.yml` are that kind of replacement: the 4-node lane runs on its nightly schedule, and the single-node upgrade lane runs on its weekly Monday schedule, on version tags, and on manual dispatch. Neither is part of `Test and Lint`. Dispatch the workflow against a candidate branch when that signal is needed before merge.
 
 Superseded PR attempts are cancelled. A running main validation finishes, with only the newest pending main run retained, so frequent merges cannot continuously cancel the full baseline. Having a `merge_group` trigger does not itself require use of the merge queue.
 
@@ -66,7 +66,6 @@ The four site-replication state-writer concurrency proofs reserve the available
 | PR touching `paths` in `fuzz.yml` | `Build Fuzz Harness`, `Smoke / <target>` | `fuzz.yml` `fuzz-build`, `pr-fuzz-smoke` | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
 | PR touching `paths` in `windows-filesystem.yml` | `Rename Safety`, `Startup Smoke` | `windows-filesystem.yml` `rename-safety`, `startup-smoke` | Report-only | the `cargo test -p rustfs-ecstore --lib <filter>` commands in the job, on Windows; `cargo build --locked -p rustfs --bin rustfs`, then `rustfs.exe --version` and a single-volume server polled on `/health/ready` |
 | PR touching `paths` in `coverage.yml` | `Workspace line coverage` | `coverage.yml` `coverage` | Report-only | `make coverage`; `python3 scripts/check_security_coverage.py target/llvm-cov/coverage.json` |
-| PR touching `paths` in `e2e-upgrade.yml` | `Upgrade compatibility` | `e2e-upgrade.yml` `upgrade` | Report-only | `python3 scripts/e2e_binary.py build`, then the job's `python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test` commands with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
 | PR touching `paths` in `oidc-keycloak.yml` | `OIDC Keycloak live gate` | `oidc-keycloak.yml` `oidc-keycloak-live` | Report-only | `cargo build --locked -p rustfs --bin rustfs`, then `bash scripts/test/oidc_keycloak_live.sh ./target/debug/rustfs` |
 | PR touching `paths` in `targets-integration.yml` | `PostgreSQL, MySQL, AMQP, and NATS` | `targets-integration.yml` `targets-live` | Report-only | start the containers as in the job, export the `RUSTFS_TEST_*` DSNs, then the job's `cargo test --locked -p rustfs-targets --test <name> -- --ignored --test-threads=1` commands |
 | PR, documentation-only selection | `Quick Checks`, `Typos`, `Test and Lint` | `ci.yml` `quick-checks`, `typos`, `required-checks` | Required directly or via aggregate | Quick Checks commands; `python3 scripts/ci_gate.py --self-test` |
@@ -90,7 +89,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run. The v
 | `cache-warm.yml` (every six hours, plus main pushes that change `Cargo.lock`, `rust-toolchain.toml` or the setup action) | `warm-ci-dev`, `warm-ci-feat-rio`, `warm-ci-feat-proto`, `warm-ci-uring` | no verdict; sole writer of the `rust-cache` keys that `ci.yml` restores | yes | dispatch `cache-warm.yml` |
 | `build.yml` (nightly) | `build-rustfs` over the six-target platform matrix in `prepare-platform-matrix` (four Linux, macOS aarch64, Windows x86_64) | build/package integrity | yes | dispatch `build.yml` with an exact platform set |
 | `e2e-replication-nightly.yml` (nightly) | `repl-nightly`, `cluster-nightly`, `protocols-nightly` | three independent gates; JUnit, membership listing, server logs | yes | `python3 scripts/e2e_binary.py build --bins`, then `python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--features e2e-test-hooks` on both steps with `--profile e2e-nightly`; `--features ftps,webdav,sftp` on both steps with `-j 1 --profile e2e-protocols` |
-| `e2e-distributed.yml` (storage-sensitive PRs + nightly) | `distributed` | fail-closed 4-node 4-disk S3, durability, replication, movement, fault, and direct/rolling upgrade gate; JUnit, membership listing, per-node server logs | yes, with `never_ran_grace_until` | download the pinned previous release as in the workflow, export `RUSTFS_UPGRADE_SOURCE_BINARY`, run `python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks`, then `python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-distributed -p e2e_test` |
+| `e2e-distributed.yml` (nightly) | `distributed` | fail-closed 4-node 4-disk S3, durability, replication, movement, fault, and direct/rolling upgrade gate; JUnit, membership listing, per-node server logs | yes | download the pinned previous release as in the workflow, export `RUSTFS_UPGRADE_SOURCE_BINARY`, run `python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks`, then `python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-distributed -p e2e_test` |
 | `e2e-s3tests.yml` (weekly) | `s3tests` (single and distributed, four shards each), `upstream-head-canary` | compatibility gate; report, JUnit, node IDs, server logs | yes | `scripts/s3-tests/run.sh` against an existing single or distributed target |
 | `fuzz.yml` (nightly) | `nightly-fuzz-corpus` per target | gate; corpus and crash artifacts | yes (disabled) | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
 | `minio-interop.yml` (nightly) | `minio-interop` | EC + SSE read-parity gate | yes, with `never_ran_grace_until` | pinned Docker fixture steps in the workflow |
@@ -100,7 +99,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run. The v
 | `mint.yml` (weekly) | `mint` | report-only by design; per-suite PASS/FAIL/NA and raw `log.json` | yes | pinned Docker sequence in the workflow |
 | `coverage.yml` (weekly) | `coverage` | report-only trend; lcov and JSON artifact | yes (disabled) | `make coverage` |
 | `runner-hygiene.yml` (monthly) | `check-ephemerality` | runner ephemerality | yes | dispatch |
-| `e2e-upgrade.yml` (weekly) | `upgrade` (serial cases sharing one harness build) | upgrade and rollback gate; server logs | yes | see the PR row |
+| `e2e-upgrade.yml` (weekly, and on version tags) | `upgrade` (serial cases sharing one harness build) | upgrade and rollback gate; server logs | yes | `python3 scripts/e2e_binary.py build`, then the job's `python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test` commands with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
 | `oidc-keycloak.yml` (weekly) | `oidc-keycloak-live` | live OIDC gate | yes | see the PR row |
 | `targets-integration.yml` (nightly) | `targets-live` | live target gate; container logs | yes (disabled) | see the PR row |
 | `gateway-integration-sync.yml` (weekdays) | `sync` | merges `main` into `gateway/integration` as a `--no-ff` merge commit; a conflicting merge is aborted and opens a `sync/<source>-<date>` pull request listing the conflicting files, which keeps the run green; a rejected push or missing pull request fails it | yes | dispatch, optionally with `source_ref` |

From 00a238f838f9910c5bd21870f3237962d9a4d21b Mon Sep 17 00:00:00 2001
From: Henry Guo <marshawcoco@gmail.com>
Date: Mon, 2 Nov 2026 23:04:22 +0800
Subject: [PATCH 5/5] fix(table-catalog): resolve superseded staged commits
 (#9124)

fix(table-catalog): diagnose superseded staged commits

Co-authored-by: Chris An <anzhengchao@gmail.com>
---
 docs/architecture/s3-tables-support-matrix.md |   2 +-
 rustfs/src/table_catalog/iceberg/commit.rs    |  66 ++-
 rustfs/src/table_catalog/model.rs             |   1 +
 rustfs/src/table_catalog/store/object.rs      |  12 +-
 rustfs/src/table_catalog/tests.rs             |   1 +
 .../tests/superseded_commit_recovery.rs       | 487 ++++++++++++++++++
 6 files changed, 555 insertions(+), 14 deletions(-)
 create mode 100644 rustfs/src/table_catalog/tests/superseded_commit_recovery.rs

diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md
index 40b1a4a21..a2c18d6fd 100644
--- a/docs/architecture/s3-tables-support-matrix.md
+++ b/docs/architecture/s3-tables-support-matrix.md
@@ -64,7 +64,7 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t
 | Namespaces | Supported | Create, list, load, exists, drop on both prefixes; `pageSize`/`pageToken` pagination with context-bound tokens; identifiers limited to 512 ASCII characters. |
 | Tables | Supported | Direct and staged create, register, list, load, exists, rename, commit, metadata-location get/update, and drop on both prefixes. Staged create is stateless until an `assert-create` commit atomically publishes the initial metadata and table entry. Rename uses a bucket-scoped persistent fence, recoverable intent, and conditional publication; the source name is reusable only via an ETag-conditional tombstone replacement. Commit identifiers must match the URL; unknown requirements/updates fail as bad requests; register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. |
 | Commit CAS | Supported | Single-table commits validate base metadata, version token, referenced object existence, warehouse scope, and Iceberg requirements before advancing the pointer; external metadata transitions preserve monotonic assignment watermarks and immutable retained definitions. `idempotency-key-lifetime` is not advertised; mutation-wide `Idempotency-Key` semantics are unsupported. |
-| Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose and repair finalization gaps without moving the pointer. |
+| Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose and repair finalization gaps without moving the pointer. Object-backed recovery reports a staged request as SUPERSEDED only when a reachable finalized same-base winner proves it lost CAS and its idempotency binding is intact. The original staged receipt is retained; its old request still conflicts. The winning history must be retained with the losing receipt; missing or contradictory history requires manual review. |
 | Snapshot refs | Supported | List, create/replace, delete via commits; `main` is protected; refs with explicit retention need forced delete. |
 | Iceberg views | Supported | Create, list, load, replace, exists, drop with view-scoped authorization; only view format version 1. |
 | LoadTable and table credentials endpoint | Supported | Vending only on negotiated `vended-credentials`; one temporary session scoped to the warehouse prefix and current metadata location; missing credential permission falls back to metadata-only with an explicit reason; responses carry `Cache-Control: no-store, private`. |
diff --git a/rustfs/src/table_catalog/iceberg/commit.rs b/rustfs/src/table_catalog/iceberg/commit.rs
index 19ee14ca3..fe99dac9c 100644
--- a/rustfs/src/table_catalog/iceberg/commit.rs
+++ b/rustfs/src/table_catalog/iceberg/commit.rs
@@ -44,45 +44,58 @@ pub(crate) fn table_matches_staged_base(table: &TableEntry, commit_log: &CommitL
 pub(crate) struct TableCommitHistoryIndex<'a> {
     table_id: &'a str,
     reachable_states: BTreeSet<(&'a str, &'a str)>,
+    reachable_version_tokens: BTreeSet<&'a str>,
+    committed_successors: BTreeMap<(&'a str, &'a str), &'a CommitLogEntry>,
     ambiguous_states: BTreeSet<(&'a str, &'a str)>,
     cycle_detected: bool,
+    version_token_reused: bool,
 }
 
 impl<'a> TableCommitHistoryIndex<'a> {
     pub(crate) fn new(table: &'a TableEntry, commits: impl IntoIterator<Item = &'a CommitLogEntry>) -> Self {
-        let mut by_new_state = BTreeMap::<(&str, &str), Option<(&str, &str)>>::new();
+        let mut by_new_state = BTreeMap::<(&str, &str), Option<&CommitLogEntry>>::new();
         let mut ambiguous_states = BTreeSet::new();
         for commit in commits
             .into_iter()
             .filter(|commit| commit.table_id == table.table_id && !matches!(commit.status, CommitLogStatus::Failed))
         {
             let key = (commit.new_metadata_location.as_str(), commit.new_version_token.as_str());
-            let previous = (commit.previous_metadata_location.as_str(), commit.expected_version_token.as_str());
             by_new_state
                 .entry(key)
                 .and_modify(|candidate| {
                     *candidate = None;
                     ambiguous_states.insert(key);
                 })
-                .or_insert(Some(previous));
+                .or_insert(Some(commit));
         }
 
         let mut reachable_states = BTreeSet::new();
+        let mut reachable_version_tokens = BTreeSet::new();
+        let mut committed_successors = BTreeMap::new();
+        let mut version_token_reused = false;
         let mut state = (table.metadata_location.as_str(), table.version_token.as_str());
         let cycle_detected = loop {
             if !reachable_states.insert(state) {
                 break true;
             }
-            let Some(Some(previous)) = by_new_state.get(&state) else {
+            version_token_reused |= !reachable_version_tokens.insert(state.1);
+            let Some(commit) = by_new_state.get(&state).copied().flatten() else {
                 break false;
             };
-            state = *previous;
+            let previous = (commit.previous_metadata_location.as_str(), commit.expected_version_token.as_str());
+            if matches!(commit.status, CommitLogStatus::Committed) {
+                committed_successors.insert(previous, commit);
+            }
+            state = previous;
         };
         Self {
             table_id: &table.table_id,
             reachable_states,
+            reachable_version_tokens,
+            committed_successors,
             ambiguous_states,
             cycle_detected,
+            version_token_reused,
         }
     }
 
@@ -97,6 +110,49 @@ impl<'a> TableCommitHistoryIndex<'a> {
                 .reachable_states
                 .contains(&(target.new_metadata_location.as_str(), target.new_version_token.as_str()))
     }
+
+    fn proven_superseding_commit(&self, target: &CommitLogEntry) -> Option<&CommitLogEntry> {
+        if self.cycle_detected
+            || self.version_token_reused
+            || !self.ambiguous_states.is_empty()
+            || self.table_id != target.table_id
+            || !matches!(target.status, CommitLogStatus::Staged)
+            || self.reachable_version_tokens.contains(target.new_version_token.as_str())
+        {
+            return None;
+        }
+        let base = (target.previous_metadata_location.as_str(), target.expected_version_token.as_str());
+        self.committed_successors
+            .get(&base)
+            .copied()
+            .filter(|winner| winner.commit_id != target.commit_id)
+    }
+}
+
+/// Classifies an object-backed staged loser without rewriting its receipt.
+///
+/// A reachable finalized successor from the same pointer/token base proves that
+/// the old request lost CAS. Its STAGED record and idempotency binding remain
+/// available, and replay of that old request must still conflict. Missing or
+/// contradictory proof remains a manual-review state.
+pub(crate) fn table_commit_recovery_entry_from_history(
+    table: &TableEntry,
+    commit_log: &CommitLogEntry,
+    idempotency_commit: Option<&CommitLogEntry>,
+    history: &TableCommitHistoryIndex<'_>,
+) -> TableCommitRecoveryEntry {
+    let mut entry = table_commit_recovery_entry(table, commit_log, idempotency_commit, history.proves_committed(commit_log));
+    if entry.recovery_state == TableCommitRecoveryState::ManualReview
+        && matches!(
+            entry.idempotency_index_status,
+            TableCommitIdempotencyIndexStatus::NotRequired | TableCommitIdempotencyIndexStatus::Matches
+        )
+        && let Some(winner) = history.proven_superseding_commit(commit_log)
+    {
+        entry.recovery_state = TableCommitRecoveryState::Superseded;
+        entry.reason = format!("staged request lost CAS to reachable committed successor {}", winner.commit_id);
+    }
+    entry
 }
 
 pub(crate) fn table_catalog_recovery_summary(
diff --git a/rustfs/src/table_catalog/model.rs b/rustfs/src/table_catalog/model.rs
index 5d98d9d38..f02abefd8 100644
--- a/rustfs/src/table_catalog/model.rs
+++ b/rustfs/src/table_catalog/model.rs
@@ -1415,6 +1415,7 @@ pub(crate) struct TableCatalogDiagnosticsReport {
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 pub(crate) enum TableCommitRecoveryState {
     Committed,
+    Superseded,
     StagedBeforeTableUpdate,
     FinalizationRequired,
     IdempotencyIndexRepairRequired,
diff --git a/rustfs/src/table_catalog/store/object.rs b/rustfs/src/table_catalog/store/object.rs
index 09c4ee9da..6a9e3a12b 100644
--- a/rustfs/src/table_catalog/store/object.rs
+++ b/rustfs/src/table_catalog/store/object.rs
@@ -3065,11 +3065,11 @@ where
                 }
                 None => None,
             };
-            commits.push(table_commit_recovery_entry(
+            commits.push(table_commit_recovery_entry_from_history(
                 entry,
                 commit_log,
                 idempotency_commit.as_ref(),
-                history.proves_committed(commit_log),
+                &history,
             ));
         }
         commits.sort_by(|left, right| left.commit_id.cmp(&right.commit_id));
@@ -3168,12 +3168,8 @@ where
                 }
                 None => None,
             };
-            let recovery_entry = table_commit_recovery_entry(
-                &entry,
-                commit_log,
-                idempotency_commit.as_ref(),
-                history.proves_committed(commit_log),
-            );
+            let recovery_entry =
+                table_commit_recovery_entry_from_history(&entry, commit_log, idempotency_commit.as_ref(), &history);
             if matches!(
                 recovery_entry.recovery_state,
                 TableCommitRecoveryState::FinalizationRequired | TableCommitRecoveryState::IdempotencyIndexRepairRequired
diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs
index c6dd204c7..3d2b7c6b9 100644
--- a/rustfs/src/table_catalog/tests.rs
+++ b/rustfs/src/table_catalog/tests.rs
@@ -20,6 +20,7 @@ use std::sync::Arc;
 
 mod strong_maintenance_planning;
 mod strong_maintenance_planning_paths;
+mod superseded_commit_recovery;
 
 const TABLE_CATALOG_TEST_TIMEOUT: StdDuration = StdDuration::from_secs(30);
 
diff --git a/rustfs/src/table_catalog/tests/superseded_commit_recovery.rs b/rustfs/src/table_catalog/tests/superseded_commit_recovery.rs
new file mode 100644
index 000000000..5ffbf6f9b
--- /dev/null
+++ b/rustfs/src/table_catalog/tests/superseded_commit_recovery.rs
@@ -0,0 +1,487 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+
+struct StagedLoserFixture {
+    backend: TestCatalogObjectBackend,
+    store: ObjectTableCatalogStore<TestCatalogObjectBackend>,
+    loser: TableCommitRequest,
+    winner: TableCommitRequest,
+    committed: TableCommitResult,
+}
+
+async fn staged_loser_fixture(interrupt_winner_finalization: bool) -> StagedLoserFixture {
+    let backend = TestCatalogObjectBackend::default();
+    let store = ObjectTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").unwrap();
+    let table = IdentifierSegment::parse("orders").unwrap();
+    let initial = test_table_entry(
+        bucket,
+        &namespace,
+        &table,
+        default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"),
+    );
+    store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap();
+    store
+        .create_namespace(test_namespace_entry(bucket, &namespace))
+        .await
+        .unwrap();
+    store.create_table(initial.clone()).await.unwrap();
+    for name in ["00001", "loser", "winner", "reloaded"] {
+        let location = default_table_metadata_file_path(&namespace, &table, &format!("{name}.metadata.json"));
+        let mut metadata = super::super::test_support::table_metadata_json(&initial.table_uuid, &initial.warehouse_location);
+        metadata["properties"] = serde_json::json!({"writer": name});
+        backend
+            .seed_object(bucket, &location, serde_json::to_vec(&metadata).unwrap())
+            .await;
+    }
+    let paths = TableCatalogObjectPaths::default();
+    let table_path = paths.table_entry_path(bucket, &namespace, &table);
+    backend.fail_next_put(RUSTFS_META_BUCKET, &table_path).await;
+    let loser = TableCommitRequest {
+        table_bucket: bucket.to_string(),
+        namespace: namespace.public_name(),
+        table: table.as_str().to_string(),
+        commit_id: "paused-writer-a0".to_string(),
+        idempotency_key: Some("a0-request".to_string()),
+        operation: "set-properties".to_string(),
+        expected_version_token: initial.version_token.clone(),
+        expected_metadata_location: initial.metadata_location.clone(),
+        new_metadata_location: default_table_metadata_file_path(&namespace, &table, "loser.metadata.json"),
+        requirements: Vec::new(),
+        writer: None,
+    };
+    assert_matches!(store.commit_table(loser.clone()).await, Err(TableCatalogStoreError::Internal(_)));
+    let winner = TableCommitRequest {
+        commit_id: "winner-c1".to_string(),
+        idempotency_key: Some("c1-request".to_string()),
+        new_metadata_location: default_table_metadata_file_path(&namespace, &table, "winner.metadata.json"),
+        ..loser.clone()
+    };
+    if interrupt_winner_finalization {
+        let winner_log = paths.commit_log_entry_path(bucket, &initial.table_id, &winner.commit_id);
+        backend.fail_put_attempt(RUSTFS_META_BUCKET, &winner_log, 2).await;
+    }
+    let committed = store.commit_table(winner.clone()).await.unwrap();
+    StagedLoserFixture {
+        backend,
+        store,
+        loser,
+        winner,
+        committed,
+    }
+}
+
+#[tokio::test]
+async fn superseded_staged_commit_recovers_without_rewriting_receipts_or_pointer() {
+    let StagedLoserFixture {
+        backend,
+        store,
+        loser,
+        winner,
+        committed,
+    } = staged_loser_fixture(false).await;
+    let bucket = loser.table_bucket.as_str();
+    let before = backend.state.lock().await.objects.clone();
+
+    let recovered = store.recover_table_commits(bucket, "sales", "orders").await.unwrap();
+    assert_eq!(recovered.manual_review_count, 0, "a durable same-base winner proves the staged loser");
+    assert_eq!(recovered.finalized_count, 0, "a losing writer must never be finalized as committed");
+    let lost = recovered
+        .commits
+        .iter()
+        .find(|entry| entry.commit_id == loser.commit_id)
+        .unwrap();
+    assert_eq!(lost.status, CommitLogStatus::Staged);
+    assert_eq!(serde_json::to_value(&lost.recovery_state).unwrap(), "SUPERSEDED");
+    assert_eq!(lost.idempotency_index_status, TableCommitIdempotencyIndexStatus::Matches);
+    let after = backend.state.lock().await.objects.clone();
+    assert_eq!(after.len(), before.len());
+    for (path, original) in &before {
+        let actual = after.get(path).unwrap();
+        assert_eq!(actual.data, original.data, "recovery changed bytes at {path:?}");
+        assert_eq!(actual.etag, original.etag, "recovery rewrote {path:?}");
+    }
+
+    let restarted = ObjectTableCatalogStore::new(backend.clone());
+    let diagnostics = restarted.diagnose_table_catalog(bucket, "sales", "orders", 0).await.unwrap();
+    assert_eq!(diagnostics.recovery_status, TableCatalogRecoveryStatus::Healthy);
+    assert_matches!(restarted.commit_table(loser.clone()).await, Err(TableCatalogStoreError::Conflict(_)));
+    let mut stale_request = loser.clone();
+    stale_request.commit_id = "stale-new-request".to_string();
+    stale_request.idempotency_key = Some("stale-new-key".to_string());
+    assert_matches!(restarted.commit_table(stale_request).await, Err(TableCatalogStoreError::Conflict(_)));
+    let mut different_payload = loser.clone();
+    different_payload.operation = "replace".to_string();
+    assert_matches!(
+        restarted.commit_table(different_payload.clone()).await,
+        Err(TableCatalogStoreError::Conflict(_))
+    );
+    assert_eq!(
+        restarted.load_table(bucket, "sales", "orders").await.unwrap(),
+        Some(committed.table.clone())
+    );
+    assert_eq!(restarted.commit_table(winner).await.unwrap().table, committed.table);
+
+    let reloaded = TableCommitRequest {
+        commit_id: "reloaded-writer-a1".to_string(),
+        idempotency_key: Some("a1-request".to_string()),
+        expected_version_token: committed.table.version_token.clone(),
+        expected_metadata_location: committed.table.metadata_location.clone(),
+        new_metadata_location: default_table_metadata_file_path(
+            &Namespace::parse("sales").unwrap(),
+            &IdentifierSegment::parse("orders").unwrap(),
+            "reloaded.metadata.json",
+        ),
+        ..different_payload
+    };
+    let next = restarted.commit_table(reloaded.clone()).await.unwrap();
+    assert_eq!(next.table.generation, committed.table.generation + 1);
+    assert_eq!(restarted.commit_table(reloaded).await.unwrap().table, next.table);
+    let restarted = ObjectTableCatalogStore::new(backend);
+    assert_eq!(
+        restarted
+            .diagnose_table_catalog(bucket, "sales", "orders", 0)
+            .await
+            .unwrap()
+            .recovery_status,
+        TableCatalogRecoveryStatus::Healthy
+    );
+}
+
+#[tokio::test]
+async fn superseded_staged_commit_waits_for_durable_winner_finalization() {
+    let StagedLoserFixture {
+        backend,
+        store,
+        loser,
+        committed,
+        ..
+    } = staged_loser_fixture(true).await;
+    let paths = TableCatalogObjectPaths::default();
+    let log_path = paths.commit_log_entry_path(&loser.table_bucket, &committed.table.table_id, &loser.commit_id);
+    let before = backend.read_object(RUSTFS_META_BUCKET, &log_path).await.unwrap().unwrap();
+    let planned = store
+        .plan_table_commit_recovery("analytics", "sales", "orders")
+        .await
+        .unwrap();
+    assert_eq!(planned.manual_review_count, 1, "a merely staged winner is insufficient proof");
+    assert_eq!(planned.finalization_required_count, 1);
+
+    let recovered = store.recover_table_commits("analytics", "sales", "orders").await.unwrap();
+    assert_eq!(recovered.finalized_count, 1);
+    assert_eq!(recovered.manual_review_count, 0);
+    let loser_state = recovered
+        .commits
+        .iter()
+        .find(|entry| entry.commit_id == loser.commit_id)
+        .unwrap();
+    assert_eq!(loser_state.recovery_state, TableCommitRecoveryState::Superseded);
+    assert_eq!(loser_state.status, CommitLogStatus::Staged);
+    let after = backend.read_object(RUSTFS_META_BUCKET, &log_path).await.unwrap().unwrap();
+    assert_eq!(after.data, before.data);
+    assert_eq!(after.etag, before.etag);
+    assert_eq!(store.load_table("analytics", "sales", "orders").await.unwrap(), Some(committed.table));
+    assert_eq!(
+        store
+            .diagnose_table_catalog("analytics", "sales", "orders", 0)
+            .await
+            .unwrap()
+            .recovery_status,
+        TableCatalogRecoveryStatus::Healthy
+    );
+}
+
+#[tokio::test]
+async fn superseded_staged_commit_requires_retained_winner_proof_after_restart() {
+    let StagedLoserFixture {
+        backend,
+        store,
+        loser,
+        winner,
+        committed,
+    } = staged_loser_fixture(false).await;
+    let healthy = store.recover_table_commits("analytics", "sales", "orders").await.unwrap();
+    assert_eq!(healthy.manual_review_count, 0);
+    let winner_path =
+        TableCatalogObjectPaths::default().commit_log_entry_path("analytics", &committed.table.table_id, &winner.commit_id);
+    assert!(
+        backend
+            .state
+            .lock()
+            .await
+            .objects
+            .remove(&(RUSTFS_META_BUCKET.to_string(), winner_path))
+            .is_some()
+    );
+    let restarted = ObjectTableCatalogStore::new(backend);
+    let before = restarted.load_table("analytics", "sales", "orders").await.unwrap();
+    let report = restarted.recover_table_commits("analytics", "sales", "orders").await.unwrap();
+    assert_eq!(report.manual_review_count, 1);
+    assert_eq!(report.finalized_count, 0);
+    assert_eq!(report.commits[0].recovery_state, TableCommitRecoveryState::ManualReview);
+    assert_eq!(report.commits[0].status, CommitLogStatus::Staged);
+    assert_matches!(restarted.commit_table(loser).await, Err(TableCatalogStoreError::Conflict(_)));
+    assert_eq!(restarted.load_table("analytics", "sales", "orders").await.unwrap(), before);
+    let diagnostics = restarted
+        .diagnose_table_catalog("analytics", "sales", "orders", 0)
+        .await
+        .unwrap();
+    assert_eq!(diagnostics.recovery_status, TableCatalogRecoveryStatus::ManualReviewRequired);
+    assert!(
+        diagnostics
+            .recommended_actions
+            .contains(&TableCatalogRecoveryAction::ReviewCommitLog)
+    );
+}
+
+#[tokio::test]
+async fn superseded_staged_commit_does_not_repair_a_conflicting_index() {
+    let StagedLoserFixture {
+        backend,
+        store,
+        loser,
+        committed,
+        ..
+    } = staged_loser_fixture(false).await;
+    let mut conflicting = store
+        .get_commit_by_id("analytics", &committed.table.table_id, &loser.commit_id)
+        .await
+        .unwrap()
+        .unwrap();
+    conflicting.new_version_token = "contradictory-token".to_string();
+    let path = TableCatalogObjectPaths::default().commit_idempotency_entry_path(
+        "analytics",
+        &committed.table.table_id,
+        loser.idempotency_key.as_deref().unwrap(),
+    );
+    backend
+        .seed_object(RUSTFS_META_BUCKET, &path, serde_json::to_vec(&conflicting).unwrap())
+        .await;
+    let before = backend.state.lock().await.objects.clone();
+    let report = store.recover_table_commits("analytics", "sales", "orders").await.unwrap();
+    assert_eq!(report.manual_review_count, 1);
+    assert_eq!(report.finalized_count, 0);
+    let loser_state = report
+        .commits
+        .iter()
+        .find(|entry| entry.commit_id == loser.commit_id)
+        .unwrap();
+    assert_eq!(loser_state.recovery_state, TableCommitRecoveryState::ManualReview);
+    assert_eq!(loser_state.idempotency_index_status, TableCommitIdempotencyIndexStatus::Conflicting);
+    let after = backend.state.lock().await.objects.clone();
+    assert_eq!(before.len(), after.len());
+    for (key, original) in &before {
+        let actual = after.get(key).unwrap();
+        assert_eq!(actual.data, original.data);
+        assert_eq!(actual.etag, original.etag);
+    }
+    assert_matches!(store.commit_table(loser).await, Err(TableCatalogStoreError::Conflict(_)));
+}
+
+fn history_fixture() -> (TableEntry, CommitLogEntry, CommitLogEntry) {
+    let namespace = Namespace::parse("sales").unwrap();
+    let table_name = IdentifierSegment::parse("orders").unwrap();
+    let metadata = |name| default_table_metadata_file_path(&namespace, &table_name, name);
+    let mut table = test_table_entry("analytics", &namespace, &table_name, metadata("winner.metadata.json"));
+    table.version_token = "winner-token".to_string();
+    table.generation = 2;
+    let loser = CommitLogEntry {
+        version: TABLE_CATALOG_ENTRY_VERSION,
+        commit_id: "loser".to_string(),
+        idempotency_key: None,
+        table_id: table.table_id.clone(),
+        operation: "set-properties".to_string(),
+        expected_version_token: "base-token".to_string(),
+        new_version_token: "loser-token".to_string(),
+        previous_metadata_location: metadata("base.metadata.json"),
+        new_metadata_location: metadata("loser.metadata.json"),
+        requirements: Vec::new(),
+        status: CommitLogStatus::Staged,
+        writer: None,
+        created_at: None,
+        updated_at: None,
+    };
+    let winner = CommitLogEntry {
+        commit_id: "winner".to_string(),
+        new_version_token: table.version_token.clone(),
+        new_metadata_location: table.metadata_location.clone(),
+        status: CommitLogStatus::Committed,
+        ..loser.clone()
+    };
+    (table, loser, winner)
+}
+
+fn classify_history(
+    table: &TableEntry,
+    target: &CommitLogEntry,
+    commits: &[CommitLogEntry],
+    indexed: Option<&CommitLogEntry>,
+) -> TableCommitRecoveryEntry {
+    let history = TableCommitHistoryIndex::new(table, commits.iter());
+    table_commit_recovery_entry_from_history(table, target, indexed, &history)
+}
+
+#[test]
+fn superseded_history_requires_a_reachable_finalized_same_base_winner() {
+    for case in [
+        "staged-winner",
+        "failed-winner",
+        "wrong-base-token",
+        "wrong-base-location",
+        "foreign-table",
+        "disconnected-winner",
+        "same-commit-id",
+    ] {
+        let (mut table, loser, mut winner) = history_fixture();
+        match case {
+            "staged-winner" => winner.status = CommitLogStatus::Staged,
+            "failed-winner" => winner.status = CommitLogStatus::Failed,
+            "wrong-base-token" => winner.expected_version_token = "other-base-token".to_string(),
+            "wrong-base-location" => winner.previous_metadata_location = "other/base.metadata.json".to_string(),
+            "foreign-table" => winner.table_id = "other-table-id".to_string(),
+            "disconnected-winner" => {
+                table.metadata_location = "other/current.metadata.json".to_string();
+                table.version_token = "other-current-token".to_string();
+            }
+            "same-commit-id" => winner.commit_id.clone_from(&loser.commit_id),
+            _ => unreachable!(),
+        }
+        let recovery = classify_history(&table, &loser, &[loser.clone(), winner], None);
+        assert_eq!(recovery.recovery_state, TableCommitRecoveryState::ManualReview, "{case}");
+    }
+}
+
+#[test]
+fn superseded_history_is_stable_after_later_commits() {
+    let (mut table, loser, winner) = history_fixture();
+    let immediate = classify_history(&table, &loser, &[loser.clone(), winner.clone()], None);
+    assert_eq!(immediate.recovery_state, TableCommitRecoveryState::Superseded);
+    assert!(immediate.reason.contains(&winner.commit_id));
+    let later = CommitLogEntry {
+        commit_id: "later".to_string(),
+        previous_metadata_location: table.metadata_location.clone(),
+        expected_version_token: table.version_token.clone(),
+        new_metadata_location: "later/current.metadata.json".to_string(),
+        new_version_token: "later-token".to_string(),
+        ..winner.clone()
+    };
+    table.metadata_location.clone_from(&later.new_metadata_location);
+    table.version_token.clone_from(&later.new_version_token);
+    table.generation += 1;
+    let after = classify_history(&table, &loser, &[loser.clone(), winner, later], None);
+    assert_eq!(after.recovery_state, TableCommitRecoveryState::Superseded);
+    assert_eq!(after.reason, immediate.reason);
+}
+
+#[test]
+fn superseded_history_rejects_failed_or_committed_losers() {
+    for status in [CommitLogStatus::Failed, CommitLogStatus::Committed] {
+        let (table, mut loser, winner) = history_fixture();
+        loser.status = status;
+        let recovery = classify_history(&table, &loser, &[loser.clone(), winner], None);
+        assert_eq!(recovery.recovery_state, TableCommitRecoveryState::ManualReview);
+    }
+}
+
+#[test]
+fn superseded_history_rejects_ambiguity_cycles_and_reused_tokens() {
+    for case in [
+        "ambiguous-current",
+        "cycle",
+        "reused-history-token",
+        "loser-token-in-history",
+        "off-history-ambiguity",
+    ] {
+        let (mut table, mut loser, winner) = history_fixture();
+        let mut extra = CommitLogEntry {
+            commit_id: "extra".to_string(),
+            ..winner.clone()
+        };
+        let mut commits = vec![loser.clone(), winner.clone()];
+        match case {
+            "ambiguous-current" => commits.push(extra),
+            "cycle" => {
+                extra.previous_metadata_location.clone_from(&winner.new_metadata_location);
+                extra.expected_version_token.clone_from(&winner.new_version_token);
+                extra.new_metadata_location.clone_from(&winner.previous_metadata_location);
+                extra.new_version_token.clone_from(&winner.expected_version_token);
+                commits.push(extra);
+            }
+            "reused-history-token" => {
+                extra.previous_metadata_location.clone_from(&winner.new_metadata_location);
+                extra.expected_version_token.clone_from(&winner.new_version_token);
+                extra.new_metadata_location = "later/current.metadata.json".to_string();
+                extra.new_version_token.clone_from(&winner.expected_version_token);
+                table.metadata_location.clone_from(&extra.new_metadata_location);
+                table.version_token.clone_from(&extra.new_version_token);
+                commits.push(extra);
+            }
+            "loser-token-in-history" => {
+                loser.new_version_token.clone_from(&winner.expected_version_token);
+                commits[0] = loser.clone();
+            }
+            "off-history-ambiguity" => {
+                extra.new_metadata_location = "unknown/metadata.json".to_string();
+                extra.new_version_token = "unknown-token".to_string();
+                commits.push(extra.clone());
+                extra.commit_id = "another-extra".to_string();
+                commits.push(extra);
+            }
+            _ => unreachable!(),
+        }
+        assert_eq!(
+            classify_history(&table, &loser, &commits, None).recovery_state,
+            TableCommitRecoveryState::ManualReview,
+            "{case}"
+        );
+    }
+}
+
+#[test]
+fn superseded_history_requires_an_intact_idempotency_binding() {
+    let (table, mut loser, winner) = history_fixture();
+    loser.idempotency_key = Some("loser-request".to_string());
+    let commits = [loser.clone(), winner];
+    assert_eq!(
+        classify_history(&table, &loser, &commits, Some(&loser)).recovery_state,
+        TableCommitRecoveryState::Superseded
+    );
+    let missing = classify_history(&table, &loser, &commits, None);
+    assert_eq!(missing.recovery_state, TableCommitRecoveryState::ManualReview);
+    assert_eq!(missing.idempotency_index_status, TableCommitIdempotencyIndexStatus::Missing);
+    let mut stale = loser.clone();
+    stale.status = CommitLogStatus::Committed;
+    let state = classify_history(&table, &loser, &commits, Some(&stale));
+    assert_eq!(state.recovery_state, TableCommitRecoveryState::ManualReview);
+    assert_eq!(state.idempotency_index_status, TableCommitIdempotencyIndexStatus::Stale);
+    let mut conflicting = loser.clone();
+    conflicting.commit_id = "another-request".to_string();
+    let state = classify_history(&table, &loser, &commits, Some(&conflicting));
+    assert_eq!(state.recovery_state, TableCommitRecoveryState::ManualReview);
+    assert_eq!(state.idempotency_index_status, TableCommitIdempotencyIndexStatus::Conflicting);
+}
+
+#[test]
+fn superseded_history_never_relabels_a_committed_staged_ancestor() {
+    let (table, mut ancestor, winner) = history_fixture();
+    ancestor.new_metadata_location.clone_from(&winner.previous_metadata_location);
+    ancestor.new_version_token.clone_from(&winner.expected_version_token);
+    ancestor.previous_metadata_location = "older/base.metadata.json".to_string();
+    ancestor.expected_version_token = "older-base-token".to_string();
+    let recovery = classify_history(&table, &ancestor, &[ancestor.clone(), winner], None);
+    assert_eq!(recovery.recovery_state, TableCommitRecoveryState::FinalizationRequired);
+}