Merge branch 'master' into feat/active-active

Author: vladvildanov

.github/workflows/integration.yaml

@@ -27,6 +27,8 @@ env:
   CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
   # this speeds up coverage with Python 3.12: https://github.com/nedbat/coveragepy/issues/1665
   COVERAGE_CORE: sysmon
+  # patch releases get included in the base version image when they are published
+  # for example after 8.2.1 is published, 8.2 image contains 8.2.1 content
   CURRENT_CLIENT_LIBS_TEST_STACK_IMAGE_TAG: '8.2'
   CURRENT_REDIS_VERSION: '8.2'
 
@@ -74,7 +76,7 @@ jobs:
       max-parallel: 15
       fail-fast: false
       matrix:
-        redis-version: ['8.2.1-pre', '${{ needs.redis_version.outputs.CURRENT }}', '8.0.2' ,'7.4.4', '7.2.9']
+        redis-version: ['8.4-M01-pre', '${{ needs.redis_version.outputs.CURRENT }}', '8.0.2' ,'7.4.4', '7.2.9']
         python-version: ['3.9', '3.13']
         parser-backend: ['plain']
         event-loop: ['asyncio']

pyproject.toml

@@ -83,6 +83,12 @@ filterwarnings = [
     # Ignore a coverage warning when COVERAGE_CORE=sysmon for Pythons < 3.12.
     "ignore:sys.monitoring isn't available:coverage.exceptions.CoverageWarning",
 ]
+log_cli_level = "INFO"
+log_cli_date_format = "%H:%M:%S:%f"
+log_cli = false
+log_cli_format = "%(asctime)s %(levelname)s %(threadName)s: %(message)s"
+log_level = "INFO"
+capture = "yes"
 
 [tool.ruff]
 target-version = "py39"

redis/asyncio/client.py

