Merge remote-tracking branch 'origin/main' into overtrue/fix-heal-lock-ci

.config/nextest.toml

@@ -960,14 +960,20 @@ test-group = 'e2e-inline-boundaries'
 filter = 'package(e2e_test) & test(/^heal_erasure_disk_rebuild_test::/)'
 test-group = 'e2e-cluster-nightly'
 
+# The multi-pool crash fixture seeds three nodes and 36 drives. Reserve this
+# nextest run's workers so other multi-node tests do not starve its deadline.
+[[profile.e2e-full.overrides]]
+filter = 'package(e2e_test) & test(=heal_erasure_disk_rebuild_test::tests::test_cluster_root_heal_recovers_ec84_shards_across_multi_pool_after_background_target_crash)'
+threads-required = "num-test-threads"
+
 [[profile.e2e-full.overrides]]
 filter = 'package(e2e_test) & (test(/^kms::kms_vault_test::/) | test(/^kms::kms_rekey_sweep_test::/) | test(/^kms::configured_roundtrip_test::test_configured_vault_kms_admin_and_versioned_cleanup$/))'
 test-group = 'e2e-vault'
 
-# Keep the storage-heavy smoke cases isolated in the full suite too. They
-# start multiple server fixtures and issue many sequential S3 operations.
+# Keep storage-heavy cases isolated in the full suite too. The large pagination
+# fixture alone seeds 2,000 objects and can otherwise starve other servers.
 [[profile.e2e-full.overrides]]
-filter = 'package(e2e_test) & (test(/^list_objects_v2_pagination_test::tests::(test_list_objects_v2_delimiter_small_page_traverses_all|test_list_objects_v2_max_keys_above_limit_returns_token|test_list_objects_v2_maxkeys_above_limit_with_delimiter)$/) | test(=reliant::tiering::test_hermetic_transition_restore_failure_expiry_and_retry))'
+filter = 'package(e2e_test) & (test(/^list_objects_v2_pagination_test::tests::(test_list_objects_v2_delimiter_small_page_traverses_all|test_list_objects_v2_max_keys_above_limit_returns_token|test_list_objects_v2_maxkeys_above_limit_with_delimiter|test_list_objects_v2_raw_exceeds_maxkeys_but_visible_below)$/) | test(=reliant::tiering::test_hermetic_transition_restore_failure_expiry_and_retry))'
 threads-required = "num-test-threads"
 
 # Fixed, small contract for every executable server change. These existing

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -1700,6 +1700,15 @@ mod tests {
         );
 
         let mut cluster = RustFSTestClusterEnvironment::with_topology(topology.cluster_topology()).await?;
+        let bucket = "heal-restart-during-rebuild";
+        let mrf_fallback_scope = if !background_enabled {
+            let path = PathBuf::from(&cluster.temp_dir).join("skip-outage-mrf-fallback");
+            std::fs::write(&path, format!("{bucket}/cluster/written-while-node-down-"))?;
+            cluster.set_env("RUSTFS_E2E_SKIP_MRF_FALLBACK_PATH", path.to_string_lossy().into_owned());
+            Some(path)
+        } else {
+            None
+        };
         cluster.set_env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true");
         cluster.set_env("RUSTFS_HEAL_ENABLED", "true");
         if let Some(storage_class) = evidence_case.and_then(|case| case.storage_class_standard) {
@@ -1952,6 +1961,21 @@ mod tests {
             }
         };
 
