ci: prepare fast lock guard regression preflight

.github/workflows/windows-filesystem.yml

@@ -1,224 +1,166 @@
-# Copyright 2024 RustFS Team
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-name: Windows Filesystem Tests
+name: FastLock targeted preflight
 
 on:
-  push:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
-  pull_request:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
   workflow_dispatch:
+    inputs:
+      expected_head:
+        description: Exact workflow commit SHA
+        required: true
+        type: string
 
 permissions:
   contents: read
 
 concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-  cancel-in-progress: true
+  group: fast-lock-targeted-${{ github.run_id }}
+  cancel-in-progress: false
 
 env:
-  CARGO_TERM_COLOR: always
-  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-  RUST_BACKTRACE: 1
+  CARGO_TERM_COLOR: never
+  CARGO_INCREMENTAL: '0'
+  CARGO_BUILD_JOBS: '2'
+  CARGO_PROFILE_DEV_DEBUG: '0'
+  CARGO_PROFILE_TEST_DEBUG: '0'
+  RUST_BACKTRACE: '1'
+  NO_PROXY: localhost,127.0.0.1,::1
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: 'true'
 
 jobs:
-  rename-safety:
-    name: Rename Safety
-    runs-on: windows-latest
-    timeout-minutes: 60
+  targeted:
+    runs-on: ubuntu-latest
+    timeout-minutes: 130
     steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
-        with:
-          persist-credentials: false
-
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-          install-test-tools: 'false'
-
-      - name: Check production Windows dependencies
-        shell: pwsh
-        run: cargo check -p rustfs-ecstore --lib
-
-      - name: Test guarded rename publication
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib rename_all_ -- --nocapture
-
-      - name: Test multipart parent durability barriers
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib multipart_owned_record_parent_barriers_ -- --nocapture
-
-      - name: Test Windows path conversion
-        shell: pwsh
-        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
-
-      - name: Test Windows volume and disk-space probes
-        shell: pwsh
+      - name: Prepare evidence
         run: |
-          # Map a drive letter to the temp drive's administrative share so the
-          # remote-volume probe also covers mapped network drives.
-          $drive = (Split-Path -Qualifier ([System.IO.Path]::GetTempPath())).TrimEnd(':')
-          net use R: "\\localhost\$drive`$" /persistent:no
-          if ($LASTEXITCODE -eq 0) {
-            $env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE = 'R:'
-          } else {
-            Write-Output "::warning::could not map R: to \\localhost\$drive`$; mapped-drive test will skip"
-          }
-          cargo test -p rustfs-utils --features os --lib os:: -- --nocapture
-          $status = $LASTEXITCODE
-          if ($env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE) { net use R: /delete /y | Out-Null }
-          exit $status
+          set -euo pipefail
+          mkdir -p "$RUNNER_TEMP/fast-lock-targeted"
+          printf '%s\n' 'not-started' > "$RUNNER_TEMP/fast-lock-targeted/build.exit"
 
-      - name: Test Windows handle guards
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib windows_ -- --nocapture
-
-      - name: Test endpoint path parsing
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib layout::endpoint:: -- --nocapture"
-
-      - name: Test startup temporary-directory cleanup
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib cleanup_tmp_on_startup_ -- --nocapture
-
-      - name: Test object key Windows admission
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib bucket::utils:: -- --nocapture"
-
-      - name: Test Windows file-name rules
-        shell: pwsh
-        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
-
-      - name: Test fresh format publication
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib fresh_format_load_initializes_all_disks -- --nocapture
-
-  # The native Windows main thread only gets the 1 MiB PE stack reserve, while
-  # Linux and macOS hand out 8 MiB. Startup runs the whole root future on the
-  # thread that calls `block_on`, so a debug server once overflowed before it
-  # listened on Windows only. Start a real server to keep that covered.
-  startup-smoke:
-    name: Startup Smoke
-    runs-on: windows-latest
-    timeout-minutes: 90
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+      - name: Checkout exact workflow commit
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a
         with:
+          fetch-depth: 3
           persist-credentials: false
+          clean: false
 
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-          install-test-tools: 'false'
-
-      - name: Build debug server binary
-        shell: pwsh
-        run: cargo build --locked -p rustfs --bin rustfs
-
-      - name: Print version
-        shell: pwsh
-        run: |
-          & .\target\debug\rustfs.exe --version
-          if ($LASTEXITCODE -ne 0) { throw "rustfs --version exited with $LASTEXITCODE" }
-
-      - name: Start server and wait for readiness
-        shell: pwsh
+      - name: Verify source and workflow scope
         env:
-          RUSTFS_ACCESS_KEY: rustfsadmin
-          RUSTFS_SECRET_KEY: rustfsadmin
+          EXPECTED_HEAD: ${{ inputs.expected_head }}
         run: |
-          $volume = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke'
-          $stdout = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.out.log'
-          $stderr = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.err.log'
-          New-Item -ItemType Directory -Force -Path $volume | Out-Null
-          $server = Start-Process -FilePath .\target\debug\rustfs.exe `
-            -ArgumentList @('--address', '127.0.0.1:9000', "`"$volume`"") `
-            -RedirectStandardOutput $stdout -RedirectStandardError $stderr `
-            -NoNewWindow -PassThru
-          # Cache the process handle now; otherwise ExitCode reads back empty.
-          $null = $server.Handle
-          $ready = $false
-          try {
-            for ($i = 0; $i -lt 180; $i++) {
-              if ($server.HasExited) { break }
-              try {
-                $response = Invoke-WebRequest -Uri 'http://127.0.0.1:9000/health/ready' -UseBasicParsing -NoProxy -TimeoutSec 2
-                if ($response.StatusCode -eq 200) { $ready = $true; break }
-              } catch {}
-              Start-Sleep -Seconds 1
-            }
-          } finally {
-            if (-not $server.HasExited) { Stop-Process -Id $server.Id -Force }
-            $server.WaitForExit()
-            Write-Host '--- server stdout (tail) ---'
-            Get-Content $stdout -Tail 100 -ErrorAction SilentlyContinue
-            Write-Host '--- server stderr (tail) ---'
-            Get-Content $stderr -Tail 100 -ErrorAction SilentlyContinue
-          }
-          if (-not $ready) {
-            throw ("rustfs did not report ready on /health/ready (exit code: {0})" -f $server.ExitCode)
-          }
+          set -euo pipefail
+          evidence="$RUNNER_TEMP/fast-lock-targeted"
+          workflow_sha=$(git rev-parse HEAD)
+          source_sha=$(git rev-parse 'HEAD^')
+          base_sha=$(git rev-parse 'HEAD^^')
+          source_tree=$(git rev-parse "$source_sha^{tree}")
+          source_diff=$(git diff --binary --full-index "$base_sha" "$source_sha" | sha256sum | cut -d' ' -f1)
+          source_files=$(git diff --name-only "$base_sha" "$source_sha")
+          workflow_files=$(git diff --name-only "$source_sha" "$workflow_sha")
+          printf 'expected_head=%s\nworkflow=%s\nsource=%s\nbase=%s\ntree=%s\nsource_diff=%s\nsource_files=%s\nworkflow_files=%s\n' \
+            "$EXPECTED_HEAD" "$workflow_sha" "$source_sha" "$base_sha" "$source_tree" "$source_diff" \
+            "$source_files" "$workflow_files" | tee "$evidence/source-proof.txt"
+          [[ "$EXPECTED_HEAD" =~ ^[0-9a-f]{40}$ ]]
+          [[ "$GITHUB_EVENT_NAME" == 'workflow_dispatch' ]]
+          [[ "$EXPECTED_HEAD" == "$GITHUB_SHA" ]]
+          [[ "$workflow_sha" == "$GITHUB_SHA" ]]
+          [[ "$source_sha" == 'c802fe210dd01a66c9b917794c2bcf0184401cad' ]]
+          [[ "$base_sha" == '56c80a566bac7fa1c0e74430b76ccb8db787aef9' ]]
+          [[ "$source_tree" == '01b31f5c19e1f26653849fc77be74a9c5dc7e606' ]]
+          [[ "$source_diff" == '307ba0219cf70e1dec215c4e466aca3287f59edf5f1eded39f31a1a961d2b570' ]]
+          [[ "$source_files" == $'crates/lock/src/client/local.rs\ncrates/lock/src/fast_lock/guard.rs\ncrates/lock/src/fast_lock/manager.rs\ncrates/lock/src/fast_lock/shard.rs\ncrates/lock/src/fast_lock/state.rs' ]]
+          [[ "$workflow_files" == '.github/workflows/windows-filesystem.yml' ]]
+          [[ -z "$(git status --porcelain)" ]]
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8
+        with:
+          toolchain: stable
+
+      - name: Check disk and toolchain
+        run: |
+          set -euo pipefail
+          evidence="$RUNNER_TEMP/fast-lock-targeted"
+          df -h "$GITHUB_WORKSPACE" | tee "$evidence/disk.txt"
+          available_kib=$(df -Pk "$GITHUB_WORKSPACE" | awk 'NR == 2 {print $4}')
+          [[ "$available_kib" =~ ^[0-9]+$ ]]
+          [[ "$available_kib" -ge 8388608 ]]
+          rustc --version | tee "$evidence/rustc.txt"
+          cargo --version | tee "$evidence/cargo.txt"
+
+      - name: Compile lock library tests
+        timeout-minutes: 75
+        run: |
+          set -euo pipefail
+          evidence="$RUNNER_TEMP/fast-lock-targeted"
+          set +e
+          timeout --signal=TERM --kill-after=30s 70m \
+            cargo test --locked -p rustfs-lock --lib --no-run > "$evidence/build.log" 2>&1
+          result=$?
+          set -e
+          printf '%s\n' "$result" > "$evidence/build.exit"
+          tail -n 100 "$evidence/build.log"
+          exit "$result"
+
+      - name: Run exact local lock tests
+        timeout-minutes: 50
+        run: |
+          set -euo pipefail
+          evidence="$RUNNER_TEMP/fast-lock-targeted"
+          tests=(
+            'fast_lock::manager::tests::force_unlocked_exclusive_guard_drop_preserves_same_owner_successor'
+            'fast_lock::manager::tests::force_unlocked_exclusive_guard_release_preserves_same_owner_successor'
+            'fast_lock::manager::tests::force_unlock_releases_all_shared_guards_of_same_owner'
+            'fast_lock::manager::tests::force_unlocked_shared_guards_drop_preserves_same_owner_successor'
+            'fast_lock::manager::tests::force_unlocked_shared_guards_release_preserves_same_owner_successor'
+            'client::local::tests::force_before_lease_publication_does_not_report_acquisition'
+            'client::local::tests::force_revoked_lease_cannot_be_refreshed_or_reported_held'
+            'fast_lock::tests::fast_lock_tests::test_basic_write_lock_acquire_release'
+            'fast_lock::tests::fast_lock_tests::test_basic_read_lock_acquire_release'
+            'client::local::tests::lease_snapshot_tracks_refresh_without_resetting_acquisition_time'
+            'client::local::tests::refresh_after_expiry_releases_guard_without_reviving_it'
+            'fast_lock::shard::tests::force_waits_for_tracked_holder_and_registration_commit'
+            'fast_lock::shard::tests::stale_numeric_release_does_not_release_reacquired_holder'
+            'fast_lock::shard::tests::in_flight_weak_numeric_release_does_not_release_successor'
+            'fast_lock::shard::tests::in_flight_weak_release_preserves_replacement_tracked_registration'
+            'fast_lock::shard::tests::in_flight_weak_release_preserves_reregistered_raw_holder'
+            'fast_lock::shard::tests::tracked_revocation_preserves_a_replacement_weak_registration'
+            'fast_lock::shard::tests::raw_and_weak_release_prefer_raw_shared_holder'
+            'fast_lock::shard::tests::releasing_one_token_preserves_a_different_token_with_the_same_numeric_id'
+            'fast_lock::shard::tests::raw_batch_failure_rolls_back_a_previously_acquired_lock'
+            'fast_lock::shard::tests::failed_raw_batch_does_not_release_a_reacquired_holder'
+            'fast_lock::shard::tests::cancelling_raw_batch_releases_its_holders_and_waiter'
+            'fast_lock::shard::tests::pooled_state_does_not_revive_a_forced_guard'
+            'fast_lock::shard::tests::release_wakes_can_reenter_the_shard'
+          )
+          printf '%s\n' "${tests[@]}" > "$evidence/selection.txt"
+          [[ "${#tests[@]}" -eq 24 ]]
+          failures=0
+          for test_name in "${tests[@]}"; do
+            file_name=${test_name//::/_}
+            log="$evidence/$file_name.log"
+            set +e
+            timeout --signal=TERM --kill-after=10s 2m \
+              cargo test --locked -p rustfs-lock --lib -- --exact "$test_name" --test-threads=1 > "$log" 2>&1
+            result=$?
+            set -e
+            printf '%s\n' "$result" > "$evidence/$file_name.exit"
+            printf 'test=%s exit=%s\n' "$test_name" "$result"
+            tail -n 50 "$log"
+            if [[ "$result" -ne 0 ]] \
+              || [[ "$(grep -Fxc 'running 1 test' "$log" || true)" != '1' ]] \
+              || [[ "$(grep -Fxc "test $test_name ... ok" "$log" || true)" != '1' ]] \
+              || [[ "$(grep -Ec '^test result: ok\. 1 passed; 0 failed; 0 ignored; 0 measured; [0-9]+ filtered out;' "$log" || true)" != '1' ]]; then
+              printf 'Unexpected test result: %s\n' "$test_name"
+              failures=$((failures + 1))
+            fi
+          done
+          printf '%s\n' "$failures" > "$evidence/failures.txt"
+          [[ "$failures" -eq 0 ]]
+
+      - name: Upload evidence
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a
+        with:
+          name: fast-lock-targeted-evidence
+          path: ${{ runner.temp }}/fast-lock-targeted/
+          if-no-files-found: warn

crates/lock/src/client/local.rs

@@ -275,6 +275,11 @@ impl LockClient for LocalClient {
                     {
                         let shard = self.get_shard(&lock_id);
                         let mut guards = shard.write().await;
+                        // The physical guard can be revoked while waiting for
+                        // the lease storage lock. Check at publication time.
+                        if entry.guard.is_released() {
+                            return Ok(LockResponse::failure("Lock was released before acquisition completed", Duration::ZERO));
+                        }
                         guards.insert(lock_id.clone(), entry);
                     }
 
@@ -361,21 +366,29 @@ impl LockClient for LocalClient {
 
     async fn refresh(&self, lock_id: &LockId) -> Result<bool> {
         let shard = self.get_shard(lock_id);
-        let expired_entry = {
+        let (removed_entry, was_expired) = {
             let mut guards = shard.write().await;
             let Some(entry) = guards.get_mut(lock_id) else {
                 return Ok(false);
             };
-            if entry.is_expired() {
-                guards.remove(lock_id)
+            if entry.guard.is_released() {
+                (guards.remove(lock_id), false)
+            } else if entry.is_expired() {
+                (guards.remove(lock_id), true)
             } else {
                 entry.refresh();
-                None
+                if entry.guard.is_released() {
+                    (guards.remove(lock_id), false)
+                } else {
+                    (None, false)
+                }
             }
         };
 
-        if let Some(entry) = expired_entry {
-            Self::release_reclaimed_guards([entry], Some(&lock_id.resource));
+        if let Some(entry) = removed_entry {
+            if was_expired {
+                Self::release_reclaimed_guards([entry], Some(&lock_id.resource));
+            }
             Ok(false)
         } else {
             Ok(true)
@@ -390,7 +403,9 @@ impl LockClient for LocalClient {
         let shard = self.get_shard(lock_id);
         let guards = shard.read().await;
         if let Some(entry) = guards.get(lock_id) {
-            // We have an active guard for this lock
+            if entry.guard.is_released() {
+                return Ok(None);
+            }
             let lock_type = match entry.guard.mode() {
                 crate::LockMode::Shared => LockType::Shared,
                 crate::LockMode::Exclusive => LockType::Exclusive,
@@ -400,7 +415,7 @@ impl LockClient for LocalClient {
             } else {
                 LockStatus::Acquired
             };
-            Ok(Some(LockInfo {
+            let info = LockInfo {
                 id: lock_id.clone(),
                 resource: lock_id.resource.clone(),
                 lock_type,
@@ -412,7 +427,8 @@ impl LockClient for LocalClient {
                 metadata: LockMetadata::default(),
                 priority: LockPriority::Normal,
                 wait_start_time: None,
-            }))
+            };
+            Ok((!entry.guard.is_released()).then_some(info))
         } else {
             Ok(None)
         }
@@ -423,17 +439,22 @@ impl LockClient for LocalClient {
         for shard in self.guard_storage.iter() {
             let guards = shard.read().await;
             leases.reserve(guards.len());
-            leases.extend(guards.iter().map(|(lock_id, entry)| crate::LockLeaseInfo {
-                resource: lock_id.resource.clone(),
-                lock_type: match entry.guard.mode() {
-                    crate::LockMode::Shared => LockType::Shared,
-                    crate::LockMode::Exclusive => LockType::Exclusive,
-                },
-                owner: entry.guard.owner().to_string(),
-                acquired_at: entry.acquired_at,
-                remaining_ttl: entry.deadline.saturating_duration_since(Instant::now()),
-                guard_id: (!entry.guard.is_disabled()).then(|| entry.guard.guard_id()),
-            }));
+            leases.extend(
+                guards
+                    .iter()
+                    .filter(|(_, entry)| !entry.guard.is_released())
+                    .map(|(lock_id, entry)| crate::LockLeaseInfo {
+                        resource: lock_id.resource.clone(),
+                        lock_type: match entry.guard.mode() {
+                            crate::LockMode::Shared => LockType::Shared,
+                            crate::LockMode::Exclusive => LockType::Exclusive,
+                        },
+                        owner: entry.guard.owner().to_string(),
+                        acquired_at: entry.acquired_at,
+                        remaining_ttl: entry.deadline.saturating_duration_since(Instant::now()),
+                        guard_id: (!entry.guard.is_disabled()).then(|| entry.guard.guard_id()),
+                    }),
+            );
         }
         leases
     }
@@ -476,6 +497,77 @@ mod tests {
         panic!("lock guard was not reaped before test deadline");
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn force_before_lease_publication_does_not_report_acquisition() {
+        let fast = Arc::new(crate::FastObjectLockManager::new());
+        let manager = Arc::new(GlobalLockManager::Enabled(fast.clone()));
+        let client = Arc::new(LocalClient::with_manager_and_reaper_interval(manager, Duration::from_secs(60)));
+        client.reaper_started.store(true, Ordering::Release);
+        let resource = crate::ObjectKey::new("bucket", "force-before-lease-publication");
+        let request = request(resource.clone(), "owner", Duration::from_secs(30)).with_acquire_timeout(Duration::from_secs(30));
+
+        // A read guard allows the pre-acquisition scan but blocks publication.
+        let storage_read = client.get_shard(&request.lock_id).read().await;
+        let pending = {
+            let client = client.clone();
+            let request = request.clone();
+            tokio::spawn(async move { client.acquire_lock(&request).await })
+        };
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while fast.get_lock_info(&resource).is_none() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("physical lock should be acquired before lease publication");
+        assert!(!pending.is_finished());
+        assert_eq!(fast.force_unlock(&resource), 1);
+        let successor = fast
+            .acquire_write_lock(resource.clone(), "owner")
+            .await
+            .expect("successor holder");
+        drop(storage_read);
+
+        let response = tokio::time::timeout(Duration::from_secs(5), pending)
+            .await
+            .expect("acquire response")
+            .expect("acquire task")
+            .expect("local acquire");
+        assert!(!response.success, "a revoked guard must not be published as acquired");
+        assert!(client.check_status(&request.lock_id).await.expect("lease status").is_none());
+        let other = crate::ObjectLockRequest::new_write(resource.clone(), "other");
+        assert!(!fast.get_shard(&resource).try_fast_path_only(&other), "successor must remain held");
+        drop(successor);
+        fast.shutdown().await;
+    }
+
+    #[tokio::test]
+    async fn force_revoked_lease_cannot_be_refreshed_or_reported_held() {
+        let fast = Arc::new(crate::FastObjectLockManager::new());
+        let manager = Arc::new(GlobalLockManager::Enabled(fast.clone()));
+        let client = LocalClient::with_manager_and_reaper_interval(manager, Duration::from_secs(60));
+        client.reaper_started.store(true, Ordering::Release);
+        for lock_type in [LockType::Exclusive, LockType::Shared] {
+            let resource = crate::ObjectKey::new("bucket", format!("force-revoked-{lock_type:?}"));
+            let request = LockRequest::new(resource.clone(), lock_type, "owner").with_ttl(Duration::from_secs(30));
+            assert!(client.acquire_lock(&request).await.expect("initial acquire").success);
+            assert_eq!(fast.force_unlock(&resource), 1);
+            let successor = match lock_type {
+                LockType::Exclusive => fast.acquire_write_lock(resource.clone(), "owner").await,
+                LockType::Shared => fast.acquire_read_lock(resource.clone(), "owner").await,
+            }
+            .expect("successor holder");
+            assert!(client.check_status(&request.lock_id).await.expect("lease status").is_none());
+            assert!(client.list_lock_leases().await.is_empty());
+            assert!(!client.refresh(&request.lock_id).await.expect("lease refresh"));
+            assert!(!client.release(&request.lock_id).await.expect("old lease was removed"));
+            let other = crate::ObjectLockRequest::new_write(resource.clone(), "other");
+            assert!(!fast.get_shard(&resource).try_fast_path_only(&other), "successor must remain held");
+            drop(successor);
+        }
+        fast.shutdown().await;
+    }
+
     #[tokio::test]
     async fn acquire_does_not_wait_on_unrelated_guard_shards() {
         let manager = Arc::new(GlobalLockManager::new());

crates/lock/src/fast_lock/guard.rs

@@ -20,14 +20,45 @@ use rustfs_io_metrics::{
     record_read_lock_held_acquire, record_read_lock_held_release, record_write_lock_held_acquire, record_write_lock_held_release,
 };
 use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 
-/// Global counter for guard IDs to prevent double-release
+/// Global counter for guard IDs; zero is reserved for an unavailable ID.
 static GUARD_ID_COUNTER: AtomicU64 = AtomicU64::new(1);
 const EVENT_NAMESPACE_LOCK_RELEASE: &str = "namespace_lock_release";
 const LOG_COMPONENT_LOCK: &str = "lock";
 const LOG_SUBSYSTEM_NAMESPACE: &str = "namespace";
 
+#[derive(Debug)]
+pub(crate) struct GuardToken {
+    pub(crate) id: u64,
+    pub(crate) acquired: AtomicBool,
+    pub(crate) tracked: bool,
+}
+
+impl GuardToken {
+    pub(crate) fn new() -> Arc<Self> {
+        Arc::new(Self {
+            id: next_guard_id(&GUARD_ID_COUNTER),
+            acquired: AtomicBool::new(false),
+            tracked: true,
+        })
+    }
+
+    pub(crate) fn raw_batch() -> Arc<Self> {
+        Arc::new(Self {
+            id: 0,
+            acquired: AtomicBool::new(false),
+            tracked: false,
+        })
+    }
+}
+
+fn next_guard_id(counter: &AtomicU64) -> u64 {
+    counter
+        .try_update(Ordering::Relaxed, Ordering::Relaxed, |id| id.checked_add(1))
+        .unwrap_or(0)
+}
+
 /// RAII guard for fast object locks
 ///
 /// Automatically releases the lock when dropped, ensuring no lock leakage
@@ -41,12 +72,11 @@ pub struct FastLockGuard {
     disabled: bool, // True when locks are disabled globally
     /// Unique ID for this guard instance to prevent double-release
     guard_id: u64,
+    token: Option<Arc<GuardToken>>,
 }
 
 impl FastLockGuard {
-    pub(crate) fn new(key: ObjectKey, mode: LockMode, owner: Arc<str>, shard: Arc<LockShard>) -> Self {
-        let guard_id = GUARD_ID_COUNTER.fetch_add(1, Ordering::Relaxed);
-        record_lock_held_acquire(mode);
+    pub(crate) fn new(key: ObjectKey, mode: LockMode, owner: Arc<str>, shard: Arc<LockShard>, token: Arc<GuardToken>) -> Self {
         Self {
             key,
             mode,
@@ -54,13 +84,14 @@ impl FastLockGuard {
             shard: Some(shard),
             released: false,
             disabled: false,
-            guard_id,
+            guard_id: token.id,
+            token: Some(token),
         }
     }
 
     /// Create a disabled guard (when locks are globally disabled)
     pub(crate) fn new_disabled(key: ObjectKey, mode: LockMode, owner: Arc<str>) -> Self {
-        let guard_id = GUARD_ID_COUNTER.fetch_add(1, Ordering::Relaxed);
+        let guard_id = next_guard_id(&GUARD_ID_COUNTER);
         Self {
             key,
             mode,
@@ -69,6 +100,7 @@ impl FastLockGuard {
             released: false,
             disabled: true,
             guard_id,
+            token: None,
         }
     }
 
@@ -105,13 +137,13 @@ impl FastLockGuard {
             return true;
         }
 
-        if let Some(shard) = &self.shard {
-            let success = shard.release_lock_with_guard(&self.key, &self.owner, self.mode, self.guard_id);
+        if let (Some(shard), Some(token)) = (&self.shard, &self.token) {
+            if !token.acquired.load(Ordering::Acquire) {
+                return false;
+            }
+            let success = shard.release_tracked(&self.key, &self.owner, self.mode, token);
             if success {
                 self.released = true;
-                record_lock_held_release(self.mode);
-                // Unregister the guard after successful release
-                shard.unregister_guard(self.guard_id);
                 self.record_physical_release();
             }
             success
@@ -138,9 +170,15 @@ impl FastLockGuard {
         );
     }
 
-    /// Check if the lock has been released
+    /// Check if this guard has been released or locally revoked.
+    ///
+    /// A false result is only a snapshot; it cannot fence a later force unlock.
     pub fn is_released(&self) -> bool {
         self.released
+            || self
+                .token
+                .as_ref()
+                .is_some_and(|token| !token.acquired.load(Ordering::Acquire))
     }
 
     /// Check if this guard represents a disabled lock
@@ -148,14 +186,14 @@ impl FastLockGuard {
         self.disabled
     }
 
-    /// Get the unique guard ID
+    /// Get the process-local guard ID, or zero if the ID space is exhausted.
     pub fn guard_id(&self) -> u64 {
         self.guard_id
     }
 
     /// Get lock information for monitoring
     pub fn lock_info(&self) -> Option<crate::fast_lock::types::ObjectLockInfo> {
-        if self.released || self.disabled {
+        if self.is_released() || self.disabled {
             None
         } else if let Some(shard) = &self.shard {
             shard.get_lock_info(&self.key)
@@ -167,40 +205,31 @@ impl FastLockGuard {
 
 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);
-                if !success {
-                    // For high-concurrency scenarios, this is likely due to:
-                    // 1. Lock was already released by another thread
-                    // 2. Lock was cleaned up by background cleanup
-                    // 3. Legitimate double-drop scenario
-                    tracing::debug!(
-                        "Guard release failed (likely already released): key={}, owner={}, mode={:?}, guard_id={}",
-                        self.key,
-                        self.owner,
-                        self.mode,
-                        self.guard_id
-                    );
-                }
-                if success {
-                    record_lock_held_release(self.mode);
-                }
-                // Always unregister the guard to prevent leaks, regardless of release success
-                shard.unregister_guard(self.guard_id);
-                if success {
-                    self.record_physical_release();
-                }
-            } else {
-                // If guard was already released or disabled, just unregister it
-                shard.unregister_guard(self.guard_id);
+        if let (Some(shard), Some(token)) = (&self.shard, &self.token)
+            && !self.released
+            && !self.disabled
+            && token.acquired.load(Ordering::Acquire)
+        {
+            let success = shard.release_tracked(&self.key, &self.owner, self.mode, token);
+            if !success {
+                // A force unlock or owner-only release may have revoked this holder.
+                tracing::debug!(
+                    "Guard release failed (likely already released): key={}, owner={}, mode={:?}, guard_id={}",
+                    self.key,
+                    self.owner,
+                    self.mode,
+                    self.guard_id
+                );
+            }
+            if success {
+                self.record_physical_release();
             }
         }
     }
 }
 
 #[inline(always)]
-fn record_lock_held_acquire(mode: LockMode) {
+pub(crate) fn record_lock_held_acquire(mode: LockMode) {
     match mode {
         LockMode::Shared => record_read_lock_held_acquire(),
         LockMode::Exclusive => record_write_lock_held_acquire(),
@@ -208,7 +237,7 @@ fn record_lock_held_acquire(mode: LockMode) {
 }
 
 #[inline(always)]
-fn record_lock_held_release(mode: LockMode) {
+pub(crate) fn record_lock_held_release(mode: LockMode) {
     match mode {
         LockMode::Shared => record_read_lock_held_release(),
         LockMode::Exclusive => record_write_lock_held_release(),

crates/lock/src/fast_lock/manager.rs

@@ -19,7 +19,7 @@ use tracing::warn;
 
 use crate::LockError;
 use crate::fast_lock::{
-    guard::FastLockGuard,
+    guard::{FastLockGuard, GuardToken},
     manager_trait::LockManager,
     metrics::{AggregatedMetrics, GlobalMetrics},
     shard::LockShard,
@@ -96,15 +96,10 @@ 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());
-                // Register guard to prevent premature cleanup
-                shard.register_guard_with_info(guard.guard_id(), guard.key(), guard.mode(), guard.owner());
-                Ok(guard)
-            }
-            Err(err) => Err(err),
-        }
+        let token = GuardToken::new();
+        let guard = FastLockGuard::new(request.key.clone(), request.mode, request.owner.clone(), shard.clone(), token.clone());
+        shard.acquire_tracked(&request, &token).await?;
+        Ok(guard)
     }
 
     /// Acquire shared (read) lock
@@ -208,11 +203,13 @@ impl FastObjectLockManager {
                 let key = request.key.clone();
                 let owner = request.owner.clone();
                 let mode = request.mode;
+                let token = GuardToken::new();
+                let guard = FastLockGuard::new(key.clone(), mode, owner, shard.clone(), token.clone());
 
-                let acquired = if shard.try_fast_path_only(request) {
+                let acquired = if shard.try_fast_path_tracked_only(request, &token) {
                     true
                 } else {
-                    match shard.acquire_lock(request).await {
+                    match shard.acquire_tracked(request, &token).await {
                         Ok(()) => true,
                         Err(err) => {
                             all_failed.push((key.clone(), err));
@@ -222,8 +219,6 @@ impl FastObjectLockManager {
                 };
 
                 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);
                 }
@@ -249,10 +244,11 @@ impl FastObjectLockManager {
             let shard = self.shards[*shard_id].clone();
 
             for request in requests {
-                match shard.acquire_lock(request).await {
+                let token = GuardToken::new();
+                let guard =
+                    FastLockGuard::new(request.key.clone(), request.mode, request.owner.clone(), shard.clone(), token.clone());
+                match shard.acquire_tracked(request, &token).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());
                         acquired_guards.push(guard);
                     }
                     Err(err) => {
@@ -310,7 +306,8 @@ impl FastObjectLockManager {
         infos
     }
 
-    /// Enumerate held locks with holder counts and stable holder identities.
+    /// Enumerate held locks with holder counts and process-local snapshot IDs.
+    /// Missing IDs include raw holders; equal IDs across snapshots do not prove continuous ownership.
     pub fn list_locks_with_holder_generations(&self) -> Vec<(crate::fast_lock::types::ObjectLockInfo, u32, Option<Vec<u64>>)> {
         let mut infos = Vec::new();
         for shard in &self.shards {
@@ -661,4 +658,208 @@ mod tests {
 
         manager.shutdown().await;
     }
+
+    fn assert_registered_holder_blocks_writer(manager: &FastObjectLockManager, key: &ObjectKey, guard_id: u64, stage: &str) {
+        assert!(
+            manager.get_shard(key).is_guard_active(guard_id),
+            "{stage}: successor guard must remain registered"
+        );
+        assert_eq!(manager.current_lock_counts().0, 1, "{stage}: successor must remain the sole holder");
+        let contender = ObjectLockRequest::new_write(key.clone(), "contender");
+        assert!(
+            !manager.get_shard(key).try_fast_path_only(&contender),
+            "{stage}: successor must block another writer"
+        );
+    }
+
+    fn assert_free_for_new_writer(manager: &FastObjectLockManager, key: &ObjectKey, stage: &str) {
+        assert_eq!(manager.current_lock_counts().0, 0, "{stage}: no holder should remain");
+        let writer = ObjectLockRequest::new_write(key.clone(), "contender");
+        let shard = manager.get_shard(key);
+        assert!(shard.try_fast_path_only(&writer), "{stage}: a new writer must acquire");
+        assert_eq!(manager.current_lock_counts().0, 1, "{stage}: new writer must hold the lock");
+        assert!(
+            shard.release_lock(key, &writer.owner, LockMode::Exclusive),
+            "{stage}: release the new writer"
+        );
+        assert_eq!(manager.current_lock_counts().0, 0, "{stage}: new writer must release the lock");
+    }
+
+    fn assert_no_registered_guards(manager: &FastObjectLockManager) {
+        let active: usize = manager.shards.iter().map(|shard| shard.active_guard_count()).sum();
+        assert_eq!(active, 0, "all guards in the isolated manager must be unregistered");
+    }
+
+    fn assert_successor_release_or_drop(
+        manager: &FastObjectLockManager,
+        key: &ObjectKey,
+        mut successor: FastLockGuard,
+        release_first: bool,
+    ) {
+        let successor_id = successor.guard_id();
+        if release_first {
+            assert!(successor.release(), "successor must release its own physical lock");
+            assert!(successor.is_released(), "successor must report its explicit release");
+            assert!(
+                !manager.get_shard(key).is_guard_active(successor_id),
+                "released successor must no longer be registered"
+            );
+            assert_free_for_new_writer(manager, key, "after successor release and before drop");
+        }
+        drop(successor);
+        assert!(
+            !manager.get_shard(key).is_guard_active(successor_id),
+            "dropped successor must no longer be registered"
+        );
+        assert_free_for_new_writer(manager, key, "after successor drop");
+    }
+
+    async fn stale_exclusive_guard_does_not_release_successor(release_first: bool, release_successor: bool) {
+        let manager = FastObjectLockManager::new();
+        let key = ObjectKey::new("bucket", "force-unlock-exclusive-successor");
+        let mut stale = manager
+            .acquire_write_lock(key.clone(), "same-owner")
+            .await
+            .expect("acquire original exclusive guard");
+        let stale_id = stale.guard_id();
+        assert_eq!(manager.force_unlock(&key), 1);
+        assert!(stale.is_released(), "force unlock must revoke the original guard");
+        assert!(stale.lock_info().is_none(), "a revoked guard must not report its successor");
+        let successor = manager
+            .acquire_write_lock(key.clone(), "same-owner")
+            .await
+            .expect("acquire exclusive successor after force unlock");
+        assert!(!successor.is_released());
+        assert_ne!(stale.guard_id(), successor.guard_id());
+        assert_registered_holder_blocks_writer(&manager, &key, successor.guard_id(), "before stale guard action");
+
+        if release_first {
+            let _ = stale.release();
+            assert_registered_holder_blocks_writer(&manager, &key, successor.guard_id(), "after stale guard release");
+        }
+        drop(stale);
+        assert!(
+            !manager.get_shard(&key).is_guard_active(stale_id),
+            "dropped stale guard must be unregistered"
+        );
+
+        assert_registered_holder_blocks_writer(&manager, &key, successor.guard_id(), "after stale guard drop");
+        assert_successor_release_or_drop(&manager, &key, successor, release_successor);
+        assert_no_registered_guards(&manager);
+        manager.shutdown().await;
+    }
+
+    #[tokio::test]
+    async fn force_unlocked_exclusive_guard_drop_preserves_same_owner_successor() {
+        stale_exclusive_guard_does_not_release_successor(false, true).await;
+        stale_exclusive_guard_does_not_release_successor(false, false).await;
+    }
+
+    #[tokio::test]
+    async fn force_unlocked_exclusive_guard_release_preserves_same_owner_successor() {
+        stale_exclusive_guard_does_not_release_successor(true, true).await;
+        stale_exclusive_guard_does_not_release_successor(true, false).await;
+    }
+
+    #[tokio::test]
+    async fn force_unlock_releases_all_shared_guards_of_same_owner() {
+        let manager = FastObjectLockManager::new();
+        let key = ObjectKey::new("bucket", "force-unlock-shared-owners");
+        let first = manager
+            .acquire_read_lock(key.clone(), "same-owner")
+            .await
+            .expect("acquire first shared guard");
+        let second = manager
+            .acquire_read_lock(key.clone(), "same-owner")
+            .await
+            .expect("acquire second shared guard");
+        let first_id = first.guard_id();
+        let second_id = second.guard_id();
+        assert_eq!(manager.current_lock_counts().0, 2);
+
+        assert_eq!(manager.force_unlock(&key), 1, "one distinct owner was force unlocked");
+        assert!(first.is_released(), "first shared guard must be revoked");
+        assert!(second.is_released(), "second shared guard must be revoked");
+        assert_eq!(
+            manager.current_lock_counts().0,
+            0,
+            "force unlock must release every shared guard of the owner"
+        );
+        assert_free_for_new_writer(&manager, &key, "after shared force unlock");
+        drop(first);
+        assert!(
+            !manager.get_shard(&key).is_guard_active(first_id),
+            "first dropped stale guard must be unregistered"
+        );
+        drop(second);
+        assert!(
+            !manager.get_shard(&key).is_guard_active(second_id),
+            "second dropped stale guard must be unregistered"
+        );
+        assert_free_for_new_writer(&manager, &key, "after stale shared guards drop");
+        assert_no_registered_guards(&manager);
+        manager.shutdown().await;
+    }
+
+    async fn stale_shared_guards_do_not_release_successor(release_first: bool, release_successor: bool) {
+        let manager = FastObjectLockManager::new();
+        let key = ObjectKey::new("bucket", "force-unlock-shared-successor");
+        let mut first = manager
+            .acquire_read_lock(key.clone(), "same-owner")
+            .await
+            .expect("acquire first shared guard");
+        let mut second = manager
+            .acquire_read_lock(key.clone(), "same-owner")
+            .await
+            .expect("acquire second shared guard");
+        let first_id = first.guard_id();
+        let second_id = second.guard_id();
+        assert_eq!(manager.force_unlock(&key), 1);
+        // Clear both old holders even on the baseline, so the following
+        // assertion isolates stale guard release from force-unlock completeness.
+        let _ = manager.force_unlock(&key);
+        assert_eq!(manager.current_lock_counts().0, 0);
+        let successor = manager
+            .acquire_read_lock(key.clone(), "same-owner")
+            .await
+            .expect("acquire shared successor after force unlock");
+        assert_ne!(first.guard_id(), successor.guard_id());
+        assert_ne!(second.guard_id(), successor.guard_id());
+        assert_registered_holder_blocks_writer(&manager, &key, successor.guard_id(), "before stale shared guard action");
+
+        if release_first {
+            let _ = first.release();
+            assert_registered_holder_blocks_writer(&manager, &key, successor.guard_id(), "after first stale shared release");
+            let _ = second.release();
+            assert_registered_holder_blocks_writer(&manager, &key, successor.guard_id(), "after second stale shared release");
+        }
+        drop(first);
+        assert!(
+            !manager.get_shard(&key).is_guard_active(first_id),
+            "first dropped stale guard must be unregistered"
+        );
+        assert_registered_holder_blocks_writer(&manager, &key, successor.guard_id(), "after first stale shared drop");
+        drop(second);
+        assert!(
+            !manager.get_shard(&key).is_guard_active(second_id),
+            "second dropped stale guard must be unregistered"
+        );
+        assert_registered_holder_blocks_writer(&manager, &key, successor.guard_id(), "after second stale shared drop");
+
+        assert_successor_release_or_drop(&manager, &key, successor, release_successor);
+        assert_no_registered_guards(&manager);
+        manager.shutdown().await;
+    }
+
+    #[tokio::test]
+    async fn force_unlocked_shared_guards_drop_preserves_same_owner_successor() {
+        stale_shared_guards_do_not_release_successor(false, true).await;
+        stale_shared_guards_do_not_release_successor(false, false).await;
+    }
+
+    #[tokio::test]
+    async fn force_unlocked_shared_guards_release_preserves_same_owner_successor() {
+        stale_shared_guards_do_not_release_successor(true, true).await;
+        stale_shared_guards_do_not_release_successor(true, false).await;
+    }
 }

crates/lock/src/fast_lock/state.rs

@@ -367,6 +367,72 @@ pub struct SharedOwnerEntry {
     pub lock_timeout: Duration,
 }
 
+#[derive(Clone, Copy, Debug)]
+pub(crate) enum ReleaseWake {
+    None,
+    Writer,
+    PreferWriter,
+}
+
+#[derive(Debug)]
+enum ReleaseFailure {
+    SharedInconsistent(usize),
+    SharedOwnerMissing(Vec<Arc<str>>),
+    ExclusiveInconsistent(u64),
+    ExclusiveOwnerMismatch(Option<Arc<str>>),
+}
+
+#[derive(Debug)]
+pub(crate) struct ReleaseOutcome {
+    pub(crate) success: bool,
+    wake: ReleaseWake,
+    failure: Option<ReleaseFailure>,
+}
+
+impl ReleaseOutcome {
+    pub(crate) fn finish(self, state: &ObjectLockState, owner: &Arc<str>) -> bool {
+        match self.failure {
+            Some(ReleaseFailure::SharedInconsistent(remaining)) => tracing::warn!(
+                "Atomic state inconsistency during shared lock release: owner={}, remaining_entries={}",
+                owner,
+                remaining
+            ),
+            Some(ReleaseFailure::SharedOwnerMissing(entries)) => tracing::debug!(
+                "Shared lock release failed - owner not found: owner={}, current_entries={:?}",
+                owner,
+                entries
+            ),
+            Some(ReleaseFailure::ExclusiveInconsistent(bits)) => tracing::warn!(
+                "Atomic state inconsistency during exclusive lock release: owner={}, atomic_state={:b}",
+                owner,
+                bits
+            ),
+            Some(ReleaseFailure::ExclusiveOwnerMismatch(actual)) => tracing::debug!(
+                "Exclusive lock release failed - owner mismatch: expected_owner={}, actual_owner={:?}",
+                owner,
+                actual
+            ),
+            None => {}
+        }
+        match self.wake {
+            ReleaseWake::None => {}
+            ReleaseWake::Writer => state.optimized_notify.notify_writer(),
+            ReleaseWake::PreferWriter => {
+                if state
+                    .atomic_state
+                    .writers_waiting(state.atomic_state.state.load(Ordering::Acquire))
+                    > 0
+                {
+                    state.optimized_notify.notify_writer();
+                } else {
+                    state.optimized_notify.notify_readers();
+                }
+            }
+        }
+        self.success
+    }
+}
+
 impl Default for ObjectLockState {
     fn default() -> Self {
         Self::new()
@@ -427,6 +493,10 @@ impl ObjectLockState {
 
     /// Release shared lock
     pub fn release_shared(&self, owner: &Arc<str>) -> bool {
+        self.release_shared_deferred(owner).finish(self, owner)
+    }
+
+    pub(crate) fn release_shared_deferred(&self, owner: &Arc<str>) -> ReleaseOutcome {
         let mut shared = self.shared_owners.write();
         if let Some(pos) = shared.iter().position(|entry| entry.owner.as_ref() == owner.as_ref()) {
             let original_entry = shared[pos].clone();
@@ -437,17 +507,17 @@ impl ObjectLockState {
                 Some(shared.remove(pos))
             };
             if self.atomic_state.release_shared() {
-                if shared.is_empty() {
-                    drop(shared);
-                    self.optimized_notify.notify_writer();
+                ReleaseOutcome {
+                    success: true,
+                    wake: if shared.is_empty() {
+                        ReleaseWake::Writer
+                    } else {
+                        ReleaseWake::None
+                    },
+                    failure: None,
                 }
-                true
             } else {
-                tracing::warn!(
-                    "Atomic state inconsistency during shared lock release: owner={}, remaining_entries={}",
-                    owner,
-                    shared.len()
-                );
+                let remaining = shared.len();
                 // Re-add owner entry to maintain consistency when release failed
                 match removed_entry {
                     Some(entry) => {
@@ -461,53 +531,53 @@ impl ObjectLockState {
                         }
                     }
                 }
-                false
+                ReleaseOutcome {
+                    success: false,
+                    wake: ReleaseWake::None,
+                    failure: Some(ReleaseFailure::SharedInconsistent(remaining)),
+                }
             }
         } else {
-            tracing::debug!(
-                "Shared lock release failed - owner not found: owner={}, current_entries={:?}",
-                owner,
-                shared.iter().map(|s| s.owner.as_ref()).collect::<Vec<_>>()
-            );
-            false
+            ReleaseOutcome {
+                success: false,
+                wake: ReleaseWake::None,
+                failure: Some(ReleaseFailure::SharedOwnerMissing(
+                    shared.iter().map(|entry| entry.owner.clone()).collect(),
+                )),
+            }
         }
     }
 
     /// Release exclusive lock
     pub fn release_exclusive(&self, owner: &Arc<str>) -> bool {
+        self.release_exclusive_deferred(owner).finish(self, owner)
+    }
+
+    pub(crate) fn release_exclusive_deferred(&self, owner: &Arc<str>) -> ReleaseOutcome {
         let mut current = self.current_owner.write();
         if current.as_ref().is_some_and(|info| info.owner.as_ref() == owner.as_ref()) {
             if self.atomic_state.release_exclusive() {
                 *current = None;
-                drop(current);
-                // Notify waiters using optimized system - prefer writers over readers
-                if self
-                    .atomic_state
-                    .writers_waiting(self.atomic_state.state.load(Ordering::Acquire))
-                    > 0
-                {
-                    self.optimized_notify.notify_writer();
-                } else {
-                    self.optimized_notify.notify_readers();
+                ReleaseOutcome {
+                    success: true,
+                    wake: ReleaseWake::PreferWriter,
+                    failure: None,
                 }
-                true
             } else {
                 // Atomic state inconsistency - current owner matches but atomic release failed
-                tracing::warn!(
-                    "Atomic state inconsistency during exclusive lock release: owner={}, atomic_state={:b}",
-                    owner,
-                    self.atomic_state.state.load(Ordering::Acquire)
-                );
-                false
+                ReleaseOutcome {
+                    success: false,
+                    wake: ReleaseWake::None,
+                    failure: Some(ReleaseFailure::ExclusiveInconsistent(self.atomic_state.state.load(Ordering::Acquire))),
+                }
             }
         } else {
             // Owner mismatch
-            tracing::debug!(
-                "Exclusive lock release failed - owner mismatch: expected_owner={}, actual_owner={:?}",
-                owner,
-                current.as_ref().map(|s| s.owner.as_ref())
-            );
-            false
+            ReleaseOutcome {
+                success: false,
+                wake: ReleaseWake::None,
+                failure: Some(ReleaseFailure::ExclusiveOwnerMismatch(current.as_ref().map(|entry| entry.owner.clone()))),
+            }
         }
     }