fix: tolerate matched docs with missing field payload (Redis 8.8 expiry race)

docs/concepts/queries.md

@@ -350,5 +350,39 @@ query = HybridQuery(
 )
 ```
 
+## Results and expiring documents
+
+RedisVL returns query results as a list of dictionaries, one per matched document. On Redis 8.8+ a document can occasionally come back with its field payload missing: the RediSearch worker pool now defaults to a nonzero number of background worker threads, and if a document — or one of its indexed fields — expires (via TTL or `HPEXPIRE`) at the exact moment a background search is reading it, the server returns the matched id with an empty field payload instead of dropping it from the result set. This is most likely in workloads that combine search with short TTLs, such as a semantic cache or chat history.
+
+RedisVL detects and **silently skips** such a document (logging at `WARNING` level) whenever the query type guarantees a field would be present on a healthy match:
+
+- **`VectorQuery` / `VectorRangeQuery`** with `return_score=True` (the default) — a healthy match always carries a `vector_distance`.
+- **`FilterQuery` on JSON storage** that returns the whole object — a healthy match always carries its JSON payload.
+
+For a plain **`FilterQuery` / `TextQuery` on hash storage**, a race victim is indistinguishable from a legitimately sparse result (for example, a document matched via an `INDEXMISSING` field, or a query that requested only the `id`). RedisVL cannot safely tell those apart, so it returns the document as an id-only `{"id": ...}` dict rather than risk dropping a valid result. If you query hash storage directly this way, use `.get()` for optional fields rather than assuming every field is present.
+
+### Detecting incomplete results
+
+`SearchIndex.query()` returns a `SearchResults` object, which is a normal list of result dictionaries with two extra attributes so you can tell a race-shortened result set apart from one that genuinely had fewer matches:
+
+- `results.dropped_count` — how many matched documents were skipped because their field payload was missing (`0` in the normal case).
+- `results.complete` — `False` when any document was dropped.
+
+```python
+results = index.query(query)
+
+if not results.complete:
+    # e.g. retry, reconcile against a CountQuery, or annotate the answer
+    logger.warning("Result set is incomplete: %d dropped", results.dropped_count)
+```
+
+`SearchResults` behaves exactly like a `list` everywhere else, so existing code needs no changes. (Operations that build a new list — slicing, `sorted()`, concatenation — return a plain `list` without these attributes.)
+
+Practical consequences:
+
+- A query may return **fewer results than `num_results`** (or fewer than `page_size` when paginating) when some matched documents were expiring. Treat those limits as upper bounds, not guarantees; use `results.complete` to detect when it happened.
+- A `CountQuery` reports the server's match count, which still includes the expiring document, so a count can legitimately exceed the number of documents a materializing query returns at the same instant.
+- The higher-level extensions build on this: the **semantic cache** drops an expiring hit (a benign cache miss), **message history** drops an expiring message from its formatted output, and the **semantic router** drops an expiring route candidate (which, in the rare case the best match is the one expiring, can shift the selected route). Message history's `raw=True` mode returns the unprocessed hash entries and may therefore still surface an id-only record.
+
 **Learn more:** {doc}`/user_guide/11_advanced_queries` demonstrates these query types in detail.
 

redisvl/extensions/cache/llm/semantic.py

@@ -1,6 +1,7 @@
 import asyncio
 from typing import Any
 
+from pydantic import ValidationError
 from redis import Redis
 
 from redisvl.extensions.cache.llm.base import BaseLLMCache
@@ -544,10 +545,28 @@ def _process_cache_results(
         for cache_search_result in cache_search_results:
             # Pop the redis key from the result
             redis_key = cache_search_result.pop("id")
-            redis_keys.append(redis_key)
 
-            # Create and process cache hit
-            cache_hit = CacheHit(**cache_search_result)
+            # Create and process cache hit. A matched entry whose field payload
+            # came back missing (the Redis 8.8+ background-search expiry race)
+            # would arrive as an id-only dict and fail validation; skip it and
+            # do not refresh its TTL rather than raise. The core parser already
+            # drops these upstream, so this is defense-in-depth.
+            # Only ValidationError is expected here: CacheHit has no
+            # before-validator that does keyed access on the input (unlike
+            # message history's ChatMessage.generate_id, whose guard must also
+            # catch KeyError). If CacheHit ever gains such a validator, widen
+            # this accordingly.
+            try:
+                cache_hit = CacheHit(**cache_search_result)
+            except ValidationError:
+                logger.warning(
+                    "Skipping cache hit with missing field data (likely expired "
+                    "during a background search on Redis 8.8+): key=%s",
+                    redis_key,
+                )
+                continue
+
+            redis_keys.append(redis_key)
             cache_hit_dict = cache_hit.to_dict()
 
             # Filter down to only selected return fields if needed

redisvl/extensions/message_history/base_history.py

@@ -1,15 +1,20 @@
 import warnings
 from typing import Any
 
+from pydantic import ValidationError
+
 from redisvl.extensions.constants import (
     CONTENT_FIELD_NAME,
     METADATA_FIELD_NAME,
     ROLE_FIELD_NAME,
     TOOL_FIELD_NAME,
 )
 from redisvl.extensions.message_history.schema import ChatMessage, ChatRole
+from redisvl.utils.log import get_logger
 from redisvl.utils.utils import create_ulid, deserialize
 
+logger = get_logger(__name__)
+
 
 class BaseMessageHistory:
 
@@ -167,7 +172,21 @@ def _format_context(
 
         for message in messages:
 
-            chat_message = ChatMessage(**message)
+            try:
+                chat_message = ChatMessage(**message)
+            except (ValidationError, KeyError):
+                # A matched message whose field payload came back missing (the
+                # Redis 8.8+ background-search expiry race) would arrive as an
+                # id-only dict and fail construction -- either a ValidationError
+                # for missing required fields, or a KeyError raised by the
+                # ChatMessage.generate_id validator when session_tag is absent.
+                # Skip it rather than raise.
+                logger.warning(
+                    "Skipping message with missing field data (likely expired "
+                    "during a background search on Redis 8.8+): id=%s",
+                    message.get("id"),
+                )
+                continue
 
             if as_text:
                 context.append(chat_message.content)

redisvl/extensions/router/semantic.py

@@ -324,9 +324,19 @@ def get(self, route_name: str) -> Route | None:
         """
         return next((route for route in self.routes if route.name == route_name), None)
 
-    def _process_route(self, result: dict[str, Any]) -> RouteMatch:
-        """Process resulting route objects and metadata."""
+    def _process_route(self, result: dict[str, Any]) -> RouteMatch | None:
+        """Process resulting route objects and metadata.
+
+        Returns None for a row missing ``route_name``/``distance`` — e.g. a doc
+        that expired mid-search on Redis 8.8+ (the background-search expiry
+        race) — so the caller can drop it rather than raise a KeyError. Note
+        this is a stricter rule than the core ``process_aggregate_results``,
+        which only drops entirely-empty rows; the router applies it because it
+        knows its exact aggregate schema. The caller aggregates the warning.
+        """
         route_dict = make_dict(convert_bytes(result))
+        if "route_name" not in route_dict or "distance" not in route_dict:
+            return None
         return RouteMatch(
             name=route_dict["route_name"], distance=float(route_dict["distance"])
         )
@@ -409,10 +419,19 @@ def _get_route_matches(
                 )
             raise e
 
-        # process aggregation results into route matches
-        return [
+        # process aggregation results into route matches, dropping any row whose
+        # field payload came back missing (the Redis 8.8+ expiry race)
+        processed = [
             self._process_route(route_match) for route_match in aggregation_result.rows
         ]
+        matches = [match for match in processed if match is not None]
+        if len(matches) != len(processed):
+            logger.warning(
+                "Dropped %d route match(es) with missing field data (likely "
+                "expired during a background search on Redis 8.8+).",
+                len(processed) - len(matches),
+            )
+        return matches
 
     def _classify_route(
         self,

redisvl/index/__init__.py

@@ -1,3 +1,3 @@
-from redisvl.index.index import AsyncSearchIndex, SearchIndex
+from redisvl.index.index import AsyncSearchIndex, SearchIndex, SearchResults
 
-__all__ = ["SearchIndex", "AsyncSearchIndex"]
+__all__ = ["SearchIndex", "AsyncSearchIndex", "SearchResults"]