+        if let Some(path) = &mrf_fallback_scope {
+            let observed = path.with_extension("observed");
+            let expected = format!("{bucket}/{outage_key}");
+            let deadline = Instant::now() + Duration::from_secs(30);
+            loop {
+                if tokio::fs::read_to_string(&observed).await.ok().as_deref() == Some(expected.as_str()) {
+                    break;
+                }
+                if Instant::now() >= deadline {
+                    return Err("outage repair did not reach the isolated best-effort fallback before target rejoin".into());
+                }
+                sleep(Duration::from_millis(50)).await;
+            }
+        }
+
         assert!(
             !outage_target_manifest_required
                 || target_set_peer_drives
@@ -2045,7 +2069,12 @@ mod tests {
             } else {
                 cluster_heal_is_idle(&recovered)
             };
-            if ready {
+            let pool_ready = if ready && background_enabled && !outage_write_deferred_until_rejoin {
+                all_nodes_pool_metadata_readiness(&cluster, recovery_deadline).await.0
+            } else {
+                true
+            };
+            if ready && pool_ready {
                 break;
             }
             if Instant::now() >= recovery_deadline {
@@ -2542,6 +2571,10 @@ mod tests {
             );
         }
 
+        if let Some(path) = &mrf_fallback_scope {
+            std::fs::remove_file(path)?;
+        }
+
         if let Some(cycle_end) = scanner_cycle_floor {
             wait_for_scanner_cycle_after(&cluster, cycle_end).await?;
         }

crates/ecstore/src/cluster/rpc/remote_locker.rs

@@ -1315,6 +1315,34 @@ impl LockClient for RemoteClient {
         Ok(resp.success)
     }
 
+    async fn downgrade_committed_tail(&self, lock_id: &LockId) -> Result<bool> {
+        let marker = rustfs_lock::client::COMMITTED_TAIL_CONVERSION_MARKER;
+        let mut request = Self::create_unlock_request(lock_id);
+        request.metadata.tags.insert(marker.to_owned(), marker.to_owned());
+        let mut client = self.get_client().await?;
+        let resource_summary = request.resource.to_string();
+        let mut req = Request::new(GenerallyLockRequest {
+            args: serde_json::to_string(&request)
+                .map_err(|error| LockError::internal(format!("Failed to serialize request: {error}")))?,
+        });
+        attach_lock_mutation_body_digest(&mut req)?;
+        let response = self
+            .execute_rpc(
+                "refresh",
+                &resource_summary,
+                Self::rpc_timeout(),
+                async move { client.refresh(req).await },
+                RpcCleanup::new(None, None, None),
+            )
+            .await?
+            .into_inner();
+        if let Some(error) = response.error_info {
+            return Err(LockError::internal(error));
+        }
+        // Older nodes refresh normally but return no conversion marker.
+        Ok(response.success && response.lock_info.as_deref() == Some(marker))
+    }
+
     async fn force_release(&self, lock_id: &LockId) -> Result<bool> {
         info!("remote force_release for {}", lock_id);
         let force_request = Self::create_unlock_request(lock_id);

crates/ecstore/src/set_disk/mod.rs

@@ -411,6 +411,10 @@ impl ObjectLockDiagGuard {
         self
     }
 
+    async fn downgrade_committed_tail(&mut self) -> bool {
+        self.guard.downgrade_committed_tail().await
+    }
+
     /// Whether the underlying namespace lock's heartbeat has observed a
     /// refresh-quorum loss (backlog#899 Phase 2). Callers fence their commit
     /// point on this so a stale lock holder does not race a double-write.
@@ -5154,6 +5158,12 @@ impl SetDisks {
         if request.expected_bucket_incarnation_id.is_none() {
             return Ok(());
         }
+        #[cfg(feature = "e2e-test-hooks")]
+        if let Some(object) = request.object_prefix.as_deref()
+            && skip_e2e_mrf_fallback(&request.bucket, object).await
+        {
+            return Ok(());
+        }
         request.source = rustfs_heal_contracts::heal_channel::HealRequestSource::Mrf;
         #[cfg(test)]
         {
@@ -5861,6 +5871,25 @@ impl SetDisks {
     // }
 }
 
+// The isolated root-heal test suppresses only the direct best-effort fallback.
+// Persisted MRF responsibilities and their replay keep their normal behavior.
+#[cfg(feature = "e2e-test-hooks")]
+pub(in crate::set_disk) async fn skip_e2e_mrf_fallback(bucket: &str, object: &str) -> bool {
+    let Some(path) = std::env::var_os("RUSTFS_E2E_SKIP_MRF_FALLBACK_PATH") else {
+        return false;
+    };
+    let path = std::path::PathBuf::from(path);
+    let Ok(prefix) = tokio::fs::read_to_string(&path).await else {
+        return false;
+    };
+    let key = format!("{bucket}/{object}");
+    if prefix.is_empty() || !key.starts_with(&prefix) {
+        return false;
+    }
+    let _ = tokio::fs::write(path.with_extension("observed"), key).await;
+    true
+}
+
 fn is_explicit_null_version(version_id: Option<Uuid>) -> bool {
     version_id == Some(Uuid::nil())
 }

crates/ecstore/src/set_disk/ops/heal.rs

@@ -1943,6 +1943,13 @@ impl SetDisks {
                             }
                         }
                         let _read_repair_write_lock_guard = if let Some(expected_fingerprint) = read_repair_commit_fingerprint {
+                            if _read_repair_read_lock_guard
+                                .as_ref()
+                                .is_some_and(|guard| guard.is_lock_lost())
+                            {
+                                let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
+                                return Err(DiskError::DiskOngoingReq);
+                            }
                             drop(_read_repair_read_lock_guard.take());
                             match self
                                 .acquire_revalidated_read_repair_commit_lock(
@@ -1985,7 +1992,12 @@ impl SetDisks {
                                 if let Some(scope) = &bucket_heal_scope {
                                     scope.check()?;
                                 }
-                                if commit_fence.is_some_and(|check| !check()) {
+                                if commit_fence.is_some_and(|check| !check())
+                                    || _write_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                                    || _read_repair_write_lock_guard
+                                        .as_ref()
+                                        .is_some_and(|guard| guard.is_lock_lost())
+                                {
                                     let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
                                     return Err(DiskError::DiskOngoingReq);
                                 }
@@ -2123,6 +2135,12 @@ impl SetDisks {
                         self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id)
                             .await
                             .map_err(DiskError::other)?;
+                        if _read_repair_write_lock_guard
+                            .as_ref()
+                            .is_some_and(|guard| guard.is_lock_lost())
+                        {
+                            return Err(DiskError::DiskOngoingReq);
+                        }
 
                         self.record_healed_capacity_scope(&healed_disks);
 
@@ -3625,6 +3643,46 @@ impl SetDisks {
         proof: AbsenceProofRequest<'_>,
     ) -> Result<(HealResultItem, Option<Error>)> {
         *absence = None;
+        if opts.read_repair
+            && !opts.no_lock
+            && !opts.dry_run
+            && !opts.remove
+            && !has_suffix(object, SLASH_SEPARATOR)
+            && !crate::bucket::utils::is_meta_bucketname(bucket)
+        {
+            // Read repair may stage under a shared lock, but must not perform
+            // destructive cleanup or produce an authoritative absence proof.
+            let can_commit = || proof.retirement.is_none_or(|context| !context.lifecycle_guard.is_lock_lost());
+            let mut read_opts = *opts;
+            loop {
+                let (result, error) = pin_object_heal_future(|| {
+                    self.heal_object_with_explicit_version_regen(
+                        bucket,
+                        object,
+                        version_id,
+                        ExplicitVersionHeal {
+                            opts: &read_opts,
+                            allow_regeneration: false,
+                            preserve_unrepairable: true,
+                            commit_fence: Some(&can_commit),
+                            namespace_owner: None,
+                            retirement: None,
+                        },
+                        absence,
+                    )
+                })
+                .await
+                .map_err(|error| to_object_err(error.into(), vec![bucket, object]))?;
+                if !can_commit() {
+                    return Err(DiskError::DiskOngoingReq.into());
+                }
+                if matches!(error, Some(DiskError::FileCorrupt)) && read_opts.scan_mode != HealScanMode::Deep {
+                    read_opts.scan_mode = HealScanMode::Deep;
+                    continue;
+                }
+                return Ok((result, error.map(Into::into)));
+            }
+        }
         let _write_lock_guard = if !opts.no_lock {
             let ns_lock = self
                 .new_ns_lock(bucket, object)
@@ -8185,7 +8243,7 @@ mod heal_result_report_tests {
             let heal_bucket = bucket.clone();
             let heal = tokio::spawn(async move {
                 heal_set
-                    .heal_object(
+                    .heal_object_with_retirement(
                         &heal_bucket,
                         object,
                         &selected_version.to_string(),
@@ -8194,6 +8252,11 @@ mod heal_result_report_tests {
                             read_repair: true,
                             ..Default::default()
                         },
+                        &mut None,
+                        AbsenceProofRequest {
+                            retirement: None,
+                            allow_unversioned: false,
+                        },
                     )
                     .await
             });
@@ -8285,7 +8348,7 @@ mod heal_result_report_tests {
                 assert!(outcome.expect("historical version remains healable").1.is_none());
             } else {
                 assert!(
-                    matches!(outcome, Err(DiskError::FileVersionNotFound)),
+                    matches!(outcome, Err(Error::VersionNotFound(..))),
                     "deleted version must remain typed absent: {outcome:?}"
                 );
             }
@@ -9877,6 +9940,123 @@ mod heal_result_report_tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn read_repair_storage_entry_keeps_readers_available_during_reconstruction() {
+        use crate::object_api::WriteCompletion;
+        use std::time::Duration;
+        use tokio::io::AsyncReadExt;
+
+        let (dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+        let bucket = "read-repair-storage-entry";
+        let object = "object.bin";
+        let expected = vec![0x37; 1024 * 1024];
+        set.make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        set.put_object(
+            bucket,
+            object,
+            &mut PutObjReader::from_vec(expected.clone()),
+            &ObjectOptions {
+                write_completion: WriteCompletion::TailDrained,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("write object");
+        let metadata = disks[0]
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("read metadata");
+        let missing = dirs[0]
+            .path()
+            .join(bucket)
+            .join(object)
+            .join(metadata.data_dir.expect("external object data").to_string())
+            .join("part.1");
+        tokio::fs::remove_file(&missing).await.expect("remove shard");
+
+        let (_pause, arrived, release) =
+            ReadRepairPauseScope::install_at(bucket, object, super::ReadRepairPausePhase::PartStaging(1));
+        let heal_set = set.clone();
+        let heal = tokio::spawn(async move {
+            let mut absence = None;
+            let result = heal_set
+                .heal_object_with_retirement(
+                    bucket,
+                    object,
+                    "",
+                    &HealOpts {
+                        scan_mode: HealScanMode::Deep,
+                        read_repair: true,
+                        ..Default::default()
+                    },
+                    &mut absence,
+                    AbsenceProofRequest {
+                        retirement: None,
+                        allow_unversioned: false,
+                    },
+                )
+                .await;
+            (result, absence)
+        });
+        tokio::time::timeout(Duration::from_secs(10), arrived.notified())
+            .await
+            .expect("read-repair reached shard reconstruction");
+        let namespace = set.new_ns_lock(bucket, object).await.expect("object namespace");
+        let reader = namespace
+            .get_read_lock(Duration::from_millis(200))
+            .await
+            .expect("read-repair reconstruction should allow concurrent readers");
+        drop(reader);
+        release.notify_one();
+        let (result, absence) = tokio::time::timeout(Duration::from_secs(20), heal)
+            .await
+            .expect("repair finishes")
+            .expect("repair task joins");
+        let (_, error) = result.expect("repair operation succeeds");
+        assert!(error.is_none(), "repair failed: {error:?}");
+        assert!(absence.is_none());
+        assert!(missing.exists(), "missing shard was not restored");
+        let mut reader = set
+            .get_object_reader(bucket, object, None, http::HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("read repaired object");
+        let mut actual = Vec::new();
+        reader.stream.read_to_end(&mut actual).await.expect("read full object");
+        assert_eq!(actual, expected);
+    }
+
+    #[tokio::test]
+    async fn read_repair_storage_entry_does_not_prove_missing_versions_absent() {
+        let (_dirs, _disks, set) = hermetic_set_disks_isolated(4).await;
+        let bucket = "read-repair-no-absence-proof";
+        set.make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        let mut absence = None;
+        let (_, error) = set
+            .heal_object_with_retirement(
+                bucket,
+                "missing.bin",
+                &Uuid::new_v4().to_string(),
+                &HealOpts {
+                    read_repair: true,
+                    ..Default::default()
+                },
+                &mut absence,
+                AbsenceProofRequest {
+                    retirement: None,
+                    allow_unversioned: false,
+                },
+            )
+            .await
+            .expect("missing version report");
+        assert!(matches!(error, Some(Error::FileVersionNotFound)));
+        assert!(absence.is_none(), "read repair must not publish an absence proof");
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn read_repair_commit_aborts_when_object_changes_before_final_rename() {
@@ -9922,7 +10102,7 @@ mod heal_result_report_tests {
         let heal_set = set.clone();
         let heal_task = tokio::spawn(async move {
             heal_set
-                .heal_object(
+                .heal_object_with_retirement(
                     bucket,
                     object,
                     "",
@@ -9931,6 +10111,11 @@ mod heal_result_report_tests {
                         read_repair: true,
                         ..Default::default()
                     },
+                    &mut None,
+                    AbsenceProofRequest {
+                        retirement: None,
+                        allow_unversioned: false,
+                    },
                 )
                 .await
         });
@@ -9951,7 +10136,7 @@ mod heal_result_report_tests {
             .expect("read-repair should finish after release")
             .expect("read-repair task should join")
             .expect("stale read-repair should report a typed retryable error");
-        assert_eq!(err, Some(DiskError::Timeout));
+        assert!(matches!(err, Some(Error::Timeout)), "stale repair must retry: {err:?}");
 
         let final_info = set
             .get_object_info(bucket, object, &ObjectOptions::default())

crates/ecstore/src/set_disk/ops/object.rs

@@ -5622,6 +5622,7 @@ impl SetDisks {
                     pause_put_object_commit(&commit_bucket, &commit_object, PutObjectCommitPause::AfterRenameQuorum).await;
                 }
                 let mut rename_guard_release = None;
+                let mut committed_tail_guard = None;
                 let mut needs_immediate_heal = false;
                 let mut tail_owns_tmp_cleanup = false;
                 rustfs_io_metrics::record_put_object_stage_duration_from(
@@ -5651,7 +5652,10 @@ impl SetDisks {
                             .or_else(|| commit_version_suspended.then(Uuid::nil))
                             .map(|version_id| version_id.to_string());
                         request.expected_bucket_incarnation_id = source_bucket_incarnation_id;
-                        let object_lock_guard = _object_lock_guard.take();
+                        let object_lock_guard = _object_lock_guard
+                            .take()
+                            .map(|guard| Arc::new(tokio::sync::Mutex::new(guard)));
+                        committed_tail_guard = object_lock_guard.clone();
                         let publication_guard = _publication_guard.take();
                         let bucket_lifecycle_guard = _bucket_lifecycle_guard.take();
                         let decommission_object_lock_guard = _decommission_object_lock_guard.take();
@@ -5957,6 +5961,13 @@ impl SetDisks {
                     lock_held_post_rename_started,
                 );
                 let lock_held_guard_release_started = rustfs_io_metrics::put_stage_timer();
+                if let Some(guard) = committed_tail_guard.take() {
+                    // The committed tail still owns every physical rename. Convert
+                    // only after metadata invalidation and before acknowledging PUT.
+                    // Unsupported peers keep their exclusive leases and the tail
+                    // retains ownership until its physical work has drained.
+                    let _ = guard.lock().await.downgrade_committed_tail().await;
+                }
                 if let Some(release) = rename_guard_release.take() {
                     let _ = release.send(true);
                 }
@@ -9262,6 +9273,10 @@ impl SetDisks {
             // cannot safely target the object name after bucket recreation.
             return Ok(());
         };
+        #[cfg(feature = "e2e-test-hooks")]
+        if crate::set_disk::skip_e2e_mrf_fallback(bucket, object).await {
+            return Ok(());
+        }
         let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options(
             bucket.to_string(),
             Some(object.to_string()),
@@ -22156,6 +22171,52 @@ mod put_object_tmp_cleanup_tests {
         .await;
     }
 
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn early_ack_tail_drain_allows_read_after_acknowledged_put() {
+        temp_env::async_with_vars(
+            [
+                (ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true")),
+                (rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("1")),
+            ],
+            async {
+                let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+                let bucket = "put-early-ack-tail-read";
+                let object = "early-ack-tail-read-object";
+                for disk in &disk_stores {
+                    disk.make_volume(bucket).await.expect("bucket volume should be created");
+                }
+                let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+                let put_set = Arc::clone(&set_disks);
+                let put = tokio::spawn(async move {
+                    let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+                    put_set
+                        .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+                        .await
+                });
+                tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
+                    .await
+                    .expect("tail disk should pause during rename");
+                put.await
+                    .expect("PUT task should join")
+                    .expect("PUT should be acknowledged at quorum");
+                let result = set_disks
+                    .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                    .await;
+                let mut reader = result.expect("acknowledged object should remain readable while tail drains");
+                let mut body = Vec::new();
+                tokio::time::timeout(Duration::from_secs(5), reader.stream.read_to_end(&mut body))
+                    .await
+                    .expect("object body should not wait for the tail")
+                    .expect("object body should drain");
+                assert_eq!(body, vec![b'1'; TEST_OBJECT_SIZE]);
+                rename_barrier.release();
+                drop(rename_barrier);
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     #[serial_test::serial(capacity_dirty_scope)]
     async fn early_ack_tail_drain_retains_namespace_lock_until_background_rename_finishes() {

crates/ecstore/src/store/heal.rs

@@ -2845,9 +2845,18 @@ mod tests {
         let (configuration, _) = read_config_no_lock_preserve_empty_with_metadata(store.clone(), &object)
             .await
             .expect("read the current bucket configuration");
-        save_config(store.clone(), &object, configuration)
-            .await
-            .expect("publish a newer config revision in the same bucket incarnation");
+        crate::config::com::save_config_with_opts(
+            store.clone(),
+            &object,
+            configuration,
+            &ObjectOptions {
+                max_parity: true,
+                write_completion: crate::object_api::WriteCompletion::TailDrained,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("publish a newer config revision on the target disk");
         let committed =
             std::fs::read(target_path.join(RUSTFS_META_BUCKET).join(&object).join("xl.meta")).expect("new target configuration");
         assert!(

crates/lock/src/client/local.rs

@@ -382,6 +382,18 @@ impl LockClient for LocalClient {
         }
     }
 
+    async fn downgrade_committed_tail(&self, lock_id: &LockId) -> Result<bool> {
+        let shard = self.get_shard(lock_id);
+        let mut guards = shard.write().await;
+        let Some(entry) = guards.get_mut(lock_id) else {
+            return Ok(false);
+        };
+        if entry.is_expired() {
+            return Ok(false);
+        }
+        Ok(entry.guard.downgrade_committed_tail())
+    }
+
     async fn force_release(&self, lock_id: &LockId) -> Result<bool> {
         self.release(lock_id).await
     }
@@ -547,6 +559,32 @@ mod tests {
         assert!(!response.success);
     }
 
+    #[tokio::test]
+    async fn downgrade_tail_is_idempotent_and_release_preserves_readers() {
+        let client = LocalClient::with_manager(Arc::new(GlobalLockManager::new()));
+        let resource = crate::ObjectKey::new("bucket", "tail-downgrade");
+        let first = request(resource.clone(), "owner-a", Duration::from_secs(30));
+        assert!(client.acquire_lock(&first).await.expect("initial lock").success);
+        assert!(client.downgrade_committed_tail(&first.lock_id).await.expect("downgrade"));
+        assert!(
+            client
+                .downgrade_committed_tail(&first.lock_id)
+                .await
+                .expect("retry downgrade")
+        );
+        let reader = crate::LockRequest::new(resource.clone(), LockType::Shared, "reader")
+            .with_acquire_timeout(Duration::from_millis(500));
+        assert!(client.acquire_lock(&reader).await.expect("reader").success);
+        let writer = request(resource, "writer", Duration::from_secs(30)).with_acquire_timeout(Duration::from_millis(10));
+        assert!(!client.acquire_lock(&writer).await.expect("writer blocked").success);
+        assert!(client.release(&first.lock_id).await.expect("release tail"));
+        assert!(!client.acquire_lock(&writer).await.expect("reader excludes writer").success);
+        assert!(client.release(&reader.lock_id).await.expect("release reader"));
+        assert!(client.acquire_lock(&writer).await.expect("writer after drain").success);
+        assert!(client.release(&writer.lock_id).await.expect("release writer"));
+        assert!(!client.downgrade_committed_tail(&first.lock_id).await.expect("released lease"));
+    }
+
     #[tokio::test]
     async fn acquire_rejects_a_lock_id_for_another_resource() {
         let client = LocalClient::with_manager(Arc::new(GlobalLockManager::new()));

crates/lock/src/client/mod.rs

@@ -20,6 +20,9 @@ use async_trait::async_trait;
 use futures::future::join_all;
 use std::sync::Arc;
 
+/// Marker exchanged in a refresh request and response for a committed-tail conversion.
+pub const COMMITTED_TAIL_CONVERSION_MARKER: &str = "rustfs-committed-tail-v1";
+
 /// Lock client trait
 #[async_trait]
 pub trait LockClient: Send + Sync + std::fmt::Debug {
@@ -48,6 +51,12 @@ pub trait LockClient: Send + Sync + std::fmt::Debug {
     /// Refresh lock
     async fn refresh(&self, lock_id: &LockId) -> Result<bool>;
 
+    /// Convert an exclusive lease to a reader-compatible committed-tail lease.
+    /// Unsupported peers retain their original lease and report `false`.
+    async fn downgrade_committed_tail(&self, _lock_id: &LockId) -> Result<bool> {
+        Ok(false)
+    }
+
     /// Force release lock
     async fn force_release(&self, lock_id: &LockId) -> Result<bool>;
 

crates/lock/src/distributed_lock.rs

@@ -262,6 +262,8 @@ pub struct DistributedLockGuard {
     /// All underlying leases that should be refreshed and released with the guard.
     entries: Vec<HeldLockEntry>,
     lock_type: LockType,
+    /// Shared and exclusive quorums must intersect before a tail may convert.
+    committed_tail_quorums_intersect: bool,
     /// If true, Drop will not try to release (used if user manually released).
     disarmed: bool,
     /// Background lease task for renewable leases.
@@ -277,7 +279,13 @@ impl DistributedLockGuard {
     /// - `entries` is the full list of underlying leases that should be refreshed and released
     ///   with this guard.
     /// - `heartbeat`: lease TTL, optional renewal cadence, quorum, and diagnostics.
-    fn new(lock_id: LockId, entries: Vec<HeldLockEntry>, lock_type: LockType, heartbeat: LockHeartbeatConfig) -> Self {
+    fn new(
+        lock_id: LockId,
+        entries: Vec<HeldLockEntry>,
+        lock_type: LockType,
+        heartbeat: LockHeartbeatConfig,
+        committed_tail_quorums_intersect: bool,
+    ) -> Self {
         record_lock_held_acquire(lock_type);
         let lock_lost = Arc::new(LockLostSignal::default());
         match Self::quorum_valid_until(&entries, heartbeat.refresh_quorum) {
@@ -296,6 +304,7 @@ impl DistributedLockGuard {
             lock_id,
             entries,
             lock_type,
+            committed_tail_quorums_intersect,
             disarmed: false,
             refresh_task,
             lock_lost,
@@ -412,6 +421,23 @@ impl DistributedLockGuard {
         &self.lock_id
     }
 
+    /// Convert retained peer leases without ever releasing writer exclusion.
+    /// A peer without conversion support keeps its exclusive lease. Partial
+    /// conversion is safe: both lease modes continue to exclude writers.
+    pub async fn downgrade_committed_tail(&mut self) -> bool {
+        if self.disarmed || self.lock_type != LockType::Exclusive || !self.committed_tail_quorums_intersect || self.is_lock_lost()
+        {
+            return false;
+        }
+        let results = futures::future::join_all(
+            self.entries
+                .iter()
+                .map(|entry| entry.client.downgrade_committed_tail(&entry.lock_id)),
+        )
+        .await;
+        !self.is_lock_lost() && results.iter().all(|result| matches!(result, Ok(true)))
+    }
+
     /// Actual granted peer identities, which may differ from the logical guard ID.
     pub fn granted_lock_ids(&self) -> impl Iterator<Item = &LockId> {
         self.entries.iter().map(|entry| &entry.lock_id)
@@ -506,6 +532,7 @@ struct LockAcquireQuorumResult {
     individual_locks: Vec<HeldLockEntry>,
     failure_kind: Option<LockAcquireFailureKind>,
     quorum_impossible: bool,
+    attempt_timed_out: bool,
 }
 
 impl DistributedLock {
@@ -598,6 +625,7 @@ impl DistributedLock {
                 individual_locks,
                 request.lock_type,
                 heartbeat,
+                self.read_quorum().saturating_add(required_quorum) > self.clients.len(),
             )))
         } else {
             // Check if it's a timeout or quorum failure
@@ -886,9 +914,9 @@ impl DistributedLock {
     }
 
     async fn acquire_lock_quorum_with_retry(&self, request: &LockRequest) -> Result<LockAcquireQuorumResult> {
-        let start = std::time::Instant::now();
+        let start = tokio::time::Instant::now();
         let mut attempt = 1;
-        let mut last_result = None;
+        let mut last_result: Option<LockAcquireQuorumResult> = None;
 
         loop {
             let elapsed = start.elapsed();
@@ -906,13 +934,22 @@ impl DistributedLock {
                 return Ok(result);
             }
 
-            last_result = Some(result);
-            let backoff = Self::lock_acquire_retry_backoff(attempt, &mut rand::rng());
-            if start.elapsed().saturating_add(backoff) >= request.acquire_timeout {
+            // A final retry can time out without establishing whether the
+            // previously observed quorum failure has recovered.
+            let remaining = request.acquire_timeout.saturating_sub(start.elapsed());
+            let preserve_quorum_failure = result.attempt_timed_out
+                && remaining.is_zero()
+                && last_result.as_ref().is_some_and(|result| result.quorum_impossible);
+            if !preserve_quorum_failure {
+                last_result = Some(result);
+            }
+            if remaining.is_zero() {
                 break;
             }
-
-            tokio::time::sleep(backoff).await;
+            let backoff = Self::lock_acquire_retry_backoff(attempt, &mut rand::rng());
+            // Keep time for another acquire attempt instead of abandoning the
+            // caller's remaining budget when the jittered backoff grows large.
+            tokio::time::sleep(backoff.min(remaining / 2)).await;
             attempt += 1;
         }
 
@@ -927,6 +964,7 @@ impl DistributedLock {
                 individual_locks: Vec::new(),
                 failure_kind: Some(LockAcquireFailureKind::RetryableContention),
                 quorum_impossible: false,
+                attempt_timed_out: true,
             })
         }
     }
@@ -937,6 +975,7 @@ impl DistributedLock {
             individual_locks: Vec::new(),
             failure_kind: Some(LockAcquireFailureKind::RetryableContention),
             quorum_impossible: false,
+            attempt_timed_out: true,
         }
     }
 
@@ -966,6 +1005,7 @@ impl DistributedLock {
             individual_locks,
             failure_kind: Some(failure_kind),
             quorum_impossible: matches!(failure_kind, LockAcquireFailureKind::UnrecoverableQuorum),
+            attempt_timed_out: true,
         }
     }
 
@@ -1134,6 +1174,7 @@ impl DistributedLock {
                     individual_locks,
                     failure_kind: Some(failure_kind),
                     quorum_impossible: true,
+                    attempt_timed_out: false,
                 });
             }
 
@@ -1176,6 +1217,7 @@ impl DistributedLock {
                     individual_locks,
                     failure_kind: None,
                     quorum_impossible: false,
+                    attempt_timed_out: false,
                 });
             }
 
@@ -1221,6 +1263,7 @@ impl DistributedLock {
                     individual_locks,
                     failure_kind: Some(failure_kind),
                     quorum_impossible,
+                    attempt_timed_out: false,
                 });
             }
         }
@@ -1245,6 +1288,7 @@ impl DistributedLock {
                 last_failure_kind.unwrap_or(LockAcquireFailureKind::RetryableContention)
             }),
             quorum_impossible: hard_failures > 0,
+            attempt_timed_out: false,
         })
     }
 }
@@ -2246,6 +2290,87 @@ mod tests {
         );
     }
 
+    fn clients_with_initial_quorum_loss(retries: &[AcquirePlan]) -> Vec<Arc<dyn LockClient>> {
+        (0..4)
+            .map(|index| {
+                let first = if index < 2 {
+                    AcquirePlan::Success { delay: Duration::ZERO }
+                } else {
+                    AcquirePlan::Failure {
+                        error: "Remote lock RPC failed: node unavailable",
+                        delay: Duration::ZERO,
+                    }
+                };
+                let mut plans = vec![first];
+                plans.extend_from_slice(retries);
+                Arc::new(SequencedClient::new(plans, Arc::new(Mutex::new(Vec::new())))) as Arc<dyn LockClient>
+            })
+            .collect()
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn acquire_guard_preserves_quorum_failure_when_last_retry_times_out() {
+        let clients = clients_with_initial_quorum_loss(&[AcquirePlan::Success {
+            delay: Duration::from_secs(1),
+        }]);
+        let lock = DistributedLock::new("test".to_string(), clients, 3);
+        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
+            .with_acquire_timeout(Duration::from_millis(180));
+
+        let result = lock.acquire_guard(&request).await;
+        assert!(
+            matches!(result, Err(LockError::QuorumNotReached { required: 3, .. })),
+            "a late inconclusive retry must not hide an observed quorum failure: {result:?}"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn acquire_guard_preserves_quorum_failure_when_retry_backoff_fits_but_attempt_times_out() {
+        let clients = clients_with_initial_quorum_loss(&[AcquirePlan::Success {
+            delay: Duration::from_secs(1),
+        }]);
+        let lock = DistributedLock::new("test".to_string(), clients, 3);
+        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
+            .with_acquire_timeout(Duration::from_millis(500));
+
+        let result = lock.acquire_guard(&request).await;
+        assert!(
+            matches!(result, Err(LockError::QuorumNotReached { required: 3, .. })),
+            "an inconclusive final attempt must not hide an observed quorum failure: {result:?}"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn acquire_guard_recovers_quorum_with_shortened_retry() {
+        let clients = clients_with_initial_quorum_loss(&[AcquirePlan::Success { delay: Duration::ZERO }]);
+        let lock = DistributedLock::new("test".to_string(), clients, 3);
+        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
+            .with_acquire_timeout(Duration::from_millis(180));
+
+        let guard = lock
+            .acquire_guard(&request)
+            .await
+            .expect("a retry should recover from quorum loss")
+            .expect("a recovered quorum should acquire the lock");
+        assert_eq!(guard.entries.len(), 3);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn acquire_guard_replaces_quorum_failure_with_later_contention() {
+        let clients = clients_with_initial_quorum_loss(
+            &[AcquirePlan::Failure {
+                error: "Lock acquisition timeout",
+                delay: Duration::ZERO,
+            }; 64],
+        );
+        let lock = DistributedLock::new("test".to_string(), clients, 3);
+        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
+            .with_acquire_timeout(Duration::from_millis(180));
+
+        let result = lock.acquire_guard(&request).await;
+        assert!(matches!(result, Ok(None)), "later contention should replace the earlier quorum failure");
+    }
+
     #[tokio::test]
     async fn acquire_guard_retries_rpc_timeouts_until_caller_timeout() {
         let clients: Vec<Arc<dyn LockClient>> = vec![

crates/lock/src/fast_lock/guard.rs

@@ -39,6 +39,7 @@ pub struct FastLockGuard {
     shard: Option<Arc<LockShard>>, // None when locks are disabled
     released: bool,
     disabled: bool, // True when locks are disabled globally
+    committed_tail: bool,
     /// Unique ID for this guard instance to prevent double-release
     guard_id: u64,
 }
@@ -54,6 +55,7 @@ impl FastLockGuard {
             shard: Some(shard),
             released: false,
             disabled: false,
+            committed_tail: false,
             guard_id,
         }
     }
@@ -68,6 +70,7 @@ impl FastLockGuard {
             shard: None,
             released: false,
             disabled: true,
+            committed_tail: false,
             guard_id,
         }
     }
@@ -87,6 +90,49 @@ impl FastLockGuard {
         &self.owner
     }
 
+    /// The physical acquisition was force-released before registration.
+    pub(crate) fn abandon_unregistered(&mut self) {
+        if !self.released {
+            self.released = true;
+            record_lock_held_release(self.mode);
+        }
+    }
+
+    /// Keep excluding writers while permitting reads of the committed version.
+    pub fn downgrade_committed_tail(&mut self) -> bool {
+        if self.released {
+            return false;
+        }
+        if self.committed_tail {
+            return true;
+        }
+        if self.mode != LockMode::Exclusive {
+            return false;
+        }
+        if self.disabled {
+            self.mode = LockMode::Shared;
+            self.committed_tail = true;
+            return true;
+        }
+        let Some(shard) = &self.shard else { return false };
+        if !shard.downgrade_committed_tail(&self.key, &self.owner, self.guard_id) {
+            return false;
+        }
+        record_lock_held_release(LockMode::Exclusive);
+        record_lock_held_acquire(LockMode::Shared);
+        self.mode = LockMode::Shared;
+        self.committed_tail = true;
+        true
+    }
+
+    fn release_physical(&self, shard: &LockShard) -> bool {
+        if self.committed_tail {
+            shard.release_committed_tail_with_guard(&self.key, &self.owner, self.guard_id)
+        } else {
+            shard.release_lock_with_guard(&self.key, &self.owner, self.mode, self.guard_id)
+        }
+    }
+
     /// Manually release the lock early
     ///
     /// Returns true if the lock was successfully released, false if it was
@@ -106,7 +152,7 @@ impl FastLockGuard {
         }
 
         if let Some(shard) = &self.shard {
-            let success = shard.release_lock_with_guard(&self.key, &self.owner, self.mode, self.guard_id);
+            let success = self.release_physical(shard);
             if success {
                 self.released = true;
                 record_lock_held_release(self.mode);
@@ -169,7 +215,7 @@ impl Drop for FastLockGuard {
     fn drop(&mut self) {
         if let Some(shard) = &self.shard {
             if !self.released && !self.disabled {
-                let success = shard.release_lock_with_guard(&self.key, &self.owner, self.mode, self.guard_id);
+                let success = self.release_physical(shard);
                 if !success {
                     // For high-concurrency scenarios, this is likely due to:
                     // 1. Lock was already released by another thread

crates/lock/src/fast_lock/manager.rs

@@ -96,11 +96,14 @@ impl FastObjectLockManager {
     /// Acquire object lock
     pub async fn acquire_lock(&self, request: ObjectLockRequest) -> Result<FastLockGuard, LockResult> {
         let shard = self.get_shard(&request.key);
-        match shard.acquire_lock(&request).await {
-            Ok(()) => {
-                let guard = FastLockGuard::new(request.key, request.mode, request.owner, shard.clone());
+        match shard.acquire_guarded(&request).await {
+            Ok(acquired) => {
+                let mut guard = FastLockGuard::new(request.key, request.mode, request.owner, shard.clone());
                 // Register guard to prevent premature cleanup
-                shard.register_guard_with_info(guard.guard_id(), guard.key(), guard.mode(), guard.owner());
+                if !shard.register_guard_with_info(guard.guard_id(), guard.key(), guard.mode(), guard.owner(), &acquired) {
+                    guard.abandon_unregistered();
+                    return Err(LockResult::Timeout);
+                }
                 Ok(guard)
             }
             Err(err) => Err(err),
@@ -201,31 +204,15 @@ impl FastObjectLockManager {
         let mut all_failed = Vec::new();
         let mut guards = Vec::new();
 
-        for (shard_id, requests) in shard_groups {
-            let shard = self.shards[*shard_id].clone();
-
+        for (_, requests) in shard_groups {
             for request in requests {
                 let key = request.key.clone();
-                let owner = request.owner.clone();
-                let mode = request.mode;
-
-                let acquired = if shard.try_fast_path_only(request) {
-                    true
-                } else {
-                    match shard.acquire_lock(request).await {
-                        Ok(()) => true,
-                        Err(err) => {
-                            all_failed.push((key.clone(), err));
-                            false
-                        }
+                match self.acquire_lock(request.clone()).await {
+                    Ok(guard) => {
+                        all_successful.push(key);
+                        guards.push(guard);
                     }
-                };
-
-                if acquired {
-                    let guard = FastLockGuard::new(key.clone(), mode, owner.clone(), shard.clone());
-                    shard.register_guard_with_info(guard.guard_id(), guard.key(), guard.mode(), guard.owner());
-                    all_successful.push(key);
-                    guards.push(guard);
+                    Err(err) => all_failed.push((key, err)),
                 }
             }
         }
@@ -245,14 +232,10 @@ impl FastObjectLockManager {
         let mut acquired_guards = Vec::new();
         let mut failed_locks = Vec::new();
 
-        'outer: for (shard_id, requests) in shard_groups {
-            let shard = self.shards[*shard_id].clone();
-
+        'outer: for (_, requests) in shard_groups {
             for request in requests {
-                match shard.acquire_lock(request).await {
-                    Ok(()) => {
-                        let guard = FastLockGuard::new(request.key.clone(), request.mode, request.owner.clone(), shard.clone());
-                        shard.register_guard_with_info(guard.guard_id(), guard.key(), guard.mode(), guard.owner());
+                match self.acquire_lock(request.clone()).await {
+                    Ok(guard) => {
                         acquired_guards.push(guard);
                     }
                     Err(err) => {

crates/lock/src/fast_lock/object_pool.rs

@@ -94,6 +94,7 @@ impl ObjectLockState {
     pub fn reset_for_reuse(&mut self) {
         // Reset atomic state
         self.atomic_state = crate::fast_lock::state::AtomicLockState::new();
+        self.force_epoch.store(0, Ordering::Relaxed);
 
         // Clear owners
         *self.current_owner.write() = None;

crates/lock/src/fast_lock/shard.rs

@@ -15,6 +15,7 @@
 use parking_lot::RwLock;
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::sync::atomic::Ordering;
 use std::time::{Duration, Instant, SystemTime};
 use tokio::time::timeout;
 
@@ -30,6 +31,12 @@ struct ActiveGuardInfo {
     key: ObjectKey,
     mode: LockMode,
     owner: Arc<str>,
+    committed_tail: bool,
+}
+
+pub(crate) struct AcquiredLock {
+    state: Arc<ObjectLockState>,
+    force_epoch: u64,
 }
 
 #[derive(Debug, PartialEq, Eq, Hash)]
@@ -46,6 +53,8 @@ enum TestPausePoint {
     BeforeFastWrite,
     FastWriteExisting,
     SlowLookup,
+    ForceAfterTailRelease,
+    ReleaseAfterUnregister,
 }
 
 #[cfg(test)]
@@ -69,7 +78,7 @@ fn test_pause(point: TestPausePoint, key: &ObjectKey) {
 
 /// Lock shard to reduce global contention.
 ///
-/// When both are needed, lock `objects` before a state's owner metadata.
+/// When combined, lock `active_guards`, then `objects`, then a state's owner metadata.
 #[derive(Debug)]
 pub struct LockShard {
     /// Object lock states - using parking_lot for better performance
@@ -133,12 +142,16 @@ impl LockShard {
 
     /// Acquire lock with fast path optimization
     pub async fn acquire_lock(&self, request: &ObjectLockRequest) -> Result<(), LockResult> {
+        self.acquire_guarded(request).await.map(|_| ())
+    }
+
+    pub(crate) async fn acquire_guarded(&self, request: &ObjectLockRequest) -> Result<AcquiredLock, LockResult> {
         let start_time = Instant::now();
 
         // Try fast path first
-        if let Some(_state) = self.try_fast_path(request) {
+        if let Some(acquired) = self.try_fast_path(request) {
             self.metrics.record_fast_path_success();
-            return Ok(());
+            return Ok(acquired);
         }
 
         // Slow path with waiting
@@ -157,7 +170,7 @@ impl LockShard {
     }
 
     /// Try fast path lock acquisition (lock-free when possible)
-    fn try_fast_path(&self, request: &ObjectLockRequest) -> Option<Arc<ObjectLockState>> {
+    fn try_fast_path(&self, request: &ObjectLockRequest) -> Option<AcquiredLock> {
         // First try to get existing state without write lock
         {
             let objects = self.objects.read();
@@ -168,14 +181,16 @@ impl LockShard {
                 test_pause(TestPausePoint::FastExisting, &request.key);
 
                 // Try atomic acquisition
-                let success = match request.mode {
-                    LockMode::Shared => state.try_acquire_shared_fast(&request.owner, request.lock_timeout),
-                    LockMode::Exclusive => state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout),
-                };
+                let force_epoch = state.force_epoch.load(Ordering::Acquire);
+                let success = force_epoch != u64::MAX
+                    && match request.mode {
+                        LockMode::Shared => state.try_acquire_shared_fast(&request.owner, request.lock_timeout),
+                        LockMode::Exclusive => state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout),
+                    };
                 drop(objects);
 
                 if success {
-                    return Some(state);
+                    return Some(AcquiredLock { state, force_epoch });
                 }
             }
         }
@@ -195,10 +210,11 @@ impl LockShard {
                 #[cfg(test)]
                 test_pause(TestPausePoint::FastWriteExisting, &request.key);
 
-                let acquired = state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout);
+                let force_epoch = state.force_epoch.load(Ordering::Acquire);
+                let acquired = force_epoch != u64::MAX && state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout);
                 drop(objects);
                 if acquired {
-                    return Some(state);
+                    return Some(AcquiredLock { state, force_epoch });
                 }
             } else {
                 // Create new state from pool and acquire immediately
@@ -206,7 +222,7 @@ impl LockShard {
                 let state = Arc::new(*state_box);
                 if state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout) {
                     objects.insert(request.key.clone(), state.clone());
-                    return Some(state);
+                    return Some(AcquiredLock { state, force_epoch: 0 });
                 }
             }
         }
@@ -215,7 +231,7 @@ impl LockShard {
     }
 
     /// Slow path with async waiting
-    async fn acquire_lock_slow_path(&self, request: &ObjectLockRequest, start_time: Instant) -> Result<(), LockResult> {
+    async fn acquire_lock_slow_path(&self, request: &ObjectLockRequest, start_time: Instant) -> Result<AcquiredLock, LockResult> {
         // Use adaptive timeout based on current load and request priority
         let adaptive_timeout = self.calculate_adaptive_timeout(request);
         let deadline = start_time + adaptive_timeout;
@@ -234,7 +250,7 @@ impl LockShard {
 
         loop {
             // Keep the state in the map until its acquisition attempt completes.
-            let (state, success) = {
+            let (state, success, force_epoch) = {
                 let mut objects = self.objects.write();
                 let state = match objects.get(&request.key) {
                     Some(state) => state.clone(),
@@ -249,17 +265,19 @@ impl LockShard {
 
                 #[cfg(test)]
                 test_pause(TestPausePoint::SlowLookup, &request.key);
-                let success = match request.mode {
-                    LockMode::Shared => state.try_acquire_shared_fast(&request.owner, request.lock_timeout),
-                    LockMode::Exclusive => state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout),
-                };
+                let force_epoch = state.force_epoch.load(Ordering::Acquire);
+                let success = force_epoch != u64::MAX
+                    && match request.mode {
+                        LockMode::Shared => state.try_acquire_shared_fast(&request.owner, request.lock_timeout),
+                        LockMode::Exclusive => state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout),
+                    };
                 drop(objects);
-                (state, success)
+                (state, success, force_epoch)
             };
 
             if success {
                 self.metrics.record_slow_path_success();
-                return Ok(());
+                return Ok(AcquiredLock { state, force_epoch });
             }
 
             // Check timeout
@@ -381,14 +399,10 @@ impl LockShard {
 
     /// Release lock with guard ID tracking for double-release prevention
     pub fn release_lock_with_guard(&self, key: &ObjectKey, owner: &Arc<str>, mode: LockMode, guard_id: u64) -> bool {
-        // First, try to remove the guard from active set
-        let guard_was_active = {
-            let mut guards = self.active_guards.lock();
-            guards.remove(&guard_id).is_some()
-        };
-
-        // If guard was not active, this is a double-release attempt
-        if !guard_was_active {
+        // Keep force-unlock from releasing this holder and admitting a new
+        // same-owner holder between guard removal and physical release.
+        let mut guards = self.active_guards.lock();
+        if guards.remove(&guard_id).is_none() {
             tracing::debug!(
                 "Double-release attempt blocked: key={}, owner={}, mode={:?}, guard_id={}",
                 key,
@@ -399,6 +413,9 @@ impl LockShard {
             return false;
         }
 
+        #[cfg(test)]
+        test_pause(TestPausePoint::ReleaseAfterUnregister, key);
+
         // Proceed with normal release
         let should_cleanup;
         let result;
@@ -423,6 +440,8 @@ impl LockShard {
             }
         }
 
+        drop(guards);
+
         if should_cleanup {
             self.schedule_cleanup(key.clone());
         }
@@ -430,21 +449,77 @@ impl LockShard {
         result
     }
 
+    /// Convert an active exclusive guard without opening a writer gap.
+    pub(crate) fn downgrade_committed_tail(&self, key: &ObjectKey, owner: &Arc<str>, guard_id: u64) -> bool {
+        let mut guards = self.active_guards.lock();
+        let Some(Some(info)) = guards.get_mut(&guard_id) else { return false };
+        if info.key != *key || info.owner.as_ref() != owner.as_ref() || info.mode != LockMode::Exclusive {
+            return false;
+        }
+        let objects = self.objects.read();
+        let Some(state) = objects.get(key) else { return false };
+        if !state.downgrade_committed_tail(owner) {
+            return false;
+        }
+        info.mode = LockMode::Shared;
+        info.committed_tail = true;
+        true
+    }
+
+    /// Release a converted guard and its special reader-admission state.
+    pub(crate) fn release_committed_tail_with_guard(&self, key: &ObjectKey, owner: &Arc<str>, guard_id: u64) -> bool {
+        let mut guards = self.active_guards.lock();
+        let Some(Some(info)) = guards.get(&guard_id) else { return false };
+        if info.key != *key || info.owner.as_ref() != owner.as_ref() || info.mode != LockMode::Shared || !info.committed_tail {
+            return false;
+        }
+        let (success, should_cleanup) = {
+            let objects = self.objects.read();
+            let Some(state) = objects.get(key) else { return false };
+            let success = state.release_committed_tail(owner);
+            (success, success && !state.is_locked() && !state.atomic_state.has_waiters())
+        };
+        if success {
+            guards.remove(&guard_id);
+            self.metrics.record_release();
+        }
+        drop(guards);
+        if should_cleanup {
+            self.schedule_cleanup(key.clone());
+        }
+        success
+    }
+
     /// Register a guard to prevent premature cleanup
     pub fn register_guard(&self, guard_id: u64) {
         self.active_guards.lock().insert(guard_id, None);
     }
 
-    pub(crate) fn register_guard_with_info(&self, guard_id: u64, key: &ObjectKey, mode: LockMode, owner: &Arc<str>) {
+    pub(crate) fn register_guard_with_info(
+        &self,
+        guard_id: u64,
+        key: &ObjectKey,
+        mode: LockMode,
+        owner: &Arc<str>,
+        acquired: &AcquiredLock,
+    ) -> bool {
         let mut guards = self.active_guards.lock();
+        let objects = self.objects.read();
+        if !objects.get(key).is_some_and(|state| {
+            Arc::ptr_eq(state, &acquired.state) && state.force_epoch.load(Ordering::Acquire) == acquired.force_epoch
+        }) {
+            return false;
+        }
         guards.insert(
             guard_id,
             Some(ActiveGuardInfo {
                 key: key.clone(),
                 mode,
                 owner: owner.clone(),
+                committed_tail: false,
             }),
         );
+        true
     }
 
     /// Unregister a guard (called when guard is dropped)
@@ -719,27 +794,57 @@ impl LockShard {
     /// Returns the number of owners that were released. Used by the admin
     /// force-unlock path to clear a stuck resource.
     pub fn force_release_all(&self, key: &ObjectKey) -> usize {
-        let owners_modes: Vec<(Arc<str>, LockMode)> = {
-            let objects = self.objects.read();
-            let Some(state) = objects.get(key) else {
-                return 0;
-            };
-            let mut pairs = Vec::new();
-            if let Some(info) = state.current_owner.read().clone() {
-                pairs.push((info.owner, LockMode::Exclusive));
-            }
-            for entry in state.shared_owners.read().iter() {
-                pairs.push((entry.owner.clone(), LockMode::Shared));
-            }
-            pairs
-        };
-
+        // Serialize tail conversion and prevent new acquisitions while taking
+        // and releasing the owner snapshot. A new tail must never be mistaken
+        // for an ordinary shared owner by force unlock.
+        let mut guards = self.active_guards.lock();
+        let objects = self.objects.write();
+        let Some(state) = objects.get(key) else { return 0 };
+        // Acquisitions completed before this point but not yet registered
+        // must not later create guards for the released generation.
+        let _ = state
+            .force_epoch
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |epoch| epoch.checked_add(1));
+        let tail = guards.iter().find_map(|(&id, info)| {
+            info.as_ref()
+                .filter(|info| info.committed_tail && info.key == *key)
+                .map(|info| (id, info.owner.clone()))
+        });
         let mut released = 0;
-        for (owner, mode) in owners_modes {
-            if self.release_lock(key, &owner, mode) {
+        if let Some((id, owner)) = tail
+            && state.release_committed_tail(&owner)
+        {
+            guards.remove(&id);
+            self.metrics.record_release();
+            released += 1;
+        }
+        #[cfg(test)]
+        test_pause(TestPausePoint::ForceAfterTailRelease, key);
+        let mut owners_modes = Vec::new();
+        if let Some(info) = state.current_owner.read().clone() {
+            owners_modes.push((info.owner, LockMode::Exclusive, 1));
+        }
+        for entry in state.shared_owners.read().iter() {
+            owners_modes.push((entry.owner.clone(), LockMode::Shared, entry.count));
+        }
+        for (owner, mode, count) in owners_modes {
+            let mut owner_released = false;
+            for _ in 0..count {
+                let success = match mode {
+                    LockMode::Shared => state.release_shared(&owner),
+                    LockMode::Exclusive => state.release_exclusive(&owner),
+                };
+                if success {
+                    self.metrics.record_release();
+                    owner_released = true;
+                }
+            }
+            if owner_released {
                 released += 1;
             }
         }
+        // Old guard drops must not release a later holder with the same owner.
+        guards.retain(|_, info| info.as_ref().is_none_or(|info| info.key != *key));
         released
     }
 
@@ -979,6 +1084,175 @@ mod tests {
         )
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn force_unlock_cannot_interleave_an_old_release_with_a_new_holder() {
+        for reader in [false, true] {
+            let manager = test_manager();
+            let key = ObjectKey::new("bucket", "force-release-gap");
+            let old = if reader {
+                manager.acquire_read_lock(key.clone(), "same-owner").await
+            } else {
+                manager.acquire_write_lock(key.clone(), "same-owner").await
+            }
+            .expect("old holder");
+            let shard = manager.get_shard(&key).clone();
+            let (entered, resume) = install_pause(TestPausePoint::ReleaseAfterUnregister, &key);
+            let old_drop = tokio::task::spawn_blocking(move || drop(old));
+            entered.recv_timeout(Duration::from_secs(5)).expect("old release paused");
+
+            let can_force = shard.active_guards.try_lock().is_some();
+            let mut force = {
+                let manager = manager.clone();
+                let key = key.clone();
+                tokio::task::spawn_blocking(move || manager.force_unlock(&key))
+            };
+            let mut new_guard = if can_force {
+                assert_eq!(force.await.expect("force join"), 1);
+                let mut guard = manager
+                    .acquire_write_lock(key.clone(), "same-owner")
+                    .await
+                    .expect("new writer");
+                if reader {
+                    assert!(guard.downgrade_committed_tail());
+                }
+                resume.send(()).expect("resume old release");
+                old_drop.await.expect("old release join");
+                guard
+            } else {
+                assert!(tokio::time::timeout(Duration::from_millis(100), &mut force).await.is_err());
+                resume.send(()).expect("resume old release");
+                old_drop.await.expect("old release join");
+                assert_eq!(force.await.expect("force join"), 0);
+                manager
+                    .acquire_write_lock(key.clone(), "same-owner")
+                    .await
+                    .expect("new writer")
+            };
+            if reader && !can_force {
+                assert!(new_guard.downgrade_committed_tail());
+            }
+            assert!(
+                tokio::time::timeout(Duration::from_millis(100), manager.acquire_write_lock(key, "competitor"))
+                    .await
+                    .is_err(),
+                "an old release must not unlock the new writer"
+            );
+            drop(new_guard);
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn force_unlock_does_not_capture_a_new_committed_tail() {
+        let manager = test_manager();
+        let key = ObjectKey::new("bucket", "force-tail-race");
+        let mut old = manager.acquire_write_lock(key.clone(), "old").await.expect("old writer");
+        assert!(old.downgrade_committed_tail());
+        let (entered, resume) = install_pause(TestPausePoint::ForceAfterTailRelease, &key);
+        let force = {
+            let manager = manager.clone();
+            let key = key.clone();
+            tokio::task::spawn_blocking(move || manager.force_unlock(&key))
+        };
+        entered.recv_timeout(Duration::from_secs(5)).expect("force unlock paused");
+        let mut next = {
+            let manager = manager.clone();
+            let key = key.clone();
+            tokio::spawn(async move { manager.acquire_write_lock(key, "new").await })
+        };
+        assert!(tokio::time::timeout(Duration::from_millis(100), &mut next).await.is_err());
+        resume.send(()).expect("resume force unlock");
+        assert_eq!(force.await.expect("force unlock task"), 1);
+        let mut new = next.await.expect("new writer task").expect("new writer");
+        assert!(new.downgrade_committed_tail());
+        drop(old);
+        assert!(
+            tokio::time::timeout(Duration::from_millis(100), manager.acquire_write_lock(key, "competitor"))
+                .await
+                .is_err(),
+            "the new committed tail must still exclude writers"
+        );
+    }
+
+    #[tokio::test]
+    async fn force_unlock_old_readers_cannot_release_a_new_tail() {
+        let manager = test_manager();
+        let key = ObjectKey::new("bucket", "force-old-readers");
+        let old1 = manager
+            .acquire_read_lock(key.clone(), "same-owner")
+            .await
+            .expect("first reader");
+        let old2 = manager
+            .acquire_read_lock(key.clone(), "same-owner")
+            .await
+            .expect("second reader");
+        assert_eq!(manager.force_unlock(&key), 1);
+        let mut tail = manager
+            .acquire_write_lock(key.clone(), "same-owner")
+            .await
+            .expect("new writer");
+        assert!(tail.downgrade_committed_tail());
+        drop((old1, old2));
+        assert!(
+            tokio::time::timeout(Duration::from_millis(100), manager.acquire_write_lock(key, "competitor"))
+                .await
+                .is_err(),
+            "old reader drops must not release the new tail"
+        );
+    }
+
+    #[tokio::test]
+    async fn forced_unregistered_reader_cannot_release_a_new_tail() {
+        let manager = test_manager();
+        let key = ObjectKey::new("bucket", "force-unregistered-reader");
+        let shard = manager.get_shard(&key).clone();
+        let owner: Arc<str> = Arc::from("same-owner");
+        let request = ObjectLockRequest::new_read(key.clone(), owner.clone());
+        let acquired = shard
+            .acquire_guarded(&request)
+            .await
+            .expect("old reader acquired before registration");
+        assert_eq!(manager.force_unlock(&key), 1);
+        let mut tail = manager
+            .acquire_write_lock(key.clone(), owner.clone())
+            .await
+            .expect("new writer");
+        assert!(tail.downgrade_committed_tail());
+        let mut old = FastLockGuard::new(key.clone(), LockMode::Shared, owner, shard.clone());
+        assert!(!shard.register_guard_with_info(old.guard_id(), old.key(), old.mode(), old.owner(), &acquired));
+        old.abandon_unregistered();
+        drop(old);
+        assert!(
+            tokio::time::timeout(Duration::from_millis(100), manager.acquire_write_lock(key, "competitor"))
+                .await
+                .is_err(),
+            "an old reader must not release a new committed tail"
+        );
+    }
+
+    #[tokio::test]
+    async fn force_unlock_on_another_key_keeps_unregistered_acquisition_valid() {
+        let manager = test_manager();
+        let key = ObjectKey::new("bucket", "unrelated-acquisition");
+        let shard = manager.get_shard(&key).clone();
+        let owner: Arc<str> = Arc::from("owner");
+        let request = ObjectLockRequest::new_read(key.clone(), owner.clone());
+        let acquired = shard.acquire_guarded(&request).await.expect("reader acquired");
+        let other = (0..1000)
+            .map(|index| ObjectKey::new("bucket", format!("other-{index}")))
+            .find(|other| Arc::ptr_eq(&shard, manager.get_shard(other)))
+            .expect("another key on the same shard");
+        let other_guard = manager
+            .acquire_write_lock(other.clone(), "other")
+            .await
+            .expect("other writer");
+        assert_eq!(manager.force_unlock(&other), 1);
+        drop(other_guard);
+        let guard = FastLockGuard::new(key.clone(), LockMode::Shared, owner, shard.clone());
+        assert!(shard.register_guard_with_info(guard.guard_id(), guard.key(), guard.mode(), guard.owner(), &acquired));
+        drop(guard);
+        assert!(manager.acquire_write_lock(key, "next").await.is_ok());
+    }
+
     fn start_acquisition(
         manager: &Arc<FastObjectLockManager>,
         request: ObjectLockRequest,

crates/lock/src/fast_lock/state.rs

@@ -23,6 +23,8 @@ use crate::fast_lock::types::{LockMode, LockPriority};
 /// Optimized atomic lock state encoding in u64
 /// Bits: [63:48] reserved | [47:32] writers_waiting | [31:16] readers_waiting | [15:8] readers_count | [7:1] flags | [0] writer_flag
 const WRITER_FLAG_MASK: u64 = 0x1;
+// A committed write tail is a reader-compatible owner that still excludes writers.
+const COMMITTED_TAIL_FLAG: u64 = 0x2;
 const READERS_SHIFT: u8 = 8;
 const READERS_MASK: u64 = 0xFF << READERS_SHIFT; // Support up to 255 concurrent readers
 const READERS_WAITING_SHIFT: u8 = 16;
@@ -31,7 +33,6 @@ const WRITERS_WAITING_SHIFT: u8 = 32;
 const WRITERS_WAITING_MASK: u64 = 0xFFFF << WRITERS_WAITING_SHIFT;
 
 // Fast path check masks
-const NO_WRITER_AND_NO_WAITING_WRITERS: u64 = WRITER_FLAG_MASK | WRITERS_WAITING_MASK;
 const COMPLETELY_UNLOCKED: u64 = 0;
 
 /// Fast atomic lock state for single version
@@ -66,8 +67,7 @@ impl AtomicLockState {
         let state = self.state.load(Ordering::Relaxed); // Use Relaxed for better performance
         match mode {
             LockMode::Shared => {
-                // No writer and no waiting writers
-                (state & NO_WRITER_AND_NO_WAITING_WRITERS) == 0
+                (state & WRITER_FLAG_MASK) == 0 && ((state & WRITERS_WAITING_MASK) == 0 || (state & COMMITTED_TAIL_FLAG) != 0)
             }
             LockMode::Exclusive => {
                 // Completely unlocked
@@ -83,8 +83,11 @@ impl AtomicLockState {
         loop {
             let current = self.state.load(Ordering::Acquire);
 
-            // Fast path check - cannot acquire if there's a writer or writers waiting
-            if (current & NO_WRITER_AND_NO_WAITING_WRITERS) != 0 {
+            // While a committed tail owns a shared slot, let reads through
+            // even if a subsequent writer is waiting for that tail to drain.
+            if (current & WRITER_FLAG_MASK) != 0
+                || ((current & WRITERS_WAITING_MASK) != 0 && (current & COMMITTED_TAIL_FLAG) == 0)
+            {
                 return false;
             }
 
@@ -137,6 +140,44 @@ impl AtomicLockState {
         }
     }
 
+    /// Atomically retain writer exclusion while allowing reads of a committed tail.
+    fn downgrade_committed_tail(&self) -> bool {
+        loop {
+            let current = self.state.load(Ordering::Acquire);
+            if (current & WRITER_FLAG_MASK) == 0 || (current & (READERS_MASK | COMMITTED_TAIL_FLAG)) != 0 {
+                return false;
+            }
+            let next = (current & !WRITER_FLAG_MASK) | COMMITTED_TAIL_FLAG | (1 << READERS_SHIFT);
+            if self
+                .state
+                .compare_exchange_weak(current, next, Ordering::AcqRel, Ordering::Relaxed)
+                .is_ok()
+            {
+                self.update_access_time();
+                return true;
+            }
+        }
+    }
+
+    /// Release the committed tail's own shared slot and restore ordinary fairness.
+    fn release_committed_tail(&self) -> bool {
+        loop {
+            let current = self.state.load(Ordering::Acquire);
+            if (current & COMMITTED_TAIL_FLAG) == 0 || self.readers_count(current) == 0 {
+                return false;
+            }
+            let next = (current & !COMMITTED_TAIL_FLAG) - (1 << READERS_SHIFT);
+            if self
+                .state
+                .compare_exchange_weak(current, next, Ordering::AcqRel, Ordering::Relaxed)
+                .is_ok()
+            {
+                self.update_access_time();
+                return true;
+            }
+        }
+    }
+
     /// Release shared lock
     pub fn release_shared(&self) -> bool {
         loop {
@@ -334,6 +375,8 @@ pub struct ObjectLockState {
     // First cache line: Most frequently accessed data
     /// Atomic state for fast operations
     pub atomic_state: AtomicLockState,
+    /// Changes whenever an administrator force-releases this state.
+    pub(crate) force_epoch: AtomicU64,
 
     // Second cache line: Notification mechanisms
     /// Notification for readers (traditional)
@@ -377,6 +420,7 @@ impl ObjectLockState {
     pub fn new() -> Self {
         Self {
             atomic_state: AtomicLockState::new(),
+            force_epoch: AtomicU64::new(0),
             read_notify: Notify::new(),
             write_notify: Notify::new(),
             optimized_notify: OptimizedNotify::new(),
@@ -425,6 +469,50 @@ impl ObjectLockState {
         true
     }
 
+    /// Convert the current exclusive owner into the committed-tail owner.
+    pub(crate) fn downgrade_committed_tail(&self, owner: &Arc<str>) -> bool {
+        let mut current = self.current_owner.write();
+        let Some(info) = current.as_ref().filter(|info| info.owner.as_ref() == owner.as_ref()).cloned() else {
+            return false;
+        };
+        let mut shared = self.shared_owners.write();
+        if !shared.is_empty() || !self.atomic_state.downgrade_committed_tail() {
+            return false;
+        }
+        shared.push(SharedOwnerEntry {
+            owner: info.owner,
+            count: 1,
+            acquired_at: info.acquired_at,
+            lock_timeout: info.lock_timeout,
+        });
+        *current = None;
+        drop(shared);
+        drop(current);
+        self.optimized_notify.notify_readers();
+        true
+    }
+
+    /// Release the committed-tail owner's shared slot without releasing readers.
+    pub(crate) fn release_committed_tail(&self, owner: &Arc<str>) -> bool {
+        let mut shared = self.shared_owners.write();
+        let Some(pos) = shared.iter().position(|entry| entry.owner.as_ref() == owner.as_ref()) else {
+            return false;
+        };
+        if !self.atomic_state.release_committed_tail() {
+            return false;
+        }
+        if shared[pos].count > 1 {
+            shared[pos].count -= 1;
+        } else {
+            shared.remove(pos);
+        }
+        if shared.is_empty() {
+            drop(shared);
+            self.optimized_notify.notify_writer();
+        }
+        true
+    }
+
     /// Release shared lock
     pub fn release_shared(&self, owner: &Arc<str>) -> bool {
         let mut shared = self.shared_owners.write();
@@ -555,6 +643,24 @@ mod tests {
         assert!(!state.release_exclusive());
     }
 
+    #[test]
+    fn committed_tail_allows_reads_but_preserves_writer_exclusion() {
+        let state = AtomicLockState::new();
+        assert!(state.try_acquire_exclusive());
+        assert!(state.inc_writers_waiting());
+        assert!(state.downgrade_committed_tail());
+        assert!(!state.try_acquire_exclusive(), "the tail must still exclude writers");
+        assert!(state.try_acquire_shared(), "a queued writer must not block reads of the committed tail");
+        assert!(state.release_committed_tail());
+        assert!(!state.try_acquire_shared(), "writer preference resumes when the tail ends");
+        assert!(!state.try_acquire_exclusive(), "an outstanding reader must still exclude writers");
+        assert!(state.release_shared());
+        assert!(state.try_acquire_exclusive());
+        assert!(state.release_exclusive());
+        state.dec_writers_waiting();
+        assert!(state.is_free());
+    }
+
     // Regression for the waiter-preserving exclusive CAS.
     //
     // The acquisition CAS used to demand a fully-zero state word, which

crates/lock/src/fast_lock/tests.rs

@@ -111,6 +111,67 @@ mod fast_lock_tests {
         assert!(guard.is_released(), "Guard should be marked as released");
     }
 
+    #[tokio::test]
+    async fn force_unlock_old_tail_cannot_release_new_tail() {
+        let manager = create_test_manager();
+        let key = ObjectKey::new("test-bucket", "force-tail-generation");
+        let owner: Arc<str> = Arc::from("same-owner");
+        let mut old = manager
+            .acquire_write_lock(key.clone(), owner.clone())
+            .await
+            .expect("old writer");
+        assert!(old.downgrade_committed_tail());
+        assert_eq!(manager.force_unlock(&key), 1);
+        let mut new = manager
+            .acquire_write_lock(key.clone(), owner.clone())
+            .await
+            .expect("new writer");
+        assert!(new.downgrade_committed_tail());
+        drop(old);
+        let attempt =
+            tokio::time::timeout(Duration::from_millis(100), manager.acquire_write_lock(key, Arc::from("competing-writer")))
+                .await;
+        assert!(attempt.is_err(), "dropping the old guard must not release the new tail");
+        drop(new);
+    }
+
+    #[tokio::test]
+    async fn committed_tail_allows_reader_while_writer_waits() {
+        let manager = Arc::new(create_test_manager());
+        let key = ObjectKey::new("test-bucket", "tail-handoff");
+        let mut tail = manager
+            .acquire_write_lock(key.clone(), Arc::from("first-writer"))
+            .await
+            .expect("initial writer");
+        let writer = tokio::spawn({
+            let manager = Arc::clone(&manager);
+            let key = key.clone();
+            async move { manager.acquire_write_lock(key, Arc::from("second-writer")).await }
+        });
+        tokio::time::timeout(Duration::from_secs(2), async {
+            while manager.current_lock_counts().1 == 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("second writer should be queued");
+        assert!(tail.downgrade_committed_tail());
+        let reader = tokio::time::timeout(Duration::from_secs(2), manager.acquire_read_lock(key, Arc::from("reader")))
+            .await
+            .expect("reader should not wait behind committed tail")
+            .expect("reader should acquire");
+        assert!(!writer.is_finished(), "tail and reader must exclude the next writer");
+        drop(tail);
+        assert!(!writer.is_finished(), "reader still excludes the next writer");
+        drop(reader);
+        let writer = tokio::time::timeout(Duration::from_secs(2), writer)
+            .await
+            .expect("writer should resume")
+            .expect("writer task should join")
+            .expect("writer should acquire");
+        drop(writer);
+    }
+
     #[tokio::test]
     async fn current_lock_counts_report_live_holders_and_waiters() {
         let manager = Arc::new(create_test_manager());

crates/lock/src/namespace/mod.rs

@@ -107,6 +107,14 @@ impl NamespaceLockGuard {
         }
     }
 
+    /// Permit reads while continuing to exclude competing writers.
+    pub async fn downgrade_committed_tail(&mut self) -> bool {
+        match self {
+            Self::Standard(guard) => guard.downgrade_committed_tail().await,
+            Self::Fast(guard) => guard.downgrade_committed_tail(),
+        }
+    }
+
     /// Manually release the lock early
     pub fn release(&mut self) -> bool {
         match self {

crates/lock/src/namespace/tests.rs

@@ -154,6 +154,43 @@ impl crate::client::LockClient for DelayedClient {
     }
 }
 
+#[derive(Debug)]
+struct PendingTailClient(LocalClient);
+
+#[async_trait::async_trait]
+impl crate::client::LockClient for PendingTailClient {
+    async fn acquire_lock(&self, request: &LockRequest) -> crate::Result<LockResponse> {
+        self.0.acquire_lock(request).await
+    }
+    async fn release(&self, lock_id: &LockId) -> crate::Result<bool> {
+        self.0.release(lock_id).await
+    }
+    async fn refresh(&self, lock_id: &LockId) -> crate::Result<bool> {
+        self.0.refresh(lock_id).await
+    }
+    async fn downgrade_committed_tail(&self, _: &LockId) -> crate::Result<bool> {
+        std::future::pending().await
+    }
+    async fn force_release(&self, lock_id: &LockId) -> crate::Result<bool> {
+        self.0.force_release(lock_id).await
+    }
+    async fn check_status(&self, lock_id: &LockId) -> crate::Result<Option<LockInfo>> {
+        self.0.check_status(lock_id).await
+    }
+    async fn get_stats(&self) -> crate::Result<LockStats> {
+        self.0.get_stats().await
+    }
+    async fn close(&self) -> crate::Result<()> {
+        self.0.close().await
+    }
+    async fn is_online(&self) -> bool {
+        self.0.is_online().await
+    }
+    async fn is_local(&self) -> bool {
+        self.0.is_local().await
+    }
+}
+
 #[derive(Debug)]
 struct FlakyAcquireClient {
     inner: LocalClient,
@@ -579,6 +616,168 @@ async fn test_namespace_lock_guard_lock_id() {
     }
 }
 
+#[tokio::test]
+async fn tail_conversion_rejects_non_intersecting_quorums() {
+    let clients: Vec<Arc<dyn crate::client::LockClient>> = (0..4)
+        .map(|_| Arc::new(LocalClient::with_manager(Arc::new(GlobalLockManager::new()))) as Arc<dyn crate::client::LockClient>)
+        .collect();
+    let lock = NamespaceLock::with_clients_and_quorum("non-intersecting".to_owned(), clients, 2);
+    let resource = ObjectKey::new("bucket", "non-intersecting-tail");
+    let mut tail = lock
+        .get_write_lock(resource, "writer", Duration::from_secs(2))
+        .await
+        .expect("write lease");
+    assert!(
+        !tail.downgrade_committed_tail().await,
+        "conversion requires intersecting read and write quorums"
+    );
+    assert!(tail.release());
+}
+
+#[tokio::test]
+async fn partial_tail_conversion_keeps_writers_excluded() {
+    let mut clients: Vec<Arc<dyn crate::client::LockClient>> = Vec::new();
+    for index in 0..4 {
+        let inner = Arc::new(LocalClient::with_manager(Arc::new(GlobalLockManager::new())));
+        if index < 2 {
+            clients.push(inner);
+        } else {
+            clients.push(Arc::new(DelayedClient {
+                inner,
+                delay: Duration::ZERO,
+            }));
+        }
+    }
+    let lock = NamespaceLock::with_clients("mixed-peers".to_owned(), clients);
+    let resource = ObjectKey::new("bucket", "partial-tail");
+    let mut tail = lock
+        .get_write_lock(resource.clone(), "first", Duration::from_secs(2))
+        .await
+        .expect("first writer");
+    assert!(!tail.downgrade_committed_tail().await, "mixed peers cannot all confirm conversion");
+    assert!(
+        lock.get_write_lock(resource.clone(), "second", Duration::from_millis(150))
+            .await
+            .is_err(),
+        "mixed leases still exclude writers"
+    );
+    assert!(tail.release());
+    let mut writer = lock
+        .get_write_lock(resource, "second", Duration::from_secs(3))
+        .await
+        .expect("writer after release");
+    assert!(writer.release());
+}
+
+#[tokio::test]
+async fn cancelled_tail_conversion_keeps_writers_excluded() {
+    let mut clients: Vec<Arc<dyn crate::client::LockClient>> = (0..3)
+        .map(|_| Arc::new(LocalClient::with_manager(Arc::new(GlobalLockManager::new()))) as Arc<dyn crate::client::LockClient>)
+        .collect();
+    clients.push(Arc::new(PendingTailClient(LocalClient::with_manager(Arc::new(GlobalLockManager::new())))));
+    let lock = NamespaceLock::with_clients_and_quorum("cancel-tail".to_owned(), clients, 4);
+    let resource = ObjectKey::new("bucket", "cancel-tail");
+    let mut tail = lock
+        .get_write_lock(resource.clone(), "first", Duration::from_secs(2))
+        .await
+        .expect("first writer");
+    assert!(
+        tokio::time::timeout(Duration::from_millis(100), tail.downgrade_committed_tail())
+            .await
+            .is_err()
+    );
+    let mut reader = lock
+        .get_read_lock(resource.clone(), "reader", Duration::from_secs(2))
+        .await
+        .expect("other peers converted before cancellation");
+    assert!(reader.release());
+    assert!(
+        lock.get_write_lock(resource.clone(), "second", Duration::from_millis(150))
+            .await
+            .is_err(),
+        "cancelled conversion must still exclude writers"
+    );
+    assert!(tail.release());
+    let mut writer = lock
+        .get_write_lock(resource, "second", Duration::from_secs(3))
+        .await
+        .expect("writer after release");
+    assert!(writer.release());
+}
+
+#[tokio::test]
+async fn unsupported_peer_keeps_exclusive_tail_lease() {
+    let clients: Vec<Arc<dyn crate::client::LockClient>> = (0..3)
+        .map(|_| {
+            let inner = Arc::new(LocalClient::with_manager(Arc::new(GlobalLockManager::new())));
+            Arc::new(DelayedClient {
+                inner,
+                delay: Duration::ZERO,
+            }) as Arc<dyn crate::client::LockClient>
+        })
+        .collect();
+    let lock = NamespaceLock::with_clients("old-peers".to_owned(), clients);
+    let resource = ObjectKey::new("bucket", "old-tail");
+    let mut tail = lock
+        .get_write_lock(resource.clone(), "first", Duration::from_secs(2))
+        .await
+        .expect("first writer");
+    assert!(!tail.downgrade_committed_tail().await, "old peers must not claim conversion");
+    assert!(
+        lock.get_read_lock(resource.clone(), "reader", Duration::from_millis(150))
+            .await
+            .is_err()
+    );
+    assert!(
+        lock.get_write_lock(resource.clone(), "writer", Duration::from_millis(150))
+            .await
+            .is_err()
+    );
+    assert!(tail.release());
+    let mut writer = lock
+        .get_write_lock(resource, "writer", Duration::from_secs(3))
+        .await
+        .expect("writer after tail");
+    assert!(writer.release());
+}
+
+#[tokio::test]
+async fn distributed_committed_tail_allows_reader_and_excludes_writer() {
+    let clients: Vec<Arc<dyn crate::client::LockClient>> = (0..3)
+        .map(|_| Arc::new(LocalClient::with_manager(Arc::new(GlobalLockManager::new()))) as Arc<dyn crate::client::LockClient>)
+        .collect();
+    let lock = NamespaceLock::with_clients("tail-test".to_owned(), clients);
+    let resource = ObjectKey::new("bucket", "committed-tail");
+    let mut tail = lock
+        .get_write_lock(resource.clone(), "first", Duration::from_secs(2))
+        .await
+        .expect("first writer");
+    assert!(tail.downgrade_committed_tail().await, "all held peer leases should convert");
+    let mut reader = lock
+        .get_read_lock(resource.clone(), "reader", Duration::from_secs(2))
+        .await
+        .expect("committed reader");
+    assert!(
+        lock.get_write_lock(resource.clone(), "second", Duration::from_millis(150))
+            .await
+            .is_err(),
+        "tail and reader must exclude writer"
+    );
+    assert!(tail.release());
+    assert!(
+        lock.get_write_lock(resource.clone(), "second", Duration::from_millis(150))
+            .await
+            .is_err(),
+        "reader still excludes writer"
+    );
+    assert!(reader.release());
+    let mut writer = lock
+        .get_write_lock(resource, "second", Duration::from_secs(3))
+        .await
+        .expect("writer after drain");
+    assert!(writer.release());
+}
+
 #[tokio::test]
 async fn test_namespace_lock_distributed_multi_node_simulation() {
     // Simulate a 3-node distributed environment where each node has its own lock backend
@@ -979,6 +1178,30 @@ async fn test_namespace_lock_distributed_retries_transient_acquire_timeout() {
     drop(guard);
 }
 
+#[tokio::test(start_paused = true)]
+async fn test_namespace_lock_retries_when_backoff_exceeds_remaining_budget() {
+    let managers = (0..3).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
+    let flaky_clients = managers
+        .iter()
+        .map(|manager| Arc::new(FlakyAcquireClient::new(manager.clone(), 1)))
+        .collect::<Vec<_>>();
+    let clients = flaky_clients
+        .iter()
+        .map(|client| client.clone() as Arc<dyn LockClient>)
+        .collect::<Vec<_>>();
+    let lock = NamespaceLock::with_clients("short-retry-budget".to_string(), clients);
+    let resource = create_test_object_key("bucket", "short-retry-budget");
+
+    // The first jittered backoff is at least 187 ms, but a second attempt can
+    // still succeed within this budget once the transient failure has cleared.
+    let guard = lock
+        .get_write_lock(resource, "owner", Duration::from_millis(180))
+        .await
+        .expect("a backoff longer than the remaining budget must not discard the remaining attempt");
+    assert!(flaky_clients.iter().any(|client| client.acquire_attempts() >= 2));
+    drop(guard);
+}
+
 #[tokio::test]
 async fn test_namespace_lock_distributed_waits_full_timeout_for_late_release() {
     let managers = (0..3).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
@@ -1136,10 +1359,16 @@ async fn test_namespace_lock_distributed_contention_quorum_miss_times_out() {
     let lock = NamespaceLock::with_clients("contention-timeout".to_string(), clients);
     let resource = create_test_object_key("bucket", "object-contention-timeout");
 
-    let err = lock
-        .get_write_lock(resource, "owner-a", Duration::from_millis(40))
+    let timeout = Duration::from_millis(40);
+    let started = std::time::Instant::now();
+    let err = tokio::time::timeout(Duration::from_secs(2), lock.get_write_lock(resource, "owner-a", timeout))
         .await
+        .expect("contended acquisition must remain bounded")
         .expect_err("ordinary contention should exhaust acquire timeout");
+    assert!(
+        started.elapsed() >= timeout,
+        "a contended acquisition must not return before its configured deadline"
+    );
 
     let err_str = err.to_string().to_lowercase();
     assert!(err_str.contains("timeout"), "expected timeout after contention retries, got: {err}");

rustfs/src/storage/rpc/node_service.rs

@@ -8915,28 +8915,47 @@ mod tests {
             ))
         }
 
-        async fn internal_snapshot(root: &std::path::Path) -> std::collections::BTreeMap<std::path::PathBuf, Option<Vec<u8>>> {
+        async fn internal_snapshot_once(
+            root: &std::path::Path,
+        ) -> std::io::Result<std::collections::BTreeMap<std::path::PathBuf, Option<Vec<u8>>>> {
             let mut snapshot = std::collections::BTreeMap::new();
             let mut directories = (0..4)
                 .map(|index| std::path::PathBuf::from(format!("disk{index}/{RUSTFS_META_BUCKET}")))
                 .collect::<Vec<_>>();
             while let Some(relative) = directories.pop() {
-                let mut entries = tokio::fs::read_dir(root.join(&relative))
-                    .await
-                    .expect("read internal snapshot directory");
+                let mut entries = tokio::fs::read_dir(root.join(&relative)).await?;
                 snapshot.insert(relative.clone(), None);
-                while let Some(entry) = entries.next_entry().await.expect("read internal snapshot entry") {
+                while let Some(entry) = entries.next_entry().await? {
                     let path = relative.join(entry.file_name());
-                    let file_type = entry.file_type().await.expect("read internal snapshot entry type");
+                    let file_type = entry.file_type().await?;
                     if file_type.is_dir() {
                         directories.push(path);
                     } else {
                         assert!(file_type.is_file(), "fixture snapshot must contain only directories and regular files");
-                        snapshot.insert(path, Some(tokio::fs::read(entry.path()).await.expect("read snapshot file bytes")));
+                        snapshot.insert(path, Some(tokio::fs::read(entry.path()).await?));
                     }
                 }
             }
-            snapshot
+            Ok(snapshot)
+        }
+
+        async fn internal_snapshot(root: &std::path::Path) -> std::collections::BTreeMap<std::path::PathBuf, Option<Vec<u8>>> {
+            // Startup cleanup can remove temporary paths after the namespace
+            // commit drains. Capture the complete tree only once it is stable.
+            super::timeout(Duration::from_secs(10), async {
+                let mut previous = None;
+                loop {
+                    match internal_snapshot_once(root).await {
+                        Ok(snapshot) if previous.as_ref() == Some(&snapshot) => return snapshot,
+                        Ok(snapshot) => previous = Some(snapshot),
+                        Err(error) if error.kind() == std::io::ErrorKind::NotFound => previous = None,
+                        Err(error) => panic!("read internal snapshot: {error}"),
+                    }
+                    tokio::time::sleep(Duration::from_millis(20)).await;
+                }
+            })
+            .await
+            .expect("internal snapshot must stabilize within ten seconds")
         }
 
         fn file_info(object: &str, version: Uuid, body: Bytes) -> FileInfo {

rustfs/src/storage/rpc/node_service/lock.rs

@@ -49,6 +49,26 @@ async fn refresh_lock(
     lock_client: &std::sync::Arc<dyn rustfs_lock::client::LockClient>,
     args: &LockRequest,
 ) -> GenerallyLockResponse {
+    let marker = rustfs_lock::client::COMMITTED_TAIL_CONVERSION_MARKER;
+    if args.metadata.tags.get(marker).is_some_and(|value| value == marker) {
+        return match lock_client.downgrade_committed_tail(&args.lock_id).await {
+            Ok(true) => GenerallyLockResponse {
+                success: true,
+                error_info: None,
+                lock_info: Some(marker.to_owned()),
+            },
+            Ok(false) => GenerallyLockResponse {
+                success: false,
+                error_info: None,
+                lock_info: None,
+            },
+            Err(error) => GenerallyLockResponse {
+                success: false,
+                error_info: Some(error.to_string()),
+                lock_info: None,
+            },
+        };
+    }
     match lock_client.refresh(&args.lock_id).await {
         Ok(true) => GenerallyLockResponse {
             success: true,
@@ -347,6 +367,30 @@ mod tests {
         assert!(result.lock_info.is_none());
     }
 
+    #[tokio::test]
+    async fn refresh_can_convert_committed_tail_and_acknowledges_protocol_version() {
+        use super::refresh_lock;
+        use rustfs_lock::client::{COMMITTED_TAIL_CONVERSION_MARKER, LockClient};
+        use rustfs_lock::{LocalClient, LockRequest, LockType, ObjectKey};
+        use std::{sync::Arc, time::Duration};
+
+        let client: Arc<dyn LockClient> = Arc::new(LocalClient::new());
+        let resource = ObjectKey::new("bucket", "tail-conversion-rpc");
+        let mut request = LockRequest::new(resource.clone(), LockType::Exclusive, "owner");
+        assert!(client.acquire_lock(&request).await.expect("acquire write lease").success);
+        request
+            .metadata
+            .tags
+            .insert(COMMITTED_TAIL_CONVERSION_MARKER.into(), COMMITTED_TAIL_CONVERSION_MARKER.into());
+        let response = refresh_lock(&client, &request).await;
+        assert!(response.success);
+        assert_eq!(response.lock_info.as_deref(), Some(COMMITTED_TAIL_CONVERSION_MARKER));
+        let read = LockRequest::new(resource, LockType::Shared, "reader").with_acquire_timeout(Duration::from_secs(1));
+        assert!(client.acquire_lock(&read).await.expect("read committed tail").success);
+        assert!(client.release(&read.lock_id).await.expect("release reader"));
+        assert!(client.release(&request.lock_id).await.expect("release tail"));
+    }
+
     // B1 -- refresh_lock delegates to LocalClient::refresh for a held lock, returns
     //        success=true, and actually extends expires_at (inverts the "parse then drop" stub).
     #[tokio::test]