Repository: redis/redis-py
Pull request #3551 (2025-03-09 11:09:04)

Got rid of `time.time()` everywhere
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [x] Do tests and lints pass with this change?
- [x] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [x] Is the new or changed code fully tested?
- [ ] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [ ] Is there an example added to the examples folder (if applicable)?
- [ ] Was the change added to CHANGES file?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

Made time calculations more stable, in library itself and in tests as well.

excitoon: @andymccurdy please run CI/CD

Patch:
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 42544410..412d5a24 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -478,6 +478,7 @@ class Redis(
         blocking_timeout: Optional[float] = None,
         lock_class: Optional[Type[Lock]] = None,
         thread_local: bool = True,
+        raise_on_release_error: bool = True,
     ) -> Lock:
         """
         Return a new Lock object using key ``name`` that mimics
@@ -524,6 +525,11 @@ class Redis(
                      thread-1 would see the token value as "xyz" and would be
                      able to successfully release the thread-2's lock.
 
+        ``raise_on_release_error`` indicates whether to raise an exception when
+        the lock is no longer owned when exiting the context manager. By default,
+        this is True, meaning an exception will be raised. If False, the warning
+        will be logged and the exception will be suppressed.
+
         In some use cases it's necessary to disable thread local storage. For
         example, if you have code where one thread acquires a lock and passes
         that lock instance to a worker thread to release later. If thread
@@ -541,6 +547,7 @@ class Redis(
             blocking=blocking,
             blocking_timeout=blocking_timeout,
             thread_local=thread_local,
+            raise_on_release_error=raise_on_release_error,
         )
 
     def pubsub(self, **kwargs) -> "PubSub":
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index f343e26b..51dce895 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -839,6 +839,7 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
         blocking_timeout: Optional[float] = None,
         lock_class: Optional[Type[Lock]] = None,
         thread_local: bool = True,
+        raise_on_release_error: bool = True,
     ) -> Lock:
         """
         Return a new Lock object using key ``name`` that mimics
@@ -885,6 +886,11 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
                      thread-1 would see the token value as "xyz" and would be
                      able to successfully release the thread-2's lock.
 
+        ``raise_on_release_error`` indicates whether to raise an exception when
+        the lock is no longer owned when exiting the context manager. By default,
+        this is True, meaning an exception will be raised. If False, the warning
+        will be logged and the exception will be suppressed.
+
         In some use cases it's necessary to disable thread local storage. For
         example, if you have code where one thread acquires a lock and passes
         that lock instance to a worker thread to release later. If thread
@@ -902,6 +908,7 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
             blocking=blocking,
             blocking_timeout=blocking_timeout,
             thread_local=thread_local,
+            raise_on_release_error=raise_on_release_error,
         )
 
 
@@ -1597,18 +1604,24 @@ class ClusterPipeline(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterComm
                         result.args = (msg,) + result.args[1:]
                         raise result
 
-            default_node = nodes.get(client.get_default_node().name)
-            if default_node is not None:
-                # This pipeline execution used the default node, check if we need
-                # to replace it.
-                # Note: when the error is raised we'll reset the default node in the
-                # caller function.
-                for cmd in default_node[1]:
-                    # Check if it has a command that failed with a relevant
-                    # exception
-                    if type(cmd.result) in self.__class__.ERRORS_ALLOW_RETRY:
-                        client.replace_default_node()
-                        break
+            default_cluster_node = client.get_default_node()
+
+            # Check whether the default node was used. In some cases,
+            # 'client.get_default_node()' may return None. The check below
+            # prevents a potential AttributeError.
+            if default_cluster_node is not None:
+                default_node = nodes.get(default_cluster_node.name)
+                if default_node is not None:
+                    # This pipeline execution used the default node, check if we need
+                    # to replace it.
+                    # Note: when the error is raised we'll reset the default node in the
+                    # caller function.
+                    for cmd in default_node[1]:
+                        # Check if it has a command that failed with a relevant
+                        # exception
+                        if type(cmd.result) in self.__class__.ERRORS_ALLOW_RETRY:
+                            client.replace_default_node()
+                            break
 
         return [cmd.result for cmd in stack]
 
diff --git a/redis/asyncio/lock.py b/redis/asyncio/lock.py
index f70a8d09..16d7fb69 100644
--- a/redis/asyncio/lock.py
+++ b/redis/asyncio/lock.py
@@ -1,4 +1,5 @@
 import asyncio
+import logging
 import threading
 import uuid
 from types import SimpleNamespace
@@ -10,6 +11,8 @@ from redis.typing import Number
 if TYPE_CHECKING:
     from redis.asyncio import Redis, RedisCluster
 
+logger = logging.getLogger(__name__)
+
 
 class Lock:
     """
@@ -85,6 +88,7 @@ class Lock:
         blocking: bool = True,
         blocking_timeout: Optional[Number] = None,
         thread_local: bool = True,
+        raise_on_release_error: bool = True,
     ):
         """
         Create a new Lock instance named ``name`` using the Redis client
@@ -128,6 +132,11 @@ class Lock:
                      thread-1 would see the token value as "xyz" and would be
                      able to successfully release the thread-2's lock.
 
+        ``raise_on_release_error`` indicates whether to raise an exception when
+        the lock is no longer owned when exiting the context manager. By default,
+        this is True, meaning an exception will be raised. If False, the warning
+        will be logged and the exception will be suppressed.
+
         In some use cases it's necessary to disable thread local storage. For
         example, if you have code where one thread acquires a lock and passes
         that lock instance to a worker thread to release later. If thread
@@ -144,6 +153,7 @@ class Lock:
         self.blocking_timeout = blocking_timeout
         self.thread_local = bool(thread_local)
         self.local = threading.local() if self.thread_local else SimpleNamespace()
+        self.raise_on_release_error = raise_on_release_error
         self.local.token = None
         self.register_scripts()
 
@@ -163,7 +173,14 @@ class Lock:
         raise LockError("Unable to acquire lock within the time specified")
 
     async def __aexit__(self, exc_type, exc_value, traceback):
-        await self.release()
+        try:
+            await self.release()
+        except LockError:
+            if self.raise_on_release_error:
+                raise
+            logger.warning(
+                "Lock was unlocked or no longer owned when exiting context manager."
+            )
 
     async def acquire(
         self,
diff --git a/redis/client.py b/redis/client.py
index 2bacbe14..ea29a864 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -473,6 +473,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
         blocking_timeout: Optional[float] = None,
         lock_class: Union[None, Any] = None,
         thread_local: bool = True,
+        raise_on_release_error: bool = True,
     ):
         """
         Return a new Lock object using key ``name`` that mimics
@@ -519,6 +520,11 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
                      thread-1 would see the token value as "xyz" and would be
                      able to successfully release the thread-2's lock.
 
+        ``raise_on_release_error`` indicates whether to raise an exception when
+        the lock is no longer owned when exiting the context manager. By default,
+        this is True, meaning an exception will be raised. If False, the warning
+        will be logged and the exception will be suppressed.
+
         In some use cases it's necessary to disable thread local storage. For
         example, if you have code where one thread acquires a lock and passes
         that lock instance to a worker thread to release later. If thread
@@ -536,6 +542,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
             blocking=blocking,
             blocking_timeout=blocking_timeout,
             thread_local=thread_local,
+            raise_on_release_error=raise_on_release_error,
         )
 
     def pubsub(self, **kwargs):
@@ -950,7 +957,7 @@ class PubSub:
                 "did you forget to call subscribe() or psubscribe()?"
             )
 
-        if conn.health_check_interval and time.time() > conn.next_health_check:
+        if conn.health_check_interval and time.monotonic() > conn.next_health_check:
             conn.send_command("PING", self.HEALTH_CHECK_MESSAGE, check_health=False)
             self.health_check_response_counter += 1
 
@@ -1100,12 +1107,12 @@ class PubSub:
         """
         if not self.subscribed:
             # Wait for subscription
-            start_time = time.time()
+            start_time = time.monotonic()
             if self.subscribed_event.wait(timeout) is True:
                 # The connection was subscribed during the timeout time frame.
                 # The timeout should be adjusted based on the time spent
                 # waiting for the subscription
-                time_spent = time.time() - start_time
+                time_spent = time.monotonic() - start_time
                 timeout = max(0.0, timeout - time_spent)
             else:
                 # The connection isn't subscribed to any channels or patterns,
diff --git a/redis/cluster.py b/redis/cluster.py
index ef4500f8..c9523e2a 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -822,6 +822,7 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
         blocking_timeout=None,
         lock_class=None,
         thread_local=True,
+        raise_on_release_error: bool = True,
     ):
         """
         Return a new Lock object using key ``name`` that mimics
@@ -868,6 +869,11 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
                      thread-1 would see the token value as "xyz" and would be
                      able to successfully release the thread-2's lock.
 
+        ``raise_on_release_error`` indicates whether to raise an exception when
+        the lock is no longer owned when exiting the context manager. By default,
+        this is True, meaning an exception will be raised. If False, the warning
+        will be logged and the exception will be suppressed.
+
         In some use cases it's necessary to disable thread local storage. For
         example, if you have code where one thread acquires a lock and passes
         that lock instance to a worker thread to release later. If thread
@@ -885,6 +891,7 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
             blocking=blocking,
             blocking_timeout=blocking_timeout,
             thread_local=thread_local,
+            raise_on_release_error=raise_on_release_error,
         )
 
     def set_response_callback(self, command, callback):
diff --git a/redis/commands/search/commands.py b/redis/commands/search/commands.py
index 1db57c23..96c6d9c2 100644
--- a/redis/commands/search/commands.py
+++ b/redis/commands/search/commands.py
@@ -500,7 +500,7 @@ class SearchCommands:
         For more information see `FT.SEARCH <https://redis.io/commands/ft.search>`_.
         """  # noqa
         args, query = self._mk_query_args(query, query_params=query_params)
