perf(embeddings-cache): refresh and set TTLs in the same round trip as each lookup and write

docs/user_guide/installation.md

@@ -227,7 +227,7 @@ RedisSearchError: Error while fetching llmcache index info:
 User <name> has no permissions to run the 'FT.INFO' command
 ```
 
-Reading a cache is not read-only either. When a cache has a TTL configured — at construction or later via `set_ttl()` — every read that hits refreshes the matched entries' TTL, so it issues `EXPIRE`. That includes `SemanticCache.check()`, every `EmbeddingsCache` getter, their async equivalents, and the reads a vectorizer performs for you when constructed with `cache=`. `EXPIRE` is in `@write`, not `@read` (measured with `ACL CAT write`), so a lookup-only credential fails on a cache *hit* rather than on the write that populated it.
+Reading a cache is not read-only either. When a cache has a TTL configured — at construction or later via `set_ttl()` — every read that hits refreshes the matched entries' TTL, so it issues `EXPIRE`. That includes `SemanticCache.check()`, every `EmbeddingsCache` getter, their async equivalents, and the reads a vectorizer performs for you when constructed with `cache=`. `EXPIRE` is in `@write`, not `@read` (measured with `ACL CAT write`), so a lookup-only credential fails on a cache *hit* rather than on the write that populated it. `EmbeddingsCache` getters send `EXPIRE` in the same pipeline as the lookup, so for them it fails on a miss too.
 
 Granting the command is only half of it, because the key patterns must permit writes too. Measured on 8.4.5 with `ACL DRYRUN <user> EXPIRE llmcache:abc 60`:
 

redisvl/extensions/cache/embeddings/embeddings.py

@@ -189,16 +189,7 @@ def get_by_key(self, key: str) -> dict[str, Any] | None:
                 )
                 EmbeddingsCache._warning_shown = True
 
-        client = self._get_redis_client()
-
-        # Get all fields
-        data = client.hgetall(key)
-
-        # Refresh TTL if data exists
-        if data:
-            self.expire(key)
-
-        return self._process_cache_data(data)  # type: ignore
+        return self.mget_by_keys([key])[0]
 
     def mget_by_keys(self, keys: list[str]) -> list[dict[str, Any] | None]:
         """Get multiple embeddings by their Redis keys.
@@ -233,21 +224,21 @@ def mget_by_keys(self, keys: list[str]) -> list[dict[str, Any] | None]:
                 EmbeddingsCache._warning_shown = True
 
         client = self._get_redis_client()
+        _ttl = self._resolve_ttl()
 
+        # Not transactional, so on a Redis Cluster the keys may span slots.
         with client.pipeline(transaction=False) as pipeline:
-            # Queue all hgetall operations
+            # Refresh each TTL in the same pipeline; EXPIRE on a missing key is a no-op.
             for key in keys:
                 pipeline.hgetall(key)
+                if _ttl:
+                    pipeline.expire(key, _ttl)
             results = pipeline.execute()
 
-        # Process results
-        processed_results = []
-        for i, result in enumerate(results):
-            if result:  # If cache hit, refresh TTL separately
-                self.expire(keys[i])
-            processed_results.append(self._process_cache_data(result))
-
-        return processed_results
+        # Replies alternate HGETALL, EXPIRE; keep the lookups.
+        if _ttl:
+            results = results[::2]
+        return [self._process_cache_data(result) for result in results]
 
     def mget(
         self, contents: Iterable[bytes | str], model_name: str
@@ -310,11 +301,6 @@ def set(
                 metadata={"source": "user_query"}
             )
         """
-        # Prepare data
-        key, cache_entry = self._prepare_entry_data(
-            content, model_name, embedding, metadata
-        )
-
         if self._should_warn_for_async_only():
             if not EmbeddingsCache._warning_shown:
                 logger.warning(
@@ -323,14 +309,13 @@ def set(
                 )
                 EmbeddingsCache._warning_shown = True
 
-        # Store in Redis
-        client = self._get_redis_client()
-        client.hset(name=key, mapping=cache_entry)  # type: ignore[arg-type]
-
-        # Set TTL if specified
-        self.expire(key, ttl)
-
-        return key
+        item = {
+            "content": content,
+            "model_name": model_name,
+            "embedding": embedding,
+            "metadata": metadata,
+        }
+        return self.mset([item], ttl=ttl)[0]
 
     def mset(
         self,
@@ -624,16 +609,7 @@ async def aget_by_key(self, key: str) -> dict[str, Any] | None:
 
             embedding_data = await cache.aget_by_key("embedcache:1234567890abcdef")
         """
-        client = await self._get_async_redis_client()
-
-        # Get all fields
-        data = await client.hgetall(key)  # type: ignore
-
-        # Refresh TTL if data exists
-        if data:
-            await self.aexpire(key)
-
-        return self._process_cache_data(data)
+        return (await self.amget_by_keys([key]))[0]
 
     async def amget_by_keys(self, keys: list[str]) -> list[dict[str, Any] | None]:
         """Async get multiple embeddings by their Redis keys.
@@ -660,22 +636,21 @@ async def amget_by_keys(self, keys: list[str]) -> list[dict[str, Any] | None]:
             return []
 
         client = await self._get_async_redis_client()
+        _ttl = self._resolve_ttl()
 
-        # Use pipeline only for retrieval
+        # Not transactional, so on a Redis Cluster the keys may span slots.
         async with client.pipeline(transaction=False) as pipeline:
-            # Queue all hgetall operations
+            # Refresh each TTL in the same pipeline; EXPIRE on a missing key is a no-op.
             for key in keys:
                 pipeline.hgetall(key)
+                if _ttl:
+                    pipeline.expire(key, _ttl)
             results = await pipeline.execute()
 
-        # Process results and refresh TTLs separately
-        processed_results = []
-        for i, result in enumerate(results):
-            if result:  # If cache hit, refresh TTL
-                await self.aexpire(keys[i])
-            processed_results.append(self._process_cache_data(result))
-
-        return processed_results
+        # Replies alternate HGETALL, EXPIRE; keep the lookups.
+        if _ttl:
+            results = results[::2]
+        return [self._process_cache_data(result) for result in results]
 
     async def amget(
         self, contents: Iterable[bytes | str], model_name: str
@@ -740,19 +715,13 @@ async def aset(
                 metadata={"source": "user_query"}
             )
         """
-        # Prepare data
-        key, cache_entry = self._prepare_entry_data(
-            content, model_name, embedding, metadata
-        )
-
-        # Store in Redis
-        client = await self._get_async_redis_client()
-        await client.hset(name=key, mapping=cache_entry)  # type: ignore
-
-        # Set TTL if specified
-        await self.aexpire(key, ttl)
-
-        return key
+        item = {
+            "content": content,
+            "model_name": model_name,
+            "embedding": embedding,
+            "metadata": metadata,
+        }
+        return (await self.amset([item], ttl=ttl))[0]
 
     async def amset(
         self,

tests/integration/test_embedcache.py

@@ -728,6 +728,36 @@ def test_batch_with_ttl(cache_with_ttl, sample_embedding_data):
     assert all(exist_results)
 
 
+@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"])
+@pytest.mark.asyncio
+async def test_mget_by_keys_refreshes_every_hit_ttl(
+    cache_with_ttl, sample_embedding_data, monkeypatch, use_async
+):
+    """Every hit's TTL is reset within the lookup pipeline; a miss is not created."""
+    items = [
+        {key: sample[key] for key in ("content", "model_name", "embedding")}
+        for sample in sample_embedding_data
+    ]
+    keys = cache_with_ttl.mset(items, ttl=1000)
+    missing = cache_with_ttl._make_key("missing")
+
+    def refuse(*args, **kwargs):
+        raise AssertionError("TTL refreshed outside the lookup pipeline")
+
+    monkeypatch.setattr(cache_with_ttl, "expire", refuse)
+    monkeypatch.setattr(cache_with_ttl, "aexpire", refuse)
+
+    if use_async:
+        results = await cache_with_ttl.amget_by_keys(keys + [missing])
+    else:
+        results = cache_with_ttl.mget_by_keys(keys + [missing])
+
+    assert [result is not None for result in results] == [True, True, True, False]
+    client = cache_with_ttl._get_redis_client()
+    assert all(0 < client.ttl(key) <= 2 for key in keys)
+    assert client.exists(missing) == 0
+
+
 def test_large_batch_operations(cache):
     """Test operations with larger batches to ensure scalability."""
     # Create a larger batch of items

tests/integration/test_embedcache_warnings.py

@@ -30,10 +30,9 @@ async def test_sync_methods_warn_with_async_only_client(
 
     # Mock _get_redis_client to prevent actual connection attempt
     with patch.object(cache, "_get_redis_client") as mock_get_client:
-        # Mock the Redis client methods that would be called
-        mock_client = mock_get_client.return_value
-        mock_client.hgetall.return_value = {}  # Empty result for get_by_key
-        mock_client.hset.return_value = 1  # Success for set
+        # Both calls run through a pipeline; with no TTL, one empty reply is a miss
+        pipeline = mock_get_client.return_value.pipeline.return_value.__enter__()
+        pipeline.execute.return_value = [{}]
 
         # Capture log warnings
         with caplog.at_level(logging.WARNING):
@@ -46,6 +45,7 @@ async def test_sync_methods_warn_with_async_only_client(
                 "initialized with async_redis_client only" in caplog.records[0].message
             )
             assert "Use async methods" in caplog.records[0].message
+            assert "(get_by_key)" in caplog.records[0].message
 
             # Clear captured logs
             caplog.clear()