@@ -1161,9 +1161,12 @@ async def get_message(
             return await self.handle_message(response, ignore_subscribe_messages)
         return None
 
-    def ping(self, message=None) -> Awaitable:
+    def ping(self, message=None) -> Awaitable[bool]:
         """
-        Ping the Redis server
+        Ping the Redis server to test connectivity.
+
+        Sends a PING command to the Redis server and returns True if the server
+        responds with "PONG".
         """
         args = ["PING", message] if message is not None else ["PING"]
         return self.execute_command(*args)

redis/client.py

@@ -1186,7 +1186,10 @@ def get_message(
 
     def ping(self, message: Union[str, None] = None) -> bool:
         """
-        Ping the Redis server
+        Ping the Redis server to test connectivity.
+
+        Sends a PING command to the Redis server and returns True if the server
+        responds with "PONG".
         """
         args = ["PING", message] if message is not None else ["PING"]
         return self.execute_command(*args)

redis/commands/core.py

@@ -1210,11 +1210,18 @@ def latency_reset(self, *events: str) -> ResponseT:
         """
         return self.execute_command("LATENCY RESET", *events)
 
-    def ping(self, **kwargs) -> ResponseT:
+    def ping(self, **kwargs) -> Union[Awaitable[bool], bool]:
         """
-        Ping the Redis server
+        Ping the Redis server to test connectivity.
 
-        For more information, see https://redis.io/commands/ping
+        Sends a PING command to the Redis server and returns True if the server
+        responds with "PONG".
+
+        This command is useful for:
+        - Testing whether a connection is still alive
+        - Verifying the server's ability to serve data
+
+        For more information on the underlying ping command see https://redis.io/commands/ping
         """
         return self.execute_command("PING", **kwargs)
 

tests/test_asyncio/test_commands.py

@@ -3158,6 +3158,7 @@ async def test_xgroup_destroy(self, r: redis.Redis):
         assert await r.xgroup_destroy(stream, group)
 
     @skip_if_server_version_lt("7.0.0")
+    @skip_if_server_version_gte("8.2.2")
     async def test_xgroup_setid(self, r: redis.Redis):
         stream = "stream"
         group = "group"
@@ -3178,6 +3179,28 @@ async def test_xgroup_setid(self, r: redis.Redis):
         ]
         assert await r.xinfo_groups(stream) == expected
 
+    @skip_if_server_version_lt("8.2.2")
+    async def test_xgroup_setid_fixed_max_entries_read(self, r):
+        stream = "stream"
+        group = "group"
+        message_id = await r.xadd(stream, {"foo": "bar"})
+        await r.xadd(stream, {"foo1": "bar1"})
+
+        await r.xgroup_create(stream, group, 0)
+        # advance the last_delivered_id to the message_id
+        await r.xgroup_setid(stream, group, message_id, entries_read=2)
+        expected = [
+            {
+                "name": group.encode(),
+                "consumers": 0,
+                "pending": 0,
+                "last-delivered-id": message_id,
+                "entries-read": 2,
+                "lag": 0,
+            }
+        ]
+        assert await r.xinfo_groups(stream) == expected
+
     @skip_if_server_version_lt("7.2.0")
     async def test_xinfo_consumers(self, r: redis.Redis):
         stream = "stream"

tests/test_asyncio/test_vsets.py

@@ -502,8 +502,8 @@ async def test_vsim_epsilon(d_client):
     await d_client.vset().vadd("myset", [2, 1, 1], "a")
     await d_client.vset().vadd("myset", [2, 0, 1], "b")
     await d_client.vset().vadd("myset", [2, 0, 0], "c")
-    await d_client.vset().vadd("myset", [2, 0, -1], "d")
-    await d_client.vset().vadd("myset", [2, -1, -1], "e")
+    await d_client.vset().vadd("myset", [2, 0, 2], "d")
+    await d_client.vset().vadd("myset", [-2, -1, -1], "e")
 
     res1 = await d_client.vset().vsim("myset", [2, 1, 1])
     assert 5 == len(res1)

tests/test_commands.py

@@ -4607,6 +4607,7 @@ def test_xgroup_create_mkstream(self, r):
         assert r.xinfo_groups(stream) == expected
 
     @skip_if_server_version_lt("7.0.0")
+    @skip_if_server_version_gte("8.2.2")
     def test_xgroup_create_entriesread(self, r: redis.Redis):
         stream = "stream"
         group = "group"
@@ -4628,6 +4629,33 @@ def test_xgroup_create_entriesread(self, r: redis.Redis):
         ]
         assert r.xinfo_groups(stream) == expected
 
+    @skip_if_server_version_lt("8.2.2")
+    def test_xgroup_create_entriesread_fixed_max_entries_read(self, r: redis.Redis):
+        stream = "stream"
+        group = "group"
+        r.xadd(stream, {"foo": "bar"})
+        r.xadd(stream, {"foo1": "bar1"})
+        r.xadd(stream, {"foo2": "bar2"})
+
+        # no group is setup yet, no info to obtain
+        assert r.xinfo_groups(stream) == []
+
+        assert r.xgroup_create(stream, group, 0, entries_read=7)
+        # validate the entries-read is max the number of entries
+        # in the stream and lag shows the entries between
+        # last_delivered_id and entries_added
+        expected = [
+            {
+                "name": group.encode(),
+                "consumers": 0,
+                "pending": 0,
+                "last-delivered-id": b"0-0",
+                "entries-read": 3,
+                "lag": 3,
+            }
+        ]
+        assert r.xinfo_groups(stream) == expected
+
     @skip_if_server_version_lt("5.0.0")
     def test_xgroup_delconsumer(self, r):
         stream = "stream"
@@ -4675,6 +4703,7 @@ def test_xgroup_destroy(self, r):
         assert r.xgroup_destroy(stream, group)
 
     @skip_if_server_version_lt("7.0.0")
+    @skip_if_server_version_gte("8.2.2")
     def test_xgroup_setid(self, r):
         stream = "stream"
         group = "group"
@@ -4695,6 +4724,28 @@ def test_xgroup_setid(self, r):
         ]
         assert r.xinfo_groups(stream) == expected
 
+    @skip_if_server_version_lt("8.2.2")
+    def test_xgroup_setid_fixed_max_entries_read(self, r):
+        stream = "stream"
+        group = "group"
+        message_id = r.xadd(stream, {"foo": "bar"})
+        r.xadd(stream, {"foo1": "bar1"})
+
+        r.xgroup_create(stream, group, 0)
+        # advance the last_delivered_id to the message_id
+        r.xgroup_setid(stream, group, message_id, entries_read=5)
+        expected = [
+            {
+                "name": group.encode(),
+                "consumers": 0,
+                "pending": 0,
+                "last-delivered-id": message_id,
+                "entries-read": 2,
+                "lag": 0,
+            }
+        ]
+        assert r.xinfo_groups(stream) == expected
+
     @skip_if_server_version_lt("7.2.0")
     def test_xinfo_consumers(self, r):
         stream = "stream"

tests/test_scenario/maint_notifications_helpers.py

@@ -116,8 +116,9 @@ def find_target_node_and_empty_node(
 
         # Get all node IDs from CLUSTER NODES section
         all_nodes = set()
-        nodes_with_shards = set()
-        master_nodes = set()
+        nodes_with_any_shards = set()  # Nodes with shards from ANY database
+        nodes_with_target_db_shards = set()  # Nodes with shards from target database
+        master_nodes = set()  # Master nodes for target database only
 
         for line in lines:
             line = line.strip()
@@ -146,31 +147,45 @@ def find_target_node_and_empty_node(
                 # Parse shard line: db:1  m-standard  redis:1  node:2  master  0-8191  1.4MB  OK
                 parts = line.split()
                 if len(parts) >= 5:
+                    db_id = parts[0]  # db:1, db:2, etc.
                     node_id = parts[3]  # node:2
                     shard_role = parts[4]  # master/slave - this is what matters
 
-                    nodes_with_shards.add(node_id)
-                    if shard_role == "master":
-                        master_nodes.add(node_id)
+                    # Track ALL nodes with shards (for finding truly empty nodes)
+                    nodes_with_any_shards.add(node_id)
+
+                    # Only track master nodes for the specific database we're testing
+                    bdb_id = endpoint_config.get("bdb_id")
+                    if db_id == f"db:{bdb_id}":
+                        nodes_with_target_db_shards.add(node_id)
+                        if shard_role == "master":
+                            master_nodes.add(node_id)
             elif line.startswith("ENDPOINTS:") or not line:
                 shards_section_started = False
 
-        # Find empty node (node with no shards)
-        empty_nodes = all_nodes - nodes_with_shards
+        # Find empty node (node with no shards from ANY database)
+        nodes_with_no_shards_target_bdb = all_nodes - nodes_with_target_db_shards
 
         logging.debug(f"All nodes: {all_nodes}")
-        logging.debug(f"Nodes with shards: {nodes_with_shards}")
-        logging.debug(f"Master nodes: {master_nodes}")
-        logging.debug(f"Empty nodes: {empty_nodes}")
+        logging.debug(f"Nodes with shards from any database: {nodes_with_any_shards}")
+        logging.debug(
+            f"Nodes with target database shards: {nodes_with_target_db_shards}"
+        )
+        logging.debug(f"Master nodes (target database only): {master_nodes}")
+        logging.debug(
+            f"Nodes with no shards from target database: {nodes_with_no_shards_target_bdb}"
+        )
 
-        if not empty_nodes:
-            raise ValueError("No empty nodes (nodes without shards) found")
+        if not nodes_with_no_shards_target_bdb:
+            raise ValueError("All nodes have shards from target database")
 
         if not master_nodes:
-            raise ValueError("No nodes with master shards found")
+            raise ValueError("No nodes with master shards from target database found")
 
         # Return the first available empty node and master node (numeric part only)
-        empty_node = next(iter(empty_nodes)).split(":")[1]  # node:1 -> 1
+        empty_node = next(iter(nodes_with_no_shards_target_bdb)).split(":")[
+            1
+        ]  # node:1 -> 1
         target_node = next(iter(master_nodes)).split(":")[1]  # node:2 -> 2
 
         return target_node, empty_node

tests/test_scenario/test_maint_notifications.py

@@ -36,7 +36,7 @@
     datefmt="%Y-%m-%d %H:%M:%S:%f",
 )
 
-BIND_TIMEOUT = 30
+BIND_TIMEOUT = 60
 MIGRATE_TIMEOUT = 60
 FAILOVER_TIMEOUT = 15
 
@@ -109,29 +109,6 @@ def setup_and_cleanup(
             except Exception as e:
                 logging.error(f"Failed to revert failover: {e}")
 
-        if self._migration_executed:
-            try:
-                if self.target_node and self.empty_node:
-                    self._execute_migration(
-                        fault_injector_client=fault_injector_client,
-                        endpoints_config=endpoints_config,
-                        target_node=self.empty_node,
-                        empty_node=self.target_node,
-                    )
-                    logging.info("Migration cleanup completed")
-            except Exception as e:
-                logging.error(f"Failed to revert migration: {e}")
-
-        if self._bind_executed:
-            try:
-                if self.endpoint_id:
-                    self._execute_bind(
-                        fault_injector_client, endpoints_config, self.endpoint_id
-                    )
-                    logging.info("Bind cleanup completed")
-            except Exception as e:
-                logging.error(f"Failed to revert bind endpoint: {e}")
-
         logging.info("Cleanup finished")
 
     def _execute_failover(
@@ -916,7 +893,15 @@ def test_new_connections_receive_migrating(
         )
 
         migrate_thread.join()
+        logging.info("Executing rladmin bind endpoint command for cleanup...")
 
+        bind_thread = Thread(
+            target=self._execute_bind,
+            name="bind_thread",
+            args=(fault_injector_client, endpoints_config, self.endpoint_id),
+        )
+        bind_thread.start()
+        bind_thread.join()
         client_maint_notifications.connection_pool.release(first_conn)
         client_maint_notifications.connection_pool.release(second_connection)