asyncio.gather leaks orphaned tasks in MultiRetriever/MultiQueryTextRetriever/MultiQueryEmbeddingRetriever when one concurrent call fails

[ErenAta16]

Bug Description

MultiRetriever.run_async, MultiQueryTextRetriever.run_async, and MultiQueryEmbeddingRetriever.run_async all fan out concurrent work with asyncio.gather(...) (no return_exceptions=True) over per-query/per-retriever coroutines that do not catch every exception internally. When one of the concurrent calls fails while others are still in flight, gather propagates the first exception immediately, but it does not cancel or await the sibling tasks. They keep running detached from the caller, and their eventual result (or exception) is silently discarded, or logged as "Task exception was never retrieved" if the event loop is still alive when they finish, or produces a "Task was destroyed but it is pending" warning if the loop shuts down first.

Affected code:

• haystack/components/retrievers/multi_retriever.py, _run_one (re-raises as RuntimeError from except Exception, still propagates past gather):

async def _run_one(name: str, retriever: TextRetriever) -> list[Document]:
    try:
        result = await _execute_component_async(retriever, **run_kwargs)
        return result.get("documents", [])
    except Exception as e:
        raise RuntimeError(f"Retriever '{name}' failed: {e}") from e

document_lists = list(await asyncio.gather(*[_run_one(name, r) for name, r in retrievers_to_run.items()]))

• haystack/components/retrievers/multi_query_text_retriever.py, _run_one_async (no try/except at all):

async def _run_one_async(self, query: str, retriever_kwargs: dict[str, Any]) -> list[Document] | None:
    result = await _execute_component_async(self.retriever, query=query, **retriever_kwargs)
    ...

results = await asyncio.gather(*[self._run_one_async(q, retriever_kwargs) for q in queries])

• haystack/components/retrievers/multi_query_embedding_retriever.py, same pattern as above.

Steps to Reproduce

Minimal reproduction of the leak, independent of Haystack, showing the underlying asyncio.gather semantics these components rely on:

import asyncio

async def _run_one(name, should_fail, delay):
    try:
        if should_fail:
            await asyncio.sleep(0.01)
            raise ValueError(f"{name} broke")
        await asyncio.sleep(delay)
        return [f"doc_from_{name}"]
    except Exception as e:
        raise RuntimeError(f"Retriever '{name}' failed: {e}") from e

async def main():
    before = asyncio.all_tasks()
    try:
        await asyncio.gather(
            _run_one("fast_failing", True, 0),
            _run_one("slow_retriever", False, 2.0),
        )
    except RuntimeError as e:
        print("gather raised as expected:", e)
    leaked = asyncio.all_tasks() - before
    print("orphaned tasks immediately after gather raises:", len(leaked))

asyncio.run(main())

Output:

gather raised as expected: Retriever 'fast_failing' failed: fast_failing broke
orphaned tasks immediately after gather raises: 1

A second run simulating a pipeline step that calls the component, catches the exception, and returns to its own caller confirms the caller regains control while the slow retriever's task is still pending, running fully detached:

pipeline.run_async: caught retriever failure -> fast_failing_retriever broke
caller received exception, pipeline call has now returned: fast_failing_retriever broke
tasks still pending right after the pipeline call raised: 1

Expected Behavior

When one query/retriever fails, the other in-flight tasks in the same batch should be cancelled (or gathered with return_exceptions=True and drained) before the exception propagates, so no task from that batch is left running unsupervised past the point where the caller has already moved on. This matters most for retrievers backed by real I/O (HTTP calls to a vector DB or search API, DB cursors), where the leaked task holds those resources open for however long it takes to finish or get torn down by the loop.

System

deepset-ai/haystack main branch (current HEAD as of this report). Affects any usage of MultiRetriever.run_async, MultiQueryTextRetriever.run_async, or MultiQueryEmbeddingRetriever.run_async where at least one concurrent call can fail independently of the others, which is the normal case for retrievers hitting separate backends.

[hxaxd]

Thanks for the detailed report. I reviewed the current implementations and agree that the core issue is valid: when one concurrent retriever/query fails, the first exception is propagated while the remaining sibling tasks are left running instead of being cancelled and drained.

There also seems to be a clear implementation lineage here:

• MultiRetriever.run_async() was introduced in feat: Add new components TextEmbeddingRetriever and MultiRetriever #10872 with the current asyncio.gather() fan-out behavior.
• MultiQueryTextRetriever.run_async() and MultiQueryEmbeddingRetriever.run_async() were later introduced in feat: add run_async to TextEmbeddingRetriever, MultiQueryEmbeddingRetriever, and MultiQueryTextRetriever #11367, whose description explicitly states that they follow the pattern established by MultiRetriever.run_async().

At the Pipeline level, Haystack already applies fail-fast cleanup semantics: when one in-flight component task fails, the remaining Pipeline-managed tasks are cancelled and drained before the exception is propagated. Since these three retriever components perform a similar nested fan-out internally, I think their failure semantics should be consistent with the Pipeline behavior as well.

My proposed change would be to preserve the current first-exception behavior, while explicitly cancelling and draining the remaining sibling tasks before re-raising the original exception. I would also add regression tests for all three components to verify that no sibling task remains pending after one concurrent call fails.

If this direction sounds appropriate, would the maintainers be open to a PR implementing it?

[zeweihan]

Precise bug report with the implementation lineage traced (#10872 → #11367) — and the "fan-out without fail-fast cleanup" framing is exactly right. The existing reviewer comment already covers the mechanical fix well, so I'll add a regulated-domain lens on why the orphaned-task half of this bug is more dangerous than the crash itself, and why it matters for document-heavy RAG.

1. The crash is the favorable failure mode; the orphaned tasks are the dangerous one. When gather propagates the first exception, the caller sees a failure and the pipeline halts — that's loud and recoverable. But the sibling tasks that are still in flight when the exception fires don't vanish; they continue running detached from the caller, and their eventual results (or exceptions) are silently discarded. In a retrieval pipeline where each task queries a different corpus partition, store, or shard, that means a query silently succeeded against a subset of the sources, produced real documents, and was thrown away — while the caller moved on believing the operation failed. The signal the caller received ("retrieval failed") is now materially wrong: retrieval partially succeeded and the partial results evaporated with no audit trail.

2. The "Task exception was never retrieved" / "Task was destroyed but it is pending" warnings are the only footprint — and they live in logs, not in the retrieval result. This is the same shape as the connector-subsystem family of failures (operation reports success or failure while the actual data-flow state diverges, with no degradation signal at the result layer). For a regulated corpus (legal/compliance/financial documents), "which sources did we actually query for this answer?" is a question an auditor will eventually ask. If three of five retrievers were orphaned mid-flight, the answer layer has no way to know which sources were consulted, which returned documents that were discarded, and which never ran at all. The pipeline's bookkeeping says "failed" or "ran N retrievers" — the event loop knows something different — and neither matches what actually happened.

3. Asymmetric blast radius on the MultiQuery* variants. The MultiRetriever fan-out is across retrievers (different backends); the MultiQuery* variants fan out across query reformulations of the same retriever. When a reformulated query's task is orphaned, the retrieval result is silently missing the contribution of one entire query variant. For retrieval quality, that's not "one of N sources is absent" — it's "the recall surface for this single source was silently narrowed." A smoke test that checks "did we get documents back" passes (the non-orphaned queries returned results), but the recall is degraded in a way that's invisible to the consumer. The "plausible-but-incomplete" failure: the pipeline returned a fluent, confident answer built on a silently-narrowed retrieval set, and nothing in the output says "one reformulated query's results were lost."

4. Cheap companion independent of the gather fix: a per-run task-manifest / completion signal. The mechanical fix (cancel and drain siblings on first failure, or use return_exceptions=True and aggregate) is necessary and correct. But a durable guard that catches the class: after the fan-out completes (or aborts), emit a task-completion manifest per retrieval run — {queries_dispatched, queries_completed, queries_orphaned, queries_errored}. Then "two of five reformulated queries were silently orphaned" becomes an inspectable provenance fact on the retrieval output rather than a log-line discoverable only by grepping for "Task was destroyed." Same philosophy as a coverage/completeness signal elsewhere: the dangerous case is the run that looks complete because the surviving tasks returned results, when a subset silently never finished.

Worth noting the Pipeline-level fail-fast cleanup that the reviewer mentioned is the right reference architecture — the gap is that the individual retriever components don't apply the same sibling-cancellation discipline locally. Aligning the component-level gather with the pipeline-level cleanup semantics would close the class, and the per-run manifest would make the alignment inspectable.

[ErenAta16]

Thanks both for the deep dive, this is more thorough than the original report.

@hxaxd, cancel-and-drain is actually the better fix here over a plain return_exceptions=True gather, which is what I'd sketched in the report. return_exceptions=True alone trades the leak for added latency (the caller now waits for the slowest sibling even when it already knows the batch failed), whereas cancelling the remaining tasks on first failure keeps the current fail-fast behavior and still guarantees nothing is left running past the point where the exception propagates. That's a meaningfully better tradeoff, and matches the Pipeline-level cleanup semantics you traced, so the fix would make the retriever components consistent with how the rest of the framework already handles this.

Would support a PR along those lines. Happy to review it against a repro like the one in the report (one fast-failing retriever alongside slower ones) to confirm no task is left pending after cancellation.

[hxaxd]

@ErenAta16 Thanks — I’ve opened #11967 implementing this cancel-and-drain approach for all three affected retrievers. The regression tests cover the fast-failing/in-flight sibling scenario and verify that the sibling task receives cancellation. The PR is ready for review.