Skip to content

fix(core): cancel sibling embedding tasks when one of them fails - #23669

Open
lester076 wants to merge 2 commits into
run-llama:mainfrom
lester076:fix-embedding-batch-cancel
Open

lester076 wants to merge 2 commits into
run-llama:mainfrom
lester076:fix-embedding-batch-cancel

Conversation

@lester076

@lester076 lester076 commented Oct 23, 2026 •

Copy link
Copy Markdown

What this PR does

BaseEmbedding.aget_text_embedding_batch fans out embedding work at two levels and both used a bare asyncio.gather:

  • per text, inside _aget_text_embeddings;
  • per sub-batch, in aget_text_embedding_batch.

asyncio.gather propagates the first exception to the caller but leaves the remaining tasks running detached: nothing awaits them and nothing cancels them. For embeddings that is worse than a plain crash, because the provider has already done billable work on the sibling requests, and that work — and its vectors — is silently discarded. Any integration that relies on the default fan-out instead of a native batch call inherits both gaps.

This PR adds gather_cancel_on_error to llama_index.core.async_utils and uses it at both levels in BaseEmbedding and BaseSparseEmbedding:

futures = [asyncio.ensure_future(task) for task in tasks]
try:
    return await asyncio.gather(*futures)
except BaseException:
    for future in futures:
        future.cancel()
    await asyncio.gather(*futures, return_exceptions=True)
    raise

Cancelling and then draining before re-raising keeps the original exception as the one the caller sees, and guarantees no orphaned coroutine survives the failure. BaseException is caught (not just Exception) so an outer cancellation of the whole call also cleans up its children instead of abandoning them.

The show_progress=True branch goes through tqdm_asyncio.gather, which has no return_exceptions switch (it is built on as_completed), so that branch collects the tasks first and applies the same cancel-and-drain on failure.

Scope

  • _aget_text_embeddings (both dense and sparse) and the sub-batch fan-out in aget_text_embedding_batch (both dense and sparse).
  • run_jobs / num_workers > 1 is untouched: that helper manages its own worker pool and is used in many other places, so changing it belongs in a separate PR.
  • Behaviour on the success path is unchanged; ordering of results is preserved.

Tests

New file llama-index-core/tests/embeddings/test_async_batch_cancellation.py covers dense and sparse embeddings at both fan-out levels plus the helper itself, asserting that the failing text propagates its exception while every sibling reports CancelledError instead of completing.

Command Result
pytest tests/embeddings/test_async_batch_cancellation.py -q (with this change) 8 passed
same tests on the previous code 5 failed, 3 passed (siblings complete instead of being cancelled, including on the show_progress path)
pytest tests/embeddings tests/base -q 122 passed, 4 pre-existing errors from a Windows temp-dir permission issue unrelated to this change
ruff check / ruff format --check on the changed files clean

Regression scope: I ran pytest tests/embeddings tests/ingestion tests/indices -q with and without the source change and diffed the failing-test sets. The pre-existing failures are identical in both runs (missing optional dependencies and Windows temp-dir permissions in this environment); the only delta is the five new regressions, which fail without the change and pass with it. No test that passes on main starts failing here.

Package version bumped to 0.14.26.

AI assistance disclosure

Prepared with an AI coding agent, per the repository's guidelines. I reviewed the diff, wrote the regressions first and confirmed they fail without the source change, and ran the test and lint commands above.

Refs #22312

aget_text_embedding_batch fans out per-text and per-sub-batch calls with plain asyncio.gather, which propagates the first exception but leaves the other tasks running detached. Nothing awaits or cancels them, so a single rejected text orphans in-flight embedding requests for every sibling: the provider has already done billable work whose result is discarded.

Add gather_cancel_on_error to async_utils: it futures the coroutines, gathers them, and on any failure cancels the survivors and drains them before re-raising. Use it in BaseEmbedding and BaseSparseEmbedding at both fan-out levels, including the tqdm progress path, which has no return_exceptions switch of its own.

Adds regressions for dense and sparse embeddings at both levels plus a direct helper test; the tests fail on the previous code and pass with this change.
The show_progress branch goes through tqdm_asyncio.gather, which has no return_exceptions switch, so it needs its own cancel-and-drain coverage. Both the failure and the success case are pinned here.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant