fix(asyncio): prevent deadlock when Lock.release() is cancelled

redis/asyncio/lock.py

@@ -263,16 +263,27 @@ async def owned(self) -> bool:
             stored_token = encoder.encode(stored_token)
         return self.local.token is not None and stored_token == self.local.token
 
-    def release(self) -> Awaitable[None]:
-        """Releases the already acquired lock"""
+    async def release(self) -> None:
+        """Releases the already acquired lock.
+
+        The token is only cleared after the Redis release operation completes
+        successfully. This ensures that if the release is cancelled mid-operation,
+        the lock state remains consistent and can be retried.
+        """
         expected_token = self.local.token
         if expected_token is None:
             raise LockError(
                 "Cannot release a lock that's not owned or is already unlocked.",
                 lock_name=self.name,
             )
+        try:
+            await self.do_release(expected_token)
+        except LockNotOwnedError:
+            # Lock doesn't exist in Redis, safe to clear token
+            self.local.token = None
+            raise
+        # Only clear token after successful release
         self.local.token = None
-        return self.do_release(expected_token)
 
     async def do_release(self, expected_token: bytes) -> None:
         if not bool(

tests/test_asyncio/test_lock.py

@@ -259,6 +259,43 @@ async def test_reacquiring_lock_no_longer_owned_raises_error(self, r):
         with pytest.raises(LockNotOwnedError):
             await lock.reacquire()
 
+    async def test_release_cancellation_preserves_lock_state(self, r):
+        """
+        Test that cancelling release() doesn't leave lock in inconsistent state.
+
+        Regression test for GitHub issue #3847. Before the fix, if release()
+        was cancelled during execution, the token would be cleared but the
+        Redis key would remain, causing a permanent deadlock.
+        """
+        lock = self.get_lock(r, "foo")
+        await lock.acquire(blocking=False)
+
+        # Verify lock is owned
+        original_token = lock.local.token
+        assert original_token is not None
+        assert await lock.owned()
+
+        # Create release task and cancel it immediately
+        release_task = asyncio.create_task(lock.release())
+        release_task.cancel()
+
+        try:
+            await release_task
+        except asyncio.CancelledError:
+            # Expected: the release task was deliberately cancelled to test lock state.
+            pass
+
+        # Check the lock state after cancellation
+        if lock.local.token is not None:
+            # Release was cancelled before completion - token preserved
+            # This is the fix: we can now retry the release
+            assert lock.local.token == original_token
+            await lock.release()
+            assert await lock.locked() is False
+        else:
+            # Release completed before cancel took effect
+            assert await lock.locked() is False
+
 
 @pytest.mark.onlynoncluster
 class TestLockClassSelection: