From b3e808ff512c4b76e972f3c41b5d997652162766 Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Thu, 6 Aug 2026 18:27:28 +0200 Subject: [PATCH 1/5] fix: keep paginating past a page whose matches were all dropped `SearchIndex.paginate` and `AsyncSearchIndex.paginate` terminated on `if not results: break`, treating an empty page as an exhausted result set. That is unsound: `process_results` deliberately drops matched documents whose field payload came back missing (the Redis 8.8+ background-WORKERS TTL/expiry race), so a page can be empty while the server still has matches to report. When every document on one page was dropped, iteration stopped early and silently discarded every remaining page. Termination now asks whether the server reported any match for the page, using the `dropped_count` that `SearchResults` already carries, rather than whether any document could be materialized. A page with matches but no materialized documents is skipped and the offset still advances, so iteration always makes progress and cannot wedge. Fully-dropped pages are not yielded, so every yielded batch stays non-empty as before. Regression tests drive the termination logic with canned pages (the live 8.8 race is not reproducible on demand); 3 of them fail before this change. --- redisvl/index/index.py | 48 +++++- tests/unit/test_paginate_dropped_page.py | 189 +++++++++++++++++++++++ 2 files changed, 232 insertions(+), 5 deletions(-) create mode 100644 tests/unit/test_paginate_dropped_page.py diff --git a/redisvl/index/index.py b/redisvl/index/index.py index 1c690d47..0c7b180b 100644 --- a/redisvl/index/index.py +++ b/redisvl/index/index.py @@ -515,6 +515,25 @@ def _convert_and_drop_empty_rows(rows: Any, source: str) -> list[dict[str, Any]] return SearchResults(kept, dropped_count=dropped) +def _page_had_matches(results: Any) -> bool: + """Whether the server reported any match for one page of paginated results. + + ``paginate`` must not treat "empty page" as "result set exhausted". A page + comes back empty for two very different reasons: + + 1. the server reported no more matches — iteration is genuinely done; + 2. the server reported matches whose field payload could not be + materialized, so ``process_results`` dropped them (the Redis 8.8+ + background-WORKERS TTL/expiry race, see ``_has_missing_field_payload``). + + Stopping on case 2 silently truncates iteration and discards every remaining + page. ``SearchResults.dropped_count`` tells the two apart: a page had matches + when it either yielded documents or dropped some. ``getattr`` keeps this + tolerant of query paths that return a plain ``list`` without the metadata. + """ + return bool(len(results) or getattr(results, "dropped_count", 0)) + + class BaseSearchIndex: """Base search engine class""" @@ -1935,6 +1954,12 @@ def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator: Note: For stable pagination, the query must have a `sort_by` clause. + Note: + Iteration stops when the server reports no further matches, not when + a page yields no documents. A page whose matches were all dropped by + ``process_results`` (the Redis 8.8+ background-search expiry race) is + skipped rather than treated as the end of the result set. Such a page + is not yielded, so every yielded batch is non-empty as before. """ if not isinstance(page_size, int): raise TypeError("page_size must be an integer") @@ -1946,10 +1971,13 @@ def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator: while True: query.paging(offset, page_size) results = self._query(query) - if not results: + if not _page_had_matches(results): break - yield results - # Increment the offset for the next batch of pagination + if results: + yield results + # Increment the offset for the next batch of pagination. This happens + # unconditionally -- including when every match on this page was + # dropped -- so a page we cannot materialize can never wedge the loop. offset += page_size def listall(self) -> list[str]: @@ -3154,6 +3182,12 @@ async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerato Note: For stable pagination, the query must have a `sort_by` clause. + Note: + Iteration stops when the server reports no further matches, not when + a page yields no documents. A page whose matches were all dropped by + ``process_results`` (the Redis 8.8+ background-search expiry race) is + skipped rather than treated as the end of the result set. Such a page + is not yielded, so every yielded batch is non-empty as before. """ if not isinstance(page_size, int): raise TypeError("page_size must be of type int") @@ -3165,9 +3199,13 @@ async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerato while True: query.paging(first, page_size) results = await self._query(query) - if not results: + if not _page_had_matches(results): break - yield results + if results: + yield results + # Advance unconditionally -- including when every match on this page + # was dropped -- so a page we cannot materialize can never wedge the + # loop. first += page_size async def listall(self) -> list[str]: diff --git a/tests/unit/test_paginate_dropped_page.py b/tests/unit/test_paginate_dropped_page.py new file mode 100644 index 00000000..9770e026 --- /dev/null +++ b/tests/unit/test_paginate_dropped_page.py @@ -0,0 +1,189 @@ +"""Regression tests: ``paginate`` must not stop on a page whose matches were dropped. + +``process_results`` deliberately drops matched documents whose field payload came +back missing (the Redis 8.8+ background-WORKERS TTL/expiry race), so a page can +be empty while the server still has matches to report. ``paginate`` previously +terminated on ``if not results: break``, conflating "no more matches" with +"matches we could not materialize" and silently discarding every remaining page. + +The live 8.8 race is not reproducible on demand, so these tests drive the +termination logic directly with canned pages, mirroring the canned-page style +used elsewhere for bulk cursor tests. +""" + +import pytest + +from redisvl.index import AsyncSearchIndex, SearchIndex +from redisvl.index.index import SearchResults +from redisvl.query import VectorQuery +from redisvl.schema import IndexSchema + +sample_vector = [0.1, 0.1, 0.5, 0.15] + + +def _schema(): + return IndexSchema.from_dict( + { + "index": { + "name": "paginate_test", + "prefix": "test", + "storage_type": "hash", + }, + "fields": [ + { + "name": "user_embedding", + "type": "vector", + "attrs": { + "dims": 4, + "distance_metric": "cosine", + "algorithm": "flat", + "datatype": "float32", + }, + }, + {"name": "brand", "type": "tag"}, + ], + } + ) + + +def _query(): + return VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + return_fields=["brand"], + ) + + +def _doc(doc_id): + return {"id": doc_id, "vector_distance": "0.1", "brand": "Nike"} + + +def _canned_pages(): + """Three pages of matches where the whole middle page was dropped. + + Page 2 is empty but reports ``dropped_count=2``: the server matched two docs + and neither could be materialized. Page 4 is the genuine end of the result + set — empty with nothing dropped. + """ + return [ + SearchResults([_doc("doc:1"), _doc("doc:2")]), + SearchResults([], dropped_count=2), + SearchResults([_doc("doc:5"), _doc("doc:6")]), + SearchResults([]), + ] + + +def _install_canned_query(index, pages, offsets): + """Replace the index's query execution with a canned page sequence. + + Records the paging offset the query carried on each call so tests can assert + the offset keeps advancing across a dropped page. + """ + responses = list(pages) + + def _fake_query(query): + offsets.append(query._offset) + return responses.pop(0) if responses else SearchResults([]) + + index._query = _fake_query # type: ignore[method-assign] + + +def test_paginate_continues_past_fully_dropped_page(): + """A page whose matches were all dropped must not end iteration.""" + index = SearchIndex(_schema()) + offsets: list[int] = [] + _install_canned_query(index, _canned_pages(), offsets) + + batches = list(index.paginate(_query(), page_size=2)) + + ids = [doc["id"] for batch in batches for doc in batch] + assert ids == ["doc:1", "doc:2", "doc:5", "doc:6"] + # The dropped page is skipped, not yielded: every batch stays non-empty. + assert all(batch for batch in batches) + # Offset advances across the dropped page so iteration makes progress. + assert offsets == [0, 2, 4, 6] + + +def test_paginate_stops_when_server_reports_no_matches(): + """An empty page with nothing dropped is still the end of the result set.""" + index = SearchIndex(_schema()) + offsets: list[int] = [] + _install_canned_query( + index, + [SearchResults([_doc("doc:1")]), SearchResults([])], + offsets, + ) + + batches = list(index.paginate(_query(), page_size=1)) + + assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] + # Stopped on the empty page -- no third request. + assert offsets == [0, 1] + + +def test_paginate_terminates_when_every_page_is_dropped(): + """All-dropped pages must terminate once the server runs out of matches.""" + index = SearchIndex(_schema()) + offsets: list[int] = [] + _install_canned_query( + index, + [ + SearchResults([], dropped_count=2), + SearchResults([], dropped_count=2), + SearchResults([]), + ], + offsets, + ) + + assert list(index.paginate(_query(), page_size=2)) == [] + assert offsets == [0, 2, 4] + + +def test_paginate_tolerates_plain_list_pages(): + """A query path returning a plain ``list`` (no metadata) behaves as before.""" + index = SearchIndex(_schema()) + offsets: list[int] = [] + _install_canned_query(index, [[_doc("doc:1")], []], offsets) + + batches = list(index.paginate(_query(), page_size=1)) + + assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] + + +@pytest.mark.asyncio +async def test_async_paginate_continues_past_fully_dropped_page(): + """Async mirror: the dropped middle page must not truncate iteration.""" + index = AsyncSearchIndex(_schema()) + offsets: list[int] = [] + responses = _canned_pages() + + async def _fake_query(query): + offsets.append(query._offset) + return responses.pop(0) if responses else SearchResults([]) + + index._query = _fake_query # type: ignore[method-assign] + + batches = [batch async for batch in index.paginate(_query(), page_size=2)] + + ids = [doc["id"] for batch in batches for doc in batch] + assert ids == ["doc:1", "doc:2", "doc:5", "doc:6"] + assert all(batch for batch in batches) + assert offsets == [0, 2, 4, 6] + + +@pytest.mark.asyncio +async def test_async_paginate_stops_when_server_reports_no_matches(): + index = AsyncSearchIndex(_schema()) + offsets: list[int] = [] + responses = [SearchResults([_doc("doc:1")]), SearchResults([])] + + async def _fake_query(query): + offsets.append(query._offset) + return responses.pop(0) if responses else SearchResults([]) + + index._query = _fake_query # type: ignore[method-assign] + + batches = [batch async for batch in index.paginate(_query(), page_size=1)] + + assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] + assert offsets == [0, 1] From a80dd01a704c99e256eb583c04374aa915e49fa9 Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Thu, 6 Aug 2026 19:00:38 +0200 Subject: [PATCH 2/5] fix: address review findings on the paginate dropped-page fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five review passes (code, security, system design, documentation, testing) over b3e808f. Three real defects and a set of gaps: **`_page_had_matches` used `len()` where the old code used truthiness.** `process_results` returns a bare `int` for a `CountQuery`, so `paginate` on one went from an infinite loop (total > 0, the old code yielded the same integer forever) to `TypeError: object of type 'int' has no len()` — and for total == 0 that was a strict regression, since the old code exited cleanly. Switched to truthiness so `int`/`None`/`list` never raise, and added an explicit `CountQuery` guard with a message pointing at `index.query()`, which fixes the pre-existing infinite loop rather than trading it for an obscure `TypeError`. **A fully-dropped page took its `dropped_count` out of the stream.** A page with 1 kept and 9 dropped was yielded reporting `dropped_count=9`; a page with 0 kept and 10 dropped was invisible, leaving every batch the caller saw reporting `complete is True` while matched documents went missing. Same event, opposite observability, decided by whether one document happened to survive. A skipped page's count is now folded into the next yielded batch (`_fold_carried_drops`), which keeps the signal reachable without breaking the non-empty-batch guarantee. Trailing dropped pages have no later batch to ride on and remain warning-only; that residual gap is documented at the helper and in the concepts guide. **The helper docstring justified its `getattr` with a path that does not exist.** No query path returns a plain `list` — every one returns `SearchResults`. Reworded to say what the fallback actually guards: an override or test double. Tests re-seated on the `index.search` seam, so canned `FT.SEARCH` replies run the real `_query` -> `process_results` -> `SearchResults` chain and the `dropped_count` the fix depends on is produced by production code instead of hand-constructed. Pages are keyed by paging offset, so a failure to advance shows up as repeated documents or a wedge rather than a silently different page, and the request count is bounded so a regression fails instead of hanging the suite. Added the cases reviewers found missing: partially-dropped page (yielded, `SearchResults`, count intact), carry-forward, trailing dropped page, zero matches, `page_size` validation, `CountQuery` rejection, async mirrors, and direct contract tests for both helpers. Verified by reintroducing each of the three regressions in turn. Docs: the `paginate` notes were changelog-voice and named internal machinery that appears nowhere in the published docs; rewritten at caller altitude to state the actionable facts. `SearchResults` was a public export with no API-reference entry even though this fix's correctness rests on it — added. Pagination's guarantee added to the concepts guide, whose existing text promised `results.complete` detects short paginated pages. Deferred deliberately: short-page termination (`len + dropped < page_size`), a public `SearchResults.total`, and a separately-filed pre-existing unbounded loop in `clear()`/`drop_by_filter()` where a non-empty batch that deletes nothing re-queries offset 0 forever. --- docs/api/searchindex.rst | 12 + docs/concepts/queries.md | 1 + redisvl/index/index.py | 92 +++++-- tests/unit/test_paginate_dropped_page.py | 329 ++++++++++++++++++----- 4 files changed, 348 insertions(+), 86 deletions(-) diff --git a/docs/api/searchindex.rst b/docs/api/searchindex.rst index 6080d274..826becf3 100644 --- a/docs/api/searchindex.rst +++ b/docs/api/searchindex.rst @@ -12,6 +12,8 @@ Search Index Classes - Primary class to write, read, and search across data structures in Redis. * - :ref:`asyncsearchindex_api` - Async version of the SearchIndex to write, read, and search across data structures in Redis. + * - :ref:`searchresults_api` + - List of result documents returned by a query, which also reports result completeness. .. _searchindex_api: @@ -34,3 +36,13 @@ AsyncSearchIndex .. autoclass:: AsyncSearchIndex :inherited-members: :members: + +.. _searchresults_api: + +SearchResults +============= + +.. currentmodule:: redisvl.index + +.. autoclass:: SearchResults + :members: diff --git a/docs/concepts/queries.md b/docs/concepts/queries.md index fb080c44..06bdda55 100644 --- a/docs/concepts/queries.md +++ b/docs/concepts/queries.md @@ -382,6 +382,7 @@ 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. +- **`paginate()` still reaches the end of the result set.** If every match on one page was expiring, that page is skipped and iteration continues rather than stopping early. An empty batch is never yielded, and a skipped page's `dropped_count` is added to the next batch you receive, so `results.complete` stays meaningful across the whole stream. The exception is a result set whose *trailing* pages were entirely dropped — there is no later batch to report those on, and they appear only in the logged warning. - 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. diff --git a/redisvl/index/index.py b/redisvl/index/index.py index 0c7b180b..01660d41 100644 --- a/redisvl/index/index.py +++ b/redisvl/index/index.py @@ -528,10 +528,32 @@ def _page_had_matches(results: Any) -> bool: Stopping on case 2 silently truncates iteration and discards every remaining page. ``SearchResults.dropped_count`` tells the two apart: a page had matches - when it either yielded documents or dropped some. ``getattr`` keeps this - tolerant of query paths that return a plain ``list`` without the metadata. + when it either yielded documents or dropped some. + + Truthiness (not ``len()``) is deliberate, and ``getattr`` is defensive: every + query path returns ``SearchResults`` today, but a subclass or test double may + substitute a plain ``list``, and ``process_results`` returns a bare ``int`` + for a ``CountQuery``. Neither should raise here. + """ + return bool(results) or bool(getattr(results, "dropped_count", 0)) + + +def _fold_carried_drops(results: Any, carried: int) -> None: + """Add drops carried over from skipped pages to this page's ``dropped_count``. + + ``paginate`` never yields an empty batch, so a page whose matches were *all* + dropped would otherwise carry its ``dropped_count`` out of the stream and + leave the remaining batches reporting ``complete is True`` — the same silent + incompleteness the drop accounting exists to surface. Folding the count into + the next yielded batch keeps the signal reachable from the batches a caller + actually sees, without breaking the non-empty-batch guarantee. + + Note the residual gap: if the *trailing* pages of a result set are entirely + dropped there is no subsequent batch to fold into, and those drops are + reported only by the ``process_results`` warning. """ - return bool(len(results) or getattr(results, "dropped_count", 0)) + if carried and isinstance(results, SearchResults): + results.dropped_count += carried class BaseSearchIndex: @@ -1933,7 +1955,7 @@ def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator: batch. Defaults to 30. Yields: - A generator yielding batches of search results. + A generator yielding non-empty ``SearchResults`` batches. Raises: TypeError: If the page_size argument is not of type int. @@ -1955,11 +1977,13 @@ def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator: For stable pagination, the query must have a `sort_by` clause. Note: - Iteration stops when the server reports no further matches, not when - a page yields no documents. A page whose matches were all dropped by - ``process_results`` (the Redis 8.8+ background-search expiry race) is - skipped rather than treated as the end of the result set. Such a page - is not yielded, so every yielded batch is non-empty as before. + A yielded batch may contain fewer than ``page_size`` documents, and an + empty batch is never yielded. On Redis 8+ a matched document whose + data expires while the search is running is skipped; each batch + reports how many were skipped via ``dropped_count`` (and + ``complete``), including skips inherited from a page that was dropped + in its entirety. Pagination still runs to the end of the result set in + that case. """ if not isinstance(page_size, int): raise TypeError("page_size must be an integer") @@ -1967,17 +1991,29 @@ def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator: if page_size <= 0: raise ValueError("page_size must be greater than 0") + if isinstance(query, CountQuery): + raise TypeError( + "CountQuery cannot be paginated: it returns a match count rather " + "than documents. Use index.query(query) instead." + ) + offset = 0 + carried_drops = 0 while True: query.paging(offset, page_size) results = self._query(query) if not _page_had_matches(results): break if results: + _fold_carried_drops(results, carried_drops) + carried_drops = 0 yield results - # Increment the offset for the next batch of pagination. This happens - # unconditionally -- including when every match on this page was - # dropped -- so a page we cannot materialize can never wedge the loop. + else: + # Whole page dropped: keep its count so it reaches the caller on + # the next yielded batch instead of vanishing with the page. + carried_drops += getattr(results, "dropped_count", 0) + # Advance unconditionally, so a page we cannot materialize can never + # wedge the loop. offset += page_size def listall(self) -> list[str]: @@ -3161,7 +3197,7 @@ async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerato batch. Defaults to 30. Yields: - An async generator yielding batches of search results. + An async generator yielding non-empty ``SearchResults`` batches. Raises: TypeError: If the page_size argument is not of type int. @@ -3183,11 +3219,13 @@ async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerato For stable pagination, the query must have a `sort_by` clause. Note: - Iteration stops when the server reports no further matches, not when - a page yields no documents. A page whose matches were all dropped by - ``process_results`` (the Redis 8.8+ background-search expiry race) is - skipped rather than treated as the end of the result set. Such a page - is not yielded, so every yielded batch is non-empty as before. + A yielded batch may contain fewer than ``page_size`` documents, and an + empty batch is never yielded. On Redis 8+ a matched document whose + data expires while the search is running is skipped; each batch + reports how many were skipped via ``dropped_count`` (and + ``complete``), including skips inherited from a page that was dropped + in its entirety. Pagination still runs to the end of the result set in + that case. """ if not isinstance(page_size, int): raise TypeError("page_size must be of type int") @@ -3195,17 +3233,29 @@ async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerato if page_size <= 0: raise ValueError("page_size must be greater than 0") + if isinstance(query, CountQuery): + raise TypeError( + "CountQuery cannot be paginated: it returns a match count rather " + "than documents. Use index.query(query) instead." + ) + first = 0 + carried_drops = 0 while True: query.paging(first, page_size) results = await self._query(query) if not _page_had_matches(results): break if results: + _fold_carried_drops(results, carried_drops) + carried_drops = 0 yield results - # Advance unconditionally -- including when every match on this page - # was dropped -- so a page we cannot materialize can never wedge the - # loop. + else: + # Whole page dropped: keep its count so it reaches the caller on + # the next yielded batch instead of vanishing with the page. + carried_drops += getattr(results, "dropped_count", 0) + # Advance unconditionally, so a page we cannot materialize can never + # wedge the loop. first += page_size async def listall(self) -> list[str]: diff --git a/tests/unit/test_paginate_dropped_page.py b/tests/unit/test_paginate_dropped_page.py index 9770e026..32fc2a5a 100644 --- a/tests/unit/test_paginate_dropped_page.py +++ b/tests/unit/test_paginate_dropped_page.py @@ -6,16 +6,19 @@ terminated on ``if not results: break``, conflating "no more matches" with "matches we could not materialize" and silently discarding every remaining page. -The live 8.8 race is not reproducible on demand, so these tests drive the -termination logic directly with canned pages, mirroring the canned-page style -used elsewhere for bulk cursor tests. +The live race is not reproducible on demand, so these tests serve canned +``FT.SEARCH`` replies from a fake ``index.search``, keyed by the paging offset the +query carried. That seam runs the real ``_query`` -> ``process_results`` -> +``SearchResults`` chain, so the ``dropped_count`` the fix depends on is produced +by production code rather than hand-constructed. """ import pytest +from redis.commands.search.result import Result -from redisvl.index import AsyncSearchIndex, SearchIndex -from redisvl.index.index import SearchResults -from redisvl.query import VectorQuery +from redisvl.index import AsyncSearchIndex, SearchIndex, SearchResults +from redisvl.index.index import _fold_carried_drops, _page_had_matches +from redisvl.query import CountQuery, VectorQuery from redisvl.schema import IndexSchema sample_vector = [0.1, 0.1, 0.5, 0.15] @@ -54,45 +57,63 @@ def _query(): ) -def _doc(doc_id): - return {"id": doc_id, "vector_distance": "0.1", "brand": "Nike"} +def _healthy(doc_id): + """A raw FT.SEARCH reply fragment for a match that carries its fields.""" + return [doc_id, ["vector_distance", "0.1", "brand", "Nike"]] -def _canned_pages(): - """Three pages of matches where the whole middle page was dropped. +def _race_victim(doc_id): + """A matched id whose field array came back nil -- the expiry race victim. - Page 2 is empty but reports ``dropped_count=2``: the server matched two docs - and neither could be materialized. Page 4 is the genuine end of the result - set — empty with nothing dropped. + redis-py collapses the nil to an empty field set, leaving a ``Document`` whose + ``__dict__`` is only ``{"id": ..., "payload": None}``. For a vector query + (which always projects ``vector_distance``) ``process_results`` detects the + missing payload and drops the doc, incrementing ``dropped_count``. """ - return [ - SearchResults([_doc("doc:1"), _doc("doc:2")]), - SearchResults([], dropped_count=2), - SearchResults([_doc("doc:5"), _doc("doc:6")]), - SearchResults([]), - ] + return [doc_id, None] -def _install_canned_query(index, pages, offsets): - """Replace the index's query execution with a canned page sequence. +def _page(*fragments): + flat = [item for fragment in fragments for item in fragment] + return Result([len(fragments), *flat], True) - Records the paging offset the query carried on each call so tests can assert - the offset keeps advancing across a dropped page. + +EMPTY_PAGE = Result([0], True) + + +def _install_pages(index, pages, seen): + """Serve canned FT.SEARCH replies keyed by the query's paging offset. + + Keying by offset (rather than popping a queue) means a failure to advance the + offset shows up as repeated documents or a wedge, not as a silently different + page. ``seen`` records the request order and bounds runaway loops. It reads + redis-py's private ``Query._offset``, which ``paging()`` sets. """ - responses = list(pages) - def _fake_query(query): - offsets.append(query._offset) - return responses.pop(0) if responses else SearchResults([]) + def _fake_search(query, query_params=None): + seen.append(query._offset) + assert len(seen) <= 20, f"paginate made {len(seen)} requests; likely wedged" + return pages.get(query._offset, EMPTY_PAGE) + + index.search = _fake_search # type: ignore[method-assign] - index._query = _fake_query # type: ignore[method-assign] + +# Three pages of matches where the whole middle page was dropped: the server +# matched doc:3 and doc:4 at offset 2 and neither could be materialized. Offset 6 +# is the genuine end of the result set. +DROPPED_MIDDLE_PAGES = { + 0: _page(_healthy("doc:1"), _healthy("doc:2")), + 2: _page(_race_victim("doc:3"), _race_victim("doc:4")), + 4: _page(_healthy("doc:5"), _healthy("doc:6")), + 6: EMPTY_PAGE, +} def test_paginate_continues_past_fully_dropped_page(): """A page whose matches were all dropped must not end iteration.""" index = SearchIndex(_schema()) - offsets: list[int] = [] - _install_canned_query(index, _canned_pages(), offsets) + seen: list[int] = [] + _install_pages(index, DROPPED_MIDDLE_PAGES, seen) batches = list(index.paginate(_query(), page_size=2)) @@ -100,90 +121,268 @@ def test_paginate_continues_past_fully_dropped_page(): assert ids == ["doc:1", "doc:2", "doc:5", "doc:6"] # The dropped page is skipped, not yielded: every batch stays non-empty. assert all(batch for batch in batches) - # Offset advances across the dropped page so iteration makes progress. - assert offsets == [0, 2, 4, 6] + # Offset advanced across the dropped page so iteration made progress. + assert seen == [0, 2, 4, 6] -def test_paginate_stops_when_server_reports_no_matches(): - """An empty page with nothing dropped is still the end of the result set.""" +def test_paginate_carries_dropped_count_of_skipped_page_forward(): + """A skipped page's drops must surface on the next yielded batch. + + Otherwise the batches the caller actually sees all report ``complete is + True`` while matched documents went missing from the stream. + """ index = SearchIndex(_schema()) - offsets: list[int] = [] - _install_canned_query( + _install_pages(index, DROPPED_MIDDLE_PAGES, []) + + batches = list(index.paginate(_query(), page_size=2)) + + assert batches[0].dropped_count == 0 + assert batches[0].complete is True + # doc:3 and doc:4 were dropped with their page; the count rides along here. + assert batches[1].dropped_count == 2 + assert batches[1].complete is False + + +def test_paginate_yields_partially_dropped_page_with_its_count(): + """A page with survivors AND drops is yielded, carrying its own count.""" + index = SearchIndex(_schema()) + _install_pages( index, - [SearchResults([_doc("doc:1")]), SearchResults([])], - offsets, + { + 0: _page(_healthy("doc:1"), _race_victim("doc:2")), + 2: EMPTY_PAGE, + }, + [], ) + batches = list(index.paginate(_query(), page_size=2)) + + assert len(batches) == 1 + assert [doc["id"] for doc in batches[0]] == ["doc:1"] + # Batches must remain SearchResults, or the completeness contract is severed. + assert isinstance(batches[0], SearchResults) + assert batches[0].dropped_count == 1 + assert batches[0].complete is False + + +def test_paginate_stops_when_server_reports_no_matches(): + """An empty page with nothing dropped is still the end of the result set.""" + index = SearchIndex(_schema()) + seen: list[int] = [] + _install_pages(index, {0: _page(_healthy("doc:1")), 1: EMPTY_PAGE}, seen) + batches = list(index.paginate(_query(), page_size=1)) assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] # Stopped on the empty page -- no third request. - assert offsets == [0, 1] + assert seen == [0, 1] + + +def test_paginate_zero_matches_yields_nothing(): + """An empty first page terminates immediately rather than looping.""" + index = SearchIndex(_schema()) + seen: list[int] = [] + _install_pages(index, {}, seen) + + assert list(index.paginate(_query(), page_size=10)) == [] + assert seen == [0] + + +def test_paginate_terminates_on_trailing_dropped_page(): + """A fully-dropped final page costs one extra request, then terminates.""" + index = SearchIndex(_schema()) + seen: list[int] = [] + _install_pages( + index, + { + 0: _page(_healthy("doc:1"), _healthy("doc:2")), + 2: _page(_race_victim("doc:3")), + 4: EMPTY_PAGE, + }, + seen, + ) + + batches = list(index.paginate(_query(), page_size=2)) + + assert [doc["id"] for batch in batches for doc in batch] == ["doc:1", "doc:2"] + assert seen == [0, 2, 4] + # Trailing drops have no later batch to ride on; process_results logs them. + assert batches[0].dropped_count == 0 def test_paginate_terminates_when_every_page_is_dropped(): """All-dropped pages must terminate once the server runs out of matches.""" index = SearchIndex(_schema()) - offsets: list[int] = [] - _install_canned_query( + seen: list[int] = [] + _install_pages( index, - [ - SearchResults([], dropped_count=2), - SearchResults([], dropped_count=2), - SearchResults([]), - ], - offsets, + { + 0: _page(_race_victim("doc:1"), _race_victim("doc:2")), + 2: _page(_race_victim("doc:3"), _race_victim("doc:4")), + 4: EMPTY_PAGE, + }, + seen, ) assert list(index.paginate(_query(), page_size=2)) == [] - assert offsets == [0, 2, 4] + assert seen == [0, 2, 4] + + +def test_paginate_rejects_count_query(): + """CountQuery returns a match count, not documents, so it cannot paginate. + + ``process_results`` returns a bare ``int`` for it, which used to make + ``paginate`` yield that integer forever. + """ + index = SearchIndex(_schema()) + with pytest.raises(TypeError, match="CountQuery cannot be paginated"): + list(index.paginate(CountQuery("*"), page_size=2)) + + +def test_paginate_validates_page_size(): + index = SearchIndex(_schema()) + with pytest.raises(TypeError, match="page_size must be an integer"): + list(index.paginate(_query(), page_size="2")) # type: ignore[arg-type] + with pytest.raises(ValueError, match="page_size must be greater than 0"): + list(index.paginate(_query(), page_size=0)) def test_paginate_tolerates_plain_list_pages(): - """A query path returning a plain ``list`` (no metadata) behaves as before.""" + """``paginate`` still works if ``_query`` is overridden to return a plain list. + + No production query path does this -- every one returns ``SearchResults`` -- + but the termination check must not assume the metadata is present. + """ index = SearchIndex(_schema()) - offsets: list[int] = [] - _install_canned_query(index, [[_doc("doc:1")], []], offsets) + responses: list[list[dict]] = [[{"id": "doc:1"}], []] + index._query = lambda query: responses.pop(0) if responses else [] # type: ignore[method-assign] batches = list(index.paginate(_query(), page_size=1)) assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] +# --------------------------------------------------------------------------- +# Async mirrors. AsyncSearchIndex.paginate is a hand-duplicated copy of the sync +# logic, so it is the natural place for the two to drift. +# --------------------------------------------------------------------------- + + +def _install_async_pages(index, pages, seen): + async def _fake_search(query, query_params=None): + seen.append(query._offset) + assert len(seen) <= 20, f"paginate made {len(seen)} requests; likely wedged" + return pages.get(query._offset, EMPTY_PAGE) + + index.search = _fake_search # type: ignore[method-assign] + + @pytest.mark.asyncio async def test_async_paginate_continues_past_fully_dropped_page(): """Async mirror: the dropped middle page must not truncate iteration.""" index = AsyncSearchIndex(_schema()) - offsets: list[int] = [] - responses = _canned_pages() - - async def _fake_query(query): - offsets.append(query._offset) - return responses.pop(0) if responses else SearchResults([]) - - index._query = _fake_query # type: ignore[method-assign] + seen: list[int] = [] + _install_async_pages(index, DROPPED_MIDDLE_PAGES, seen) batches = [batch async for batch in index.paginate(_query(), page_size=2)] ids = [doc["id"] for batch in batches for doc in batch] assert ids == ["doc:1", "doc:2", "doc:5", "doc:6"] assert all(batch for batch in batches) - assert offsets == [0, 2, 4, 6] + assert seen == [0, 2, 4, 6] + # Skipped page's drops carried onto the following batch. + assert batches[1].dropped_count == 2 @pytest.mark.asyncio async def test_async_paginate_stops_when_server_reports_no_matches(): index = AsyncSearchIndex(_schema()) - offsets: list[int] = [] - responses = [SearchResults([_doc("doc:1")]), SearchResults([])] + seen: list[int] = [] + _install_async_pages(index, {0: _page(_healthy("doc:1")), 1: EMPTY_PAGE}, seen) + + batches = [batch async for batch in index.paginate(_query(), page_size=1)] + + assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] + assert seen == [0, 1] + + +@pytest.mark.asyncio +async def test_async_paginate_terminates_when_every_page_is_dropped(): + index = AsyncSearchIndex(_schema()) + seen: list[int] = [] + _install_async_pages( + index, + { + 0: _page(_race_victim("doc:1"), _race_victim("doc:2")), + 2: EMPTY_PAGE, + }, + seen, + ) + + assert [batch async for batch in index.paginate(_query(), page_size=2)] == [] + assert seen == [0, 2] + + +@pytest.mark.asyncio +async def test_async_paginate_rejects_count_query(): + index = AsyncSearchIndex(_schema()) + with pytest.raises(TypeError, match="CountQuery cannot be paginated"): + [batch async for batch in index.paginate(CountQuery("*"), page_size=2)] + + +@pytest.mark.asyncio +async def test_async_paginate_tolerates_plain_list_pages(): + index = AsyncSearchIndex(_schema()) + responses: list[list[dict]] = [[{"id": "doc:1"}], []] async def _fake_query(query): - offsets.append(query._offset) - return responses.pop(0) if responses else SearchResults([]) + return responses.pop(0) if responses else [] index._query = _fake_query # type: ignore[method-assign] batches = [batch async for batch in index.paginate(_query(), page_size=1)] assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] - assert offsets == [0, 1] + + +# --------------------------------------------------------------------------- +# Helper contracts, pinned directly so the intent survives a refactor. +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "results, expected", + [ + (SearchResults([]), False), + (SearchResults([{"id": "doc:1"}]), True), + (SearchResults([], dropped_count=1), True), + (SearchResults([{"id": "doc:1"}], dropped_count=1), True), + ([], False), + ([{"id": "doc:1"}], True), + # process_results returns a bare int for CountQuery; must not raise. + (0, False), + (5, True), + (None, False), + ], +) +def test_page_had_matches(results, expected): + assert _page_had_matches(results) is expected + + +def test_fold_carried_drops_adds_to_existing_count(): + results = SearchResults([{"id": "doc:1"}], dropped_count=1) + _fold_carried_drops(results, 2) + assert results.dropped_count == 3 + + +def test_fold_carried_drops_noop_without_carry(): + results = SearchResults([{"id": "doc:1"}], dropped_count=1) + _fold_carried_drops(results, 0) + assert results.dropped_count == 1 + + +def test_fold_carried_drops_ignores_plain_list(): + """A plain list carries no metadata; folding must not raise.""" + plain = [{"id": "doc:1"}] + _fold_carried_drops(plain, 2) + assert plain == [{"id": "doc:1"}] From 034caf89a71df045e2c27501c03e2917b0bb165d Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Fri, 7 Aug 2026 10:21:34 +0200 Subject: [PATCH 3/5] fix: stop NOCONTENT queries reporting every document as missing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Redis-docs review of the paginate change surfaced a false positive in `_has_missing_field_payload` that this branch made materially worse. FT.SEARCH with NOCONTENT "returns the document ids and not the content" for every healthy match (`RETURN 0` behaves the same way), so no field payload is expected and its absence carries no information. redis-py's `Query.no_content()` sets only `_no_content` and leaves `_return_fields` untouched, so both detection branches fire on every document: `unpack_json` stays true because `_return_fields` is empty, and `vector_distance` stays in `_return_fields` while the server sends ids only. Verified against production code: a 3-match NOCONTENT reply came back as zero documents with `dropped_count=3`, on both the JSON-unpack and vector paths. Before this branch that surfaced as one empty page and iteration stopped. After it, `_page_had_matches` is true on every page, so `paginate` walked the entire result set yielding nothing and logging a warning per page — and deep offsets can now reach the server's `search-max-search-results` cap and error instead of ending. `index.query()` was already silently returning `[]` for these queries. Short-circuits the predicate on `_no_content`, and guards the vector-normalize branch on the distance actually being present — without that, skipping the drop lets `VectorQuery(normalize_vector_distance=True).no_content()` reach `doc_dict[DISTANCE_ID]` and raise `KeyError`. Race detection and normalization of healthy documents are both unchanged (verified). Docs corrected against the FT.SEARCH reference: - The stable-pagination note asked only for a `sort_by` clause. The docs require `SORTBY` on a *unique* field — "If you use the LIMIT option without sorting, the results returned are non-deterministic, which means that subsequent queries may return duplicated or missing values." Also notes the `search-max-search-results` ceiling (1,000,000 default, 10,000 on some managed tiers), which bounds deep pagination. - The concepts bullet added in a80dd01 claimed `paginate()` "still reaches the end of the result set". True of offsets, not of documents: without a unique sort key pages can repeat or miss rows regardless of expiry. Reworded and qualified. - Widened the race framing in the text this branch introduced: the trigger is a key that expires *or is updated* mid-query, not TTL expiry alone, and such a key is still counted in the server's total. Not changed: the "Redis 8.8+ background-WORKERS" attribution in the seven docstring sites inherited from 0cfabe2. The review argued `search-workers` defaults to 0, but cited the Redis Software REST API object reference rather than the OSS configuration page, and this repo pins `--search-workers 0` in tests/docker-compose.yml precisely because 8.8 changed that default and broke `redis:latest` CI. Left alone pending evidence from the OSS docs. --- docs/concepts/queries.md | 2 +- redisvl/index/index.py | 34 +++++++++++++++++++++++++----- tests/unit/test_query_types.py | 38 ++++++++++++++++++++++++++++++++++ 3 files changed, 68 insertions(+), 6 deletions(-) diff --git a/docs/concepts/queries.md b/docs/concepts/queries.md index 06bdda55..9f635424 100644 --- a/docs/concepts/queries.md +++ b/docs/concepts/queries.md @@ -382,7 +382,7 @@ 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. -- **`paginate()` still reaches the end of the result set.** If every match on one page was expiring, that page is skipped and iteration continues rather than stopping early. An empty batch is never yielded, and a skipped page's `dropped_count` is added to the next batch you receive, so `results.complete` stays meaningful across the whole stream. The exception is a result set whose *trailing* pages were entirely dropped — there is no later batch to report those on, and they appear only in the logged warning. +- **`paginate()` keeps going past a page it could not materialize.** If every match on one page was expiring, that page is skipped and iteration continues to the end of the offset range rather than stopping early. An empty batch is never yielded, and a skipped page's `dropped_count` is added to the next batch you receive, so `results.complete` stays meaningful across the whole stream. Two caveats: a result set whose *trailing* pages were entirely dropped has no later batch to report on, so those drops appear only in the logged warning; and paging is only stable if the query sorts on a unique field — Redis documents `LIMIT` without sorting as non-deterministic, which can duplicate or miss documents independently of expiry. - 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. diff --git a/redisvl/index/index.py b/redisvl/index/index.py index 01660d41..171d76ad 100644 --- a/redisvl/index/index.py +++ b/redisvl/index/index.py @@ -330,7 +330,16 @@ def _has_missing_field_payload( match. Every other query passes through untouched. + + A ``NOCONTENT`` query breaks both guarantees: the server then returns "the + document ids and not the content" for *every* healthy match (``RETURN 0`` acts + the same way), so no field payload is expected and its absence says nothing. + redis-py's ``Query.no_content()`` sets only ``_no_content`` and leaves + ``_return_fields`` untouched, so neither predicate above notices -- without + this short-circuit every healthy document would be reported as missing. """ + if getattr(query, "_no_content", False): + return False if unpack_json: return "json" not in doc_dict if isinstance(query, BaseVectorQuery) and query.DISTANCE_ID in getattr( @@ -426,7 +435,10 @@ def _process(doc: "Document") -> Any: return {"id": doc_dict.get("id"), **json_data} raise ValueError(f"Unable to parse json data from Redis {json_data}") - if norm_fn: + # The skip guard above guarantees the distance is present for a normal + # vector query, but not for a NOCONTENT one, where the server returns ids + # only and there is no distance to normalize. + if norm_fn and query.DISTANCE_ID in doc_dict: # type: ignore # convert float back to string to be consistent doc_dict[query.DISTANCE_ID] = str( # type: ignore norm_fn(float(doc_dict[query.DISTANCE_ID])) # type: ignore @@ -523,8 +535,10 @@ def _page_had_matches(results: Any) -> bool: 1. the server reported no more matches — iteration is genuinely done; 2. the server reported matches whose field payload could not be - materialized, so ``process_results`` dropped them (the Redis 8.8+ - background-WORKERS TTL/expiry race, see ``_has_missing_field_payload``). + materialized, so ``process_results`` dropped them -- a key that expires or + is updated mid-query is returned as a matched id with a ``nil`` field + array, and is still counted in the server's total (see + ``_has_missing_field_payload``). Stopping on case 2 silently truncates iteration and discards every remaining page. ``SearchResults.dropped_count`` tells the two apart: a page had matches @@ -1974,7 +1988,12 @@ def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator: considerations and the expected volume of search results. Note: - For stable pagination, the query must have a `sort_by` clause. + For stable pagination, the query must have a `sort_by` clause on a + **unique** field. Redis documents that ``LIMIT`` without sorting is + non-deterministic, so pages may otherwise repeat or miss documents. + Very deep pagination is also bounded server-side by + ``search-max-search-results`` (1,000,000 by default, but 10,000 on + some managed tiers), past which the search errors rather than ending. Note: A yielded batch may contain fewer than ``page_size`` documents, and an @@ -3216,7 +3235,12 @@ async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerato considerations and the expected volume of search results. Note: - For stable pagination, the query must have a `sort_by` clause. + For stable pagination, the query must have a `sort_by` clause on a + **unique** field. Redis documents that ``LIMIT`` without sorting is + non-deterministic, so pages may otherwise repeat or miss documents. + Very deep pagination is also bounded server-side by + ``search-max-search-results`` (1,000,000 by default, but 10,000 on + some managed tiers), past which the search errors rather than ending. Note: A yielded batch may contain fewer than ``page_size`` documents, and an diff --git a/tests/unit/test_query_types.py b/tests/unit/test_query_types.py index ff969e11..9575c131 100644 --- a/tests/unit/test_query_types.py +++ b/tests/unit/test_query_types.py @@ -1291,6 +1291,44 @@ def test_filter_query_hash_id_only_not_skipped(): assert results == [{"id": "doc:1"}] +def test_no_content_query_not_skipped(): + """Over-skip guard: a NOCONTENT query legitimately returns ids only. + + The server "returns the document ids and not the content" for every healthy + match, so a missing field payload carries no information and must not be read + as the expiry race. redis-py's ``no_content()`` leaves ``_return_fields`` + untouched, so both detection branches would otherwise fire on every document. + """ + nocontent = Result([2, "doc:1", "doc:2"], False) + + vector_query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + return_fields=["brand"], + ).no_content() + results = process_results(nocontent, vector_query, _hash_schema()) + assert [doc["id"] for doc in results] == ["doc:1", "doc:2"] + assert results.dropped_count == 0 + + # JSON full-object unpack: no "json" key is expected under NOCONTENT either. + json_query = FilterQuery(Tag("brand") == "Nike").no_content() + results = process_results(nocontent, json_query, _json_schema()) + assert [doc["id"] for doc in results] == ["doc:1", "doc:2"] + assert results.dropped_count == 0 + + +def test_no_content_query_with_normalize_does_not_raise(): + """NOCONTENT leaves no distance to normalize; the branch must skip it.""" + nocontent = Result([1, "doc:1"], False) + query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + normalize_vector_distance=True, + ).no_content() + # Previously raised KeyError on doc_dict[query.DISTANCE_ID]. + assert process_results(nocontent, query, _hash_schema()) == [{"id": "doc:1"}] + + def test_mixed_healthy_and_nil_doc(): """In a mixed result, the healthy doc survives and the nil doc is dropped.""" mixed = Result( From f8caa6d14945eff8de3f47d882a3dfbac1879d95 Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Fri, 7 Aug 2026 10:27:02 +0200 Subject: [PATCH 4/5] test: cut the paginate suite from 27 tests to 5 that catch strictly more A mutation-testing audit found the suite I expanded in a80dd01 was mostly redundant: many tests had kill-sets that were strict subsets of others, and six parametrized rows killed no mutant at all. Worse, it missed three real defects. Collapsed the drop-shape tests into one page map that exercises every shape in a single pass -- leading fully-dropped page, a healthy page that inherits its drops, a second dropped page forcing the carry reset, a partially-dropped page combining its own drop with the carried ones, then the genuine end. The `dropped_count` sequence `[2, 3]` is what pins the accounting: losing the fold gives `[0, 1]`, losing the reset gives `[2, 5]`, assigning instead of adding gives `[2, 2]`. Two defects survived all 27 tests and are now caught (verified by mutation): - the carry counter never being reset, double-counting across two dropped pages; - the async path skipping a partially-dropped page instead of yielding it. Deleted as redundant or fictional: the empty-page, zero-match, trailing-dropped and every-page-dropped tests (kill-sets subsumed by the merged pair); the nine-row `_page_had_matches` table (three rows killed nothing; the `int`/`None` rows cover inputs the CountQuery guard makes unreachable); the three `_fold_carried_drops` unit tests (one is a provably equivalent mutant, `+= 0`); and both plain-list tolerance tests, which exercised a path their own docstring admitted does not exist in production. That last deletion also repairs a false claim in a80dd01's message. Those two tests overrode `_query` directly instead of the `index.search` seam, so they had no request bound -- under a never-terminate regression they hung indefinitely rather than failing. Every remaining fake bounds its request count, so the claim now holds for the whole file. Kept beyond the merged pair: both CountQuery guards (separate sync and async code) and `page_size` validation, which nothing else in the repo covers. Net: 27 -> 5 tests, 388 -> 191 lines, and 6 of 6 targeted mutants killed. The only coverage genuinely given up is on inputs production cannot produce. --- tests/unit/test_paginate_dropped_page.py | 341 +++++------------------ 1 file changed, 72 insertions(+), 269 deletions(-) diff --git a/tests/unit/test_paginate_dropped_page.py b/tests/unit/test_paginate_dropped_page.py index 32fc2a5a..700f9f71 100644 --- a/tests/unit/test_paginate_dropped_page.py +++ b/tests/unit/test_paginate_dropped_page.py @@ -1,23 +1,25 @@ """Regression tests: ``paginate`` must not stop on a page whose matches were dropped. ``process_results`` deliberately drops matched documents whose field payload came -back missing (the Redis 8.8+ background-WORKERS TTL/expiry race), so a page can -be empty while the server still has matches to report. ``paginate`` previously -terminated on ``if not results: break``, conflating "no more matches" with -"matches we could not materialize" and silently discarding every remaining page. +back missing (a key that expires or is updated mid-query is returned as a matched +id with a ``nil`` field array, and is still counted in the server's total), so a +page can be empty while the server still has matches to report. ``paginate`` +previously terminated on ``if not results: break``, conflating "no more matches" +with "matches we could not materialize" and silently discarding every remaining +page. The live race is not reproducible on demand, so these tests serve canned ``FT.SEARCH`` replies from a fake ``index.search``, keyed by the paging offset the query carried. That seam runs the real ``_query`` -> ``process_results`` -> -``SearchResults`` chain, so the ``dropped_count`` the fix depends on is produced -by production code rather than hand-constructed. +``SearchResults`` chain, so the ``dropped_count`` the fix depends on is produced by +production code rather than hand-constructed. Every fake bounds its request count, +so a termination regression fails loudly instead of hanging the suite. """ import pytest from redis.commands.search.result import Result from redisvl.index import AsyncSearchIndex, SearchIndex, SearchResults -from redisvl.index.index import _fold_carried_drops, _page_had_matches from redisvl.query import CountQuery, VectorQuery from redisvl.schema import IndexSchema @@ -63,7 +65,7 @@ def _healthy(doc_id): def _race_victim(doc_id): - """A matched id whose field array came back nil -- the expiry race victim. + """A matched id whose field array came back nil -- the race victim. redis-py collapses the nil to an empty field set, leaving a ``Document`` whose ``__dict__`` is only ``{"id": ..., "payload": None}``. For a vector query @@ -80,152 +82,87 @@ def _page(*fragments): EMPTY_PAGE = Result([0], True) +# One pass over every drop shape that matters: +# +# offset 0 both matches dropped -> a LEADING dropped page must not terminate +# offset 2 both healthy -> carries offset 0's 2 drops (dropped_count 2) +# offset 4 both matches dropped -> a second carry episode, forcing the reset +# offset 6 1 healthy + 1 dropped -> own 1 drop plus carried 2 (dropped_count 3) +# offset 8 empty -> the genuine end of the result set +# +# The dropped_count sequence [2, 3] is what pins the accounting: losing the fold +# gives [0, 1], losing the reset gives [2, 5], and assigning instead of adding +# gives [2, 2]. +DROP_SHAPES = { + 0: _page(_race_victim("doc:1"), _race_victim("doc:2")), + 2: _page(_healthy("doc:3"), _healthy("doc:4")), + 4: _page(_race_victim("doc:5"), _race_victim("doc:6")), + 6: _page(_healthy("doc:7"), _race_victim("doc:8")), + 8: EMPTY_PAGE, +} + -def _install_pages(index, pages, seen): +def _fake_search(pages, seen): """Serve canned FT.SEARCH replies keyed by the query's paging offset. - Keying by offset (rather than popping a queue) means a failure to advance the - offset shows up as repeated documents or a wedge, not as a silently different - page. ``seen`` records the request order and bounds runaway loops. It reads + Keying by offset (rather than popping a queue) means a failure to advance + shows up as repeated documents or a wedge, not as a silently different page. + ``seen`` records the request order and bounds runaway loops. It reads redis-py's private ``Query._offset``, which ``paging()`` sets. """ - def _fake_search(query, query_params=None): + def _search(query, query_params=None): seen.append(query._offset) assert len(seen) <= 20, f"paginate made {len(seen)} requests; likely wedged" return pages.get(query._offset, EMPTY_PAGE) - index.search = _fake_search # type: ignore[method-assign] - - -# Three pages of matches where the whole middle page was dropped: the server -# matched doc:3 and doc:4 at offset 2 and neither could be materialized. Offset 6 -# is the genuine end of the result set. -DROPPED_MIDDLE_PAGES = { - 0: _page(_healthy("doc:1"), _healthy("doc:2")), - 2: _page(_race_victim("doc:3"), _race_victim("doc:4")), - 4: _page(_healthy("doc:5"), _healthy("doc:6")), - 6: EMPTY_PAGE, -} + return _search -def test_paginate_continues_past_fully_dropped_page(): - """A page whose matches were all dropped must not end iteration.""" +def test_paginate_continues_past_dropped_pages_and_accounts_for_their_drops(): index = SearchIndex(_schema()) seen: list[int] = [] - _install_pages(index, DROPPED_MIDDLE_PAGES, seen) + index.search = _fake_search(DROP_SHAPES, seen) # type: ignore[method-assign] batches = list(index.paginate(_query(), page_size=2)) - ids = [doc["id"] for batch in batches for doc in batch] - assert ids == ["doc:1", "doc:2", "doc:5", "doc:6"] - # The dropped page is skipped, not yielded: every batch stays non-empty. + # Iteration reached the end rather than stopping at offset 0's dropped page. + assert [[doc["id"] for doc in batch] for batch in batches] == [ + ["doc:3", "doc:4"], + ["doc:7"], + ] + # Never an empty batch, and the completeness metadata survives the generator. assert all(batch for batch in batches) - # Offset advanced across the dropped page so iteration made progress. - assert seen == [0, 2, 4, 6] - + assert all(isinstance(batch, SearchResults) for batch in batches) + # Drops from skipped pages ride along on the next batch the caller sees. + assert [batch.dropped_count for batch in batches] == [2, 3] + assert [batch.complete for batch in batches] == [False, False] + assert seen == [0, 2, 4, 6, 8] -def test_paginate_carries_dropped_count_of_skipped_page_forward(): - """A skipped page's drops must surface on the next yielded batch. - Otherwise the batches the caller actually sees all report ``complete is - True`` while matched documents went missing from the stream. - """ - index = SearchIndex(_schema()) - _install_pages(index, DROPPED_MIDDLE_PAGES, []) - - batches = list(index.paginate(_query(), page_size=2)) - - assert batches[0].dropped_count == 0 - assert batches[0].complete is True - # doc:3 and doc:4 were dropped with their page; the count rides along here. - assert batches[1].dropped_count == 2 - assert batches[1].complete is False - - -def test_paginate_yields_partially_dropped_page_with_its_count(): - """A page with survivors AND drops is yielded, carrying its own count.""" - index = SearchIndex(_schema()) - _install_pages( - index, - { - 0: _page(_healthy("doc:1"), _race_victim("doc:2")), - 2: EMPTY_PAGE, - }, - [], - ) - - batches = list(index.paginate(_query(), page_size=2)) - - assert len(batches) == 1 - assert [doc["id"] for doc in batches[0]] == ["doc:1"] - # Batches must remain SearchResults, or the completeness contract is severed. - assert isinstance(batches[0], SearchResults) - assert batches[0].dropped_count == 1 - assert batches[0].complete is False - - -def test_paginate_stops_when_server_reports_no_matches(): - """An empty page with nothing dropped is still the end of the result set.""" - index = SearchIndex(_schema()) - seen: list[int] = [] - _install_pages(index, {0: _page(_healthy("doc:1")), 1: EMPTY_PAGE}, seen) - - batches = list(index.paginate(_query(), page_size=1)) - - assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] - # Stopped on the empty page -- no third request. - assert seen == [0, 1] - - -def test_paginate_zero_matches_yields_nothing(): - """An empty first page terminates immediately rather than looping.""" - index = SearchIndex(_schema()) - seen: list[int] = [] - _install_pages(index, {}, seen) - - assert list(index.paginate(_query(), page_size=10)) == [] - assert seen == [0] - - -def test_paginate_terminates_on_trailing_dropped_page(): - """A fully-dropped final page costs one extra request, then terminates.""" - index = SearchIndex(_schema()) +@pytest.mark.asyncio +async def test_async_paginate_continues_past_dropped_pages_and_accounts_for_their_drops(): + """Async mirror: ``AsyncSearchIndex.paginate`` is a hand-duplicated copy.""" + index = AsyncSearchIndex(_schema()) seen: list[int] = [] - _install_pages( - index, - { - 0: _page(_healthy("doc:1"), _healthy("doc:2")), - 2: _page(_race_victim("doc:3")), - 4: EMPTY_PAGE, - }, - seen, - ) - - batches = list(index.paginate(_query(), page_size=2)) + sync_search = _fake_search(DROP_SHAPES, seen) - assert [doc["id"] for batch in batches for doc in batch] == ["doc:1", "doc:2"] - assert seen == [0, 2, 4] - # Trailing drops have no later batch to ride on; process_results logs them. - assert batches[0].dropped_count == 0 + async def _search(query, query_params=None): + return sync_search(query, query_params) + index.search = _search # type: ignore[method-assign] -def test_paginate_terminates_when_every_page_is_dropped(): - """All-dropped pages must terminate once the server runs out of matches.""" - index = SearchIndex(_schema()) - seen: list[int] = [] - _install_pages( - index, - { - 0: _page(_race_victim("doc:1"), _race_victim("doc:2")), - 2: _page(_race_victim("doc:3"), _race_victim("doc:4")), - 4: EMPTY_PAGE, - }, - seen, - ) + batches = [batch async for batch in index.paginate(_query(), page_size=2)] - assert list(index.paginate(_query(), page_size=2)) == [] - assert seen == [0, 2, 4] + assert [[doc["id"] for doc in batch] for batch in batches] == [ + ["doc:3", "doc:4"], + ["doc:7"], + ] + assert all(batch for batch in batches) + assert all(isinstance(batch, SearchResults) for batch in batches) + assert [batch.dropped_count for batch in batches] == [2, 3] + assert [batch.complete for batch in batches] == [False, False] + assert seen == [0, 2, 4, 6, 8] def test_paginate_rejects_count_query(): @@ -239,90 +176,6 @@ def test_paginate_rejects_count_query(): list(index.paginate(CountQuery("*"), page_size=2)) -def test_paginate_validates_page_size(): - index = SearchIndex(_schema()) - with pytest.raises(TypeError, match="page_size must be an integer"): - list(index.paginate(_query(), page_size="2")) # type: ignore[arg-type] - with pytest.raises(ValueError, match="page_size must be greater than 0"): - list(index.paginate(_query(), page_size=0)) - - -def test_paginate_tolerates_plain_list_pages(): - """``paginate`` still works if ``_query`` is overridden to return a plain list. - - No production query path does this -- every one returns ``SearchResults`` -- - but the termination check must not assume the metadata is present. - """ - index = SearchIndex(_schema()) - responses: list[list[dict]] = [[{"id": "doc:1"}], []] - index._query = lambda query: responses.pop(0) if responses else [] # type: ignore[method-assign] - - batches = list(index.paginate(_query(), page_size=1)) - - assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] - - -# --------------------------------------------------------------------------- -# Async mirrors. AsyncSearchIndex.paginate is a hand-duplicated copy of the sync -# logic, so it is the natural place for the two to drift. -# --------------------------------------------------------------------------- - - -def _install_async_pages(index, pages, seen): - async def _fake_search(query, query_params=None): - seen.append(query._offset) - assert len(seen) <= 20, f"paginate made {len(seen)} requests; likely wedged" - return pages.get(query._offset, EMPTY_PAGE) - - index.search = _fake_search # type: ignore[method-assign] - - -@pytest.mark.asyncio -async def test_async_paginate_continues_past_fully_dropped_page(): - """Async mirror: the dropped middle page must not truncate iteration.""" - index = AsyncSearchIndex(_schema()) - seen: list[int] = [] - _install_async_pages(index, DROPPED_MIDDLE_PAGES, seen) - - batches = [batch async for batch in index.paginate(_query(), page_size=2)] - - ids = [doc["id"] for batch in batches for doc in batch] - assert ids == ["doc:1", "doc:2", "doc:5", "doc:6"] - assert all(batch for batch in batches) - assert seen == [0, 2, 4, 6] - # Skipped page's drops carried onto the following batch. - assert batches[1].dropped_count == 2 - - -@pytest.mark.asyncio -async def test_async_paginate_stops_when_server_reports_no_matches(): - index = AsyncSearchIndex(_schema()) - seen: list[int] = [] - _install_async_pages(index, {0: _page(_healthy("doc:1")), 1: EMPTY_PAGE}, seen) - - batches = [batch async for batch in index.paginate(_query(), page_size=1)] - - assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] - assert seen == [0, 1] - - -@pytest.mark.asyncio -async def test_async_paginate_terminates_when_every_page_is_dropped(): - index = AsyncSearchIndex(_schema()) - seen: list[int] = [] - _install_async_pages( - index, - { - 0: _page(_race_victim("doc:1"), _race_victim("doc:2")), - 2: EMPTY_PAGE, - }, - seen, - ) - - assert [batch async for batch in index.paginate(_query(), page_size=2)] == [] - assert seen == [0, 2] - - @pytest.mark.asyncio async def test_async_paginate_rejects_count_query(): index = AsyncSearchIndex(_schema()) @@ -330,59 +183,9 @@ async def test_async_paginate_rejects_count_query(): [batch async for batch in index.paginate(CountQuery("*"), page_size=2)] -@pytest.mark.asyncio -async def test_async_paginate_tolerates_plain_list_pages(): - index = AsyncSearchIndex(_schema()) - responses: list[list[dict]] = [[{"id": "doc:1"}], []] - - async def _fake_query(query): - return responses.pop(0) if responses else [] - - index._query = _fake_query # type: ignore[method-assign] - - batches = [batch async for batch in index.paginate(_query(), page_size=1)] - - assert [doc["id"] for batch in batches for doc in batch] == ["doc:1"] - - -# --------------------------------------------------------------------------- -# Helper contracts, pinned directly so the intent survives a refactor. -# --------------------------------------------------------------------------- - - -@pytest.mark.parametrize( - "results, expected", - [ - (SearchResults([]), False), - (SearchResults([{"id": "doc:1"}]), True), - (SearchResults([], dropped_count=1), True), - (SearchResults([{"id": "doc:1"}], dropped_count=1), True), - ([], False), - ([{"id": "doc:1"}], True), - # process_results returns a bare int for CountQuery; must not raise. - (0, False), - (5, True), - (None, False), - ], -) -def test_page_had_matches(results, expected): - assert _page_had_matches(results) is expected - - -def test_fold_carried_drops_adds_to_existing_count(): - results = SearchResults([{"id": "doc:1"}], dropped_count=1) - _fold_carried_drops(results, 2) - assert results.dropped_count == 3 - - -def test_fold_carried_drops_noop_without_carry(): - results = SearchResults([{"id": "doc:1"}], dropped_count=1) - _fold_carried_drops(results, 0) - assert results.dropped_count == 1 - - -def test_fold_carried_drops_ignores_plain_list(): - """A plain list carries no metadata; folding must not raise.""" - plain = [{"id": "doc:1"}] - _fold_carried_drops(plain, 2) - assert plain == [{"id": "doc:1"}] +def test_paginate_validates_page_size(): + index = SearchIndex(_schema()) + with pytest.raises(TypeError, match="page_size must be an integer"): + list(index.paginate(_query(), page_size="2")) # type: ignore[arg-type] + with pytest.raises(ValueError, match="page_size must be greater than 0"): + list(index.paginate(_query(), page_size=0)) From 564f96bfe832d97817faf6b9a3329eb47720e911 Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Mon, 24 Aug 2026 16:23:53 +0200 Subject: [PATCH 5/5] test: cover the NOCONTENT payload fix against a real server The unit tests for the ``_no_content`` short-circuit drive ``process_results`` with a hand-built ``Result``. This adds the integration counterpart, which needs no race to trigger: under ``NOCONTENT`` a real server returns ids and no field data for every healthy match, which is exactly the shape the drop heuristic uses to detect an expiry-race victim. Covers the three shapes that regressed -- plain ``FilterQuery``, ``VectorQuery``, and ``VectorQuery(normalize_vector_distance=True)`` (which raised ``KeyError`` rather than returning ids) -- plus ``paginate`` over a NOCONTENT result set, tying the fix to the termination change on this branch. Verified to fail with the short-circuit removed. --- tests/integration/test_query.py | 51 +++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/tests/integration/test_query.py b/tests/integration/test_query.py index 335a37bc..776ce25f 100644 --- a/tests/integration/test_query.py +++ b/tests/integration/test_query.py @@ -605,6 +605,57 @@ def test_paginate_range_query(index, range_query): assert all(float(item["vector_distance"]) <= 0.2 for item in all_results) +def test_no_content_query_returns_ids(index, sample_data): + """NOCONTENT results must survive the missing-field-payload filter. + + Unlike the expiry race this needs no timing: under ``NOCONTENT`` the server + returns ids and no field data for *every* healthy match, so the drop + heuristic in ``process_results`` sees the same shape it uses to detect a + race victim. Before the ``_no_content`` short-circuit this returned an empty + list for every NOCONTENT query -- including through ``paginate``. + """ + total = len(sample_data) + high_credit = sum(1 for d in sample_data if d["credit_score"] == "high") + + # Plain filter, vector, and normalize-distance shapes. The last one used to + # raise KeyError on the absent vector_distance rather than return ids. + for query, expected in [ + (FilterQuery(filter_expression=Tag("credit_score") == "high"), high_credit), + ( + VectorQuery( + vector=[0.1, 0.1, 0.5], + vector_field_name="user_embedding", + num_results=total, + ), + total, + ), + ( + VectorQuery( + vector=[0.1, 0.1, 0.5], + vector_field_name="user_embedding", + num_results=total, + normalize_vector_distance=True, + ), + total, + ), + ]: + results = index.query(query.no_content()) + assert [doc["id"] for doc in results] and len(results) == expected + assert results.dropped_count == 0 + assert results.complete + + # paginate must reach the end of a NOCONTENT result set, not stop on page 1. + paged = [ + doc + for batch in index.paginate( + FilterQuery(filter_expression=Tag("credit_score") == "high").no_content(), + page_size=2, + ) + for doc in batch + ] + assert len(paged) == high_credit + + def test_sort_filter_query(index, sorted_filter_query): t = Text("job") % "" search(sorted_filter_query, index, t, 7, sort=True)