-        st = time.time()
+        st = time.monotonic()
 
         options = {}
         if get_protocol_version(self.client) not in ["3", 3]:
@@ -512,7 +512,7 @@ class SearchCommands:
             return res
 
         return self._parse_results(
-            SEARCH_CMD, res, query=query, duration=(time.time() - st) * 1000.0
+            SEARCH_CMD, res, query=query, duration=(time.monotonic() - st) * 1000.0
         )
 
     def explain(
@@ -602,7 +602,7 @@ class SearchCommands:
         Each parameter has a name and a value.
 
         """
-        st = time.time()
+        st = time.monotonic()
         cmd = [PROFILE_CMD, self.index_name, ""]
         if limited:
             cmd.append("LIMITED")
@@ -621,7 +621,7 @@ class SearchCommands:
         res = self.execute_command(*cmd)
 
         return self._parse_results(
-            PROFILE_CMD, res, query=query, duration=(time.time() - st) * 1000.0
+            PROFILE_CMD, res, query=query, duration=(time.monotonic() - st) * 1000.0
         )
 
     def spellcheck(self, query, distance=None, include=None, exclude=None):
@@ -940,7 +940,7 @@ class AsyncSearchCommands(SearchCommands):
         For more information see `FT.SEARCH <https://redis.io/commands/ft.search>`_.
         """  # noqa
         args, query = self._mk_query_args(query, query_params=query_params)
-        st = time.time()
+        st = time.monotonic()
 
         options = {}
         if get_protocol_version(self.client) not in ["3", 3]:
@@ -952,7 +952,7 @@ class AsyncSearchCommands(SearchCommands):
             return res
 
         return self._parse_results(
-            SEARCH_CMD, res, query=query, duration=(time.time() - st) * 1000.0
+            SEARCH_CMD, res, query=query, duration=(time.monotonic() - st) * 1000.0
         )
 
     async def aggregate(
diff --git a/redis/connection.py b/redis/connection.py
index 43501800..a298542c 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -4,11 +4,11 @@ import socket
 import ssl
 import sys
 import threading
+import time
 import weakref
 from abc import abstractmethod
 from itertools import chain
 from queue import Empty, Full, LifoQueue
-from time import time
 from typing import Any, Callable, Dict, List, Optional, Type, TypeVar, Union
 from urllib.parse import parse_qs, unquote, urlparse
 
@@ -542,7 +542,7 @@ class AbstractConnection(ConnectionInterface):
 
     def check_health(self):
         """Check the health of the connection with a PING/PONG"""
-        if self.health_check_interval and time() > self.next_health_check:
+        if self.health_check_interval and time.monotonic() > self.next_health_check:
             self.retry.call_with_retry(self._send_ping, self._ping_failed)
 
     def send_packed_command(self, command, check_health=True):
@@ -632,7 +632,7 @@ class AbstractConnection(ConnectionInterface):
             raise
 
         if self.health_check_interval:
-            self.next_health_check = time() + self.health_check_interval
+            self.next_health_check = time.monotonic() + self.health_check_interval
 
         if isinstance(response, ResponseError):
             try:
diff --git a/redis/exceptions.py b/redis/exceptions.py
index 82f62730..bad447a0 100644
--- a/redis/exceptions.py
+++ b/redis/exceptions.py
@@ -89,7 +89,7 @@ class LockError(RedisError, ValueError):
 
 
 class LockNotOwnedError(LockError):
-    "Error trying to extend or release a lock that is (no longer) owned"
+    "Error trying to extend or release a lock that is not owned (anymore)"
 
     pass
 
diff --git a/redis/lock.py b/redis/lock.py
index 7a1becb3..0288496e 100644
--- a/redis/lock.py
+++ b/redis/lock.py
@@ -1,3 +1,4 @@
+import logging
 import threading
 import time as mod_time
 import uuid
@@ -7,6 +8,8 @@ from typing import Optional, Type
 from redis.exceptions import LockError, LockNotOwnedError
 from redis.typing import Number
 
+logger = logging.getLogger(__name__)
+
 
 class Lock:
     """
@@ -82,6 +85,7 @@ class Lock:
         blocking: bool = True,
         blocking_timeout: Optional[Number] = None,
         thread_local: bool = True,
+        raise_on_release_error: bool = True,
     ):
         """
         Create a new Lock instance named ``name`` using the Redis client
@@ -125,6 +129,11 @@ class Lock:
                      thread-1 would see the token value as "xyz" and would be
                      able to successfully release the thread-2's lock.
 
+        ``raise_on_release_error`` indicates whether to raise an exception when
+        the lock is no longer owned when exiting the context manager. By default,
+        this is True, meaning an exception will be raised. If False, the warning
+        will be logged and the exception will be suppressed.
+
         In some use cases it's necessary to disable thread local storage. For
         example, if you have code where one thread acquires a lock and passes
         that lock instance to a worker thread to release later. If thread
@@ -140,6 +149,7 @@ class Lock:
         self.blocking = blocking
         self.blocking_timeout = blocking_timeout
         self.thread_local = bool(thread_local)
+        self.raise_on_release_error = raise_on_release_error
         self.local = threading.local() if self.thread_local else SimpleNamespace()
         self.local.token = None
         self.register_scripts()
@@ -168,7 +178,14 @@ class Lock:
         exc_value: Optional[BaseException],
         traceback: Optional[TracebackType],
     ) -> None:
-        self.release()
+        try:
+            self.release()
+        except LockError:
+            if self.raise_on_release_error:
+                raise
+            logger.warning(
+                "Lock was unlocked or no longer owned when exiting context manager."
+            )
 
     def acquire(
         self,


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index c485d626..7eaccb1a 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -237,7 +237,7 @@ def wait_for_cluster_creation(redis_url, cluster_nodes, timeout=60):
     :param cluster_nodes: The number of nodes in the cluster
     :param timeout: the amount of time to wait (in seconds)
     """
-    now = time.time()
+    now = time.monotonic()
     end_time = now + timeout
     client = None
     print(f"Waiting for {cluster_nodes} cluster nodes to become available")
@@ -250,7 +250,7 @@ def wait_for_cluster_creation(redis_url, cluster_nodes, timeout=60):
         except RedisClusterException:
             pass
         time.sleep(1)
-        now = time.time()
+        now = time.monotonic()
     if now >= end_time:
         available_nodes = 0 if client is None else len(client.get_nodes())
         raise RedisClusterException(
diff --git a/tests/test_asyncio/test_lock.py b/tests/test_asyncio/test_lock.py
index 9973ef70..be4270ac 100644
--- a/tests/test_asyncio/test_lock.py
+++ b/tests/test_asyncio/test_lock.py
@@ -129,6 +129,36 @@ class TestLock:
             async with self.get_lock(r, "foo", blocking_timeout=0.1):
                 pass
 
+    async def test_context_manager_not_raise_on_release_lock_not_owned_error(self, r):
+        try:
+            async with self.get_lock(
+                r, "foo", timeout=0.1, raise_on_release_error=False
+            ):
+                await asyncio.sleep(0.15)
+        except LockNotOwnedError:
+            pytest.fail("LockNotOwnedError should not have been raised")
+
+        with pytest.raises(LockNotOwnedError):
+            async with self.get_lock(
+                r, "foo", timeout=0.1, raise_on_release_error=True
+            ):
+                await asyncio.sleep(0.15)
+
+    async def test_context_manager_not_raise_on_release_lock_error(self, r):
+        try:
+            async with self.get_lock(
+                r, "foo", timeout=0.1, raise_on_release_error=False
+            ) as lock:
+                lock.release()
+        except LockError:
+            pytest.fail("LockError should not have been raised")
+
+        with pytest.raises(LockError):
+            async with self.get_lock(
+                r, "foo", timeout=0.1, raise_on_release_error=True
+            ) as lock:
+                lock.release()
+
     async def test_high_sleep_small_blocking_timeout(self, r):
         lock1 = self.get_lock(r, "foo")
         assert await lock1.acquire(blocking=False)
diff --git a/tests/test_connection_pool.py b/tests/test_connection_pool.py
index 65f42923..c92d84c2 100644
--- a/tests/test_connection_pool.py
+++ b/tests/test_connection_pool.py
@@ -167,11 +167,11 @@ class TestBlockingConnectionPool:
         )
         pool.get_connection()
 
-        start = time.time()
+        start = time.monotonic()
         with pytest.raises(redis.ConnectionError):
             pool.get_connection()
         # we should have waited at least 0.1 seconds
-        assert time.time() - start >= 0.1
+        assert time.monotonic() - start >= 0.1
 
     def test_connection_pool_blocks_until_conn_available(self, master_host):
         """
@@ -188,10 +188,10 @@ class TestBlockingConnectionPool:
             time.sleep(0.1)
             pool.release(c1)
 
-        start = time.time()
+        start = time.monotonic()
         Thread(target=target).start()
         pool.get_connection()
-        assert time.time() - start >= 0.1
+        assert time.monotonic() - start >= 0.1
 
     def test_reuse_previously_released_connection(self, master_host):
         connection_kwargs = {"host": master_host[0], "port": master_host[1]}
@@ -679,18 +679,18 @@ class TestHealthCheck:
         return _get_client(redis.Redis, request, health_check_interval=self.interval)
 
     def assert_interval_advanced(self, connection):
-        diff = connection.next_health_check - time.time()
+        diff = connection.next_health_check - time.monotonic()
         assert self.interval > diff > (self.interval - 1)
 
     def test_health_check_runs(self, r):
-        r.connection.next_health_check = time.time() - 1
+        r.connection.next_health_check = time.monotonic() - 1
         r.connection.check_health()
         self.assert_interval_advanced(r.connection)
 
     def test_arbitrary_command_invokes_health_check(self, r):
         # invoke a command to make sure the connection is entirely setup
         r.get("foo")
-        r.connection.next_health_check = time.time()
+        r.connection.next_health_check = time.monotonic()
         with mock.patch.object(
             r.connection, "send_command", wraps=r.connection.send_command
         ) as m:
diff --git a/tests/test_lock.py b/tests/test_lock.py
index 136c86e4..3d6d8146 100644
--- a/tests/test_lock.py
+++ b/tests/test_lock.py
@@ -133,6 +133,32 @@ class TestLock:
             with self.get_lock(r, "foo", blocking_timeout=0.1):
                 pass
 
+    def test_context_manager_not_raise_on_release_lock_not_owned_error(self, r):
+        try:
+            with self.get_lock(r, "foo", timeout=0.1, raise_on_release_error=False):
+                time.sleep(0.15)
+        except LockNotOwnedError:
+            pytest.fail("LockNotOwnedError should not have been raised")
+
+        with pytest.raises(LockNotOwnedError):
+            with self.get_lock(r, "foo", timeout=0.1, raise_on_release_error=True):
+                time.sleep(0.15)
+
+    def test_context_manager_not_raise_on_release_lock_error(self, r):
+        try:
+            with self.get_lock(
+                r, "foo", timeout=0.1, raise_on_release_error=False
+            ) as lock:
+                lock.release()
+        except LockError:
+            pytest.fail("LockError should not have been raised")
+
+        with pytest.raises(LockError):
+            with self.get_lock(
+                r, "foo", timeout=0.1, raise_on_release_error=True
+            ) as lock:
+                lock.release()
+
     def test_high_sleep_small_blocking_timeout(self, r):
         lock1 = self.get_lock(r, "foo")
         assert lock1.acquire(blocking=False)
diff --git a/tests/test_pubsub.py b/tests/test_pubsub.py
index fb46772a..9ead455a 100644
--- a/tests/test_pubsub.py
+++ b/tests/test_pubsub.py
@@ -23,7 +23,7 @@ from .conftest import (
 def wait_for_message(
     pubsub, timeout=0.5, ignore_subscribe_messages=False, node=None, func=None
 ):
-    now = time.time()
+    now = time.monotonic()
     timeout = now + timeout
     while now < timeout:
         if node:
@@ -39,7 +39,7 @@ def wait_for_message(
         if message is not None:
             return message
         time.sleep(0.01)
-        now = time.time()
+        now = time.monotonic()
     return None