Attempting to fix cluster issues

Author: abrookins

.gitignore

@@ -216,10 +216,17 @@ pyrightconfig.json
 [Ll]ocal
 pyvenv.cfg
 pip-selfcheck.json
+env
+venv
+.venv
 
 libs/redis/docs/.Trash*
 .python-version
 .idea/*
 .vscode/settings.json
 .python-version
 tests/data
+.git
+.cursor
+.junie
+.undodir

redisvl/extensions/cache/base.py

@@ -216,7 +216,14 @@ async def aclear(self) -> None:
                 await client.delete(*keys)
             if cursor_int == 0:  # Redis returns 0 when scan is complete
                 break
-            cursor = cursor_int  # Update cursor for next iteration
+            # Cluster returns a dict of cursor values. We need to stop if these all
+            # come back as 0.
+            elif isinstance(cursor_int, Mapping):
+                cursor_values = list(cursor_int.values())
+                if all(v == 0 for v in cursor_values):
+                    break
+            else:
+                cursor = cursor_int  # Update cursor for next iteration
 
     def disconnect(self) -> None:
         """Disconnect from Redis."""

redisvl/extensions/cache/embeddings/embeddings.py

@@ -47,6 +47,7 @@ def __init__(
             name=name,
             ttl=ttl,
             redis_client=redis_client,
+            async_redis_client=async_redis_client,
             redis_url=redis_url,
             connection_kwargs=connection_kwargs,
         )

tests/cluster-compose.yml

@@ -132,28 +132,28 @@ services:
         NODES="redis-node-1:7001 redis-node-2:7002 redis-node-3:7003 redis-node-4:7004 redis-node-5:7005 redis-node-6:7006"
 
         echo "Force resetting all nodes before cluster creation..."
-        for NODE_ADDR_PORT in $NODES; do
-          NODE_HOST=$(echo $NODE_ADDR_PORT | cut -d':' -f1)
-          NODE_PORT=$(echo $NODE_ADDR_PORT | cut -d':' -f2)
-          echo "Resetting node $NODE_HOST:$NODE_PORT"
+        for NODE_ADDR_PORT in $$NODES; do
+          NODE_HOST=$$(echo $$NODE_ADDR_PORT | cut -d':' -f1)
+          NODE_PORT=$$(echo $$NODE_ADDR_PORT | cut -d':' -f2)
+          echo "Resetting node $$NODE_HOST:$$NODE_PORT"
           
           # Wait for node to be responsive
           retry_count=0
           max_retries=10
-          until redis-cli -h $NODE_HOST -p $NODE_PORT ping 2>/dev/null | grep -q PONG; do
-            retry_count=$((retry_count+1))
-            if [ "$retry_count" -gt "$max_retries" ]; then
-              echo "Error: Node $NODE_HOST:$NODE_PORT did not respond after $max_retries retries."
+          until redis-cli -h $$NODE_HOST -p $$NODE_PORT ping 2>/dev/null | grep -q PONG; do
+            retry_count=$$((retry_count+1))
+            if [ "$$retry_count" -gt "$$max_retries" ]; then
+              echo "Error: Node $$NODE_HOST:$$NODE_PORT did not respond after $$max_retries retries."
               exit 1 # Exit if a node is unresponsive
             fi
-            echo "Waiting for $NODE_HOST:$NODE_PORT to respond (attempt $retry_count/$max_retries)..."
+            echo "Waiting for $$NODE_HOST:$$NODE_PORT to respond (attempt $$retry_count/$$max_retries)..."
             sleep 3 # Increased sleep between pings
           done
           
-          echo "Flushing and hard resetting $NODE_HOST:$NODE_PORT"
-          redis-cli -h $NODE_HOST -p $NODE_PORT FLUSHALL || echo "Warning: FLUSHALL failed on $NODE_HOST:$NODE_PORT, attempting to continue..."
+          echo "Flushing and hard resetting $$NODE_HOST:$$NODE_PORT"
+          redis-cli -h $$NODE_HOST -p $$NODE_PORT FLUSHALL || echo "Warning: FLUSHALL failed on $$NODE_HOST:$$NODE_PORT, attempting to continue..."
           # Use CLUSTER RESET HARD
-          redis-cli -h $NODE_HOST -p $NODE_PORT CLUSTER RESET HARD || echo "Warning: CLUSTER RESET HARD failed on $NODE_HOST:$NODE_PORT, attempting to continue..."
+          redis-cli -h $$NODE_HOST -p $$NODE_PORT CLUSTER RESET HARD || echo "Warning: CLUSTER RESET HARD failed on $$NODE_HOST:$$NODE_PORT, attempting to continue..."
         done
         echo "Node reset complete."
         sleep 5 # Give a moment for resets to settle
@@ -162,29 +162,29 @@ services:
         ATTEMPT=1
         CLUSTER_CREATED=false
 
-        while [ $ATTEMPT -le $MAX_ATTEMPTS ]; do
-          echo "Attempting to create Redis cluster (Attempt $ATTEMPT/$MAX_ATTEMPTS)..."
-          output=$(echo yes | redis-cli --cluster create \
-            $NODES \
+        while [ $$ATTEMPT -le $$MAX_ATTEMPTS ]; do
+          echo "Attempting to create Redis cluster (Attempt $$ATTEMPT/$$MAX_ATTEMPTS)..."
+          output=$$(echo yes | redis-cli --cluster create \
+            $$NODES \
             --cluster-replicas 1 2>&1)
 
-          if echo "$output" | grep -q "\[OK\] All 16384 slots covered."; then
+          if echo "$$output" | grep -q "\[OK\] All 16384 slots covered."; then
             echo "Cluster created successfully."
             CLUSTER_CREATED=true
             break
           else
-            echo "Failed to create cluster on attempt $ATTEMPT."
-            echo "Output from redis-cli: $output"
-            if [ $ATTEMPT -lt $MAX_ATTEMPTS ]; then
+            echo "Failed to create cluster on attempt $$ATTEMPT."
+            echo "Output from redis-cli: $$output"
+            if [ $$ATTEMPT -lt $$MAX_ATTEMPTS ]; then
               echo "Retrying in 10 seconds..."
               sleep 10
             fi
           fi
-          ATTEMPT=$((ATTEMPT + 1))
+          ATTEMPT=$$((ATTEMPT + 1))
         done
 
-        if [ "$CLUSTER_CREATED" = "false" ]; then
-          echo "Failed to create cluster after $MAX_ATTEMPTS attempts. Exiting."
+        if [ "$$CLUSTER_CREATED" = "false" ]; then
+          echo "Failed to create cluster after $$MAX_ATTEMPTS attempts. Exiting."
           exit 1
         fi
 

tests/conftest.py

@@ -122,9 +122,9 @@ def redis_cluster_container(worker_id):
                     capture_output=True,
                     text=True,
                 )
-                logger.info("Docker Compose logs:\n", logs_result.stdout)
+                logger.info("Docker Compose logs:\n%s", logs_result.stdout)
                 if logs_result.stderr:
-                    logger.error("Docker Compose logs stderr:\n", logs_result.stderr)
+                    logger.error("Docker Compose logs stderr:\n%s", logs_result.stderr)
             except Exception as log_e:
                 logger.error(f"Failed to get Docker Compose logs: {repr(log_e)}")
 

tests/integration/test_redis_cluster_support.py

@@ -83,29 +83,38 @@ async def test_async_search_index_client(redis_cluster_url):
     # Test with AsyncRedis client
     cluster_client = AsyncRedisCluster.from_url(redis_cluster_url)
     index = AsyncSearchIndex(schema=schema, redis_client=cluster_client)
-    await index.create(overwrite=True)
-    await index.load([{"name": "async_test", "age": 25}])
-    results = await index.query(TextQuery("async_test", "name"))
-    assert results[0]["name"] == "async_test"
-    await index.delete(drop=True)
+    try:
+        await index.create(overwrite=True)
+        await index.load([{"name": "async_test", "age": 25}])
+        results = await index.query(TextQuery("async_test", "name"))
+        assert results[0]["name"] == "async_test"
+        await index.delete(drop=True)
+    finally:
+        # Manually close the cluster client to prevent connection leaks
+        await cluster_client.aclose()
 
 
-def test_embeddings_cache_cluster_async(redis_cluster_url):
+@pytest.mark.asyncio
+async def test_embeddings_cache_cluster_async(redis_cluster_url):
     """Test that EmbeddingsCache correctly handles AsyncRedisCluster clients."""
     cluster_client = RedisConnectionFactory.get_async_redis_cluster_connection(
         redis_cluster_url
     )
     cache = EmbeddingsCache(async_redis_client=cluster_client)
 
-    cache.set(
-        text="hey",
-        model_name="test",
-        embedding=[1, 2, 3],
-    )
-    result = cache.get("hey", "test")
-    assert result is not None
-    assert result["embedding"] == [1, 2, 3]
-    cache.clear()
+    try:
+        await cache.aset(
+            text="hey",
+            model_name="test",
+            embedding=[1, 2, 3],
+        )
+        result = await cache.aget("hey", "test")
+        assert result is not None
+        assert result["embedding"] == [1, 2, 3]
+        await cache.aclear()
+    finally:
+        # Manually close the cluster client to prevent connection leaks
+        await cluster_client.aclose()
 
 
 def test_embeddings_cache_cluster_sync(redis_cluster_url):