diff --git a/backend/packages/harness/deerflow/utils/file_conversion.py b/backend/packages/harness/deerflow/utils/file_conversion.py index 53cec1a9ab..ac7c74508f 100644 --- a/backend/packages/harness/deerflow/utils/file_conversion.py +++ b/backend/packages/harness/deerflow/utils/file_conversion.py @@ -19,6 +19,7 @@ import logging from pathlib import Path from deerflow.config.app_config import get_app_config +from deerflow.utils.file_io import await_drained, run_file_io # Backward-compat re-exports — outline extraction moved to file_outline.py. from deerflow.utils.file_outline import ( # noqa: F401 @@ -159,7 +160,10 @@ async def convert_file_to_markdown(file_path: Path, output_path: Path | None = N """ try: pdf_converter = _get_pdf_converter() - file_size = file_path.stat().st_size + # stat and the write-back are blocking filesystem calls on the upload + # ingestion event loop; offload them like the large-file conversion + # above (run_file_io keeps the caller's context in the worker). + file_size = (await run_file_io(file_path.stat)).st_size if file_size > _ASYNC_THRESHOLD_BYTES: text = await asyncio.to_thread(_do_convert, file_path, pdf_converter) @@ -167,7 +171,23 @@ async def convert_file_to_markdown(file_path: Path, output_path: Path | None = N text = _do_convert(file_path, pdf_converter) md_path = output_path if output_path is not None else file_path.with_suffix(".md") - md_path.write_text(text, encoding="utf-8") + # Drain the write-back across caller cancellation: run_file_io awaits + # run_in_executor, so a plain await would unwind while the worker is + # still writing and leave the partial output behind for the caller's + # scope to trip over. await_drained delivers the cancellation only + # after the worker finished, so the cleanup below sees a settled file. + try: + await await_drained(run_file_io(md_path.write_text, text, encoding="utf-8")) + except asyncio.CancelledError: + try: + await await_drained(run_file_io(md_path.unlink, missing_ok=True)) + except Exception: + # Cleanup must not shadow the cancellation: an unlink failure + # escaping this handler used to be swallowed by the broad + # handler below as an ordinary conversion failure, leaving + # both the caller's cancellation and the staging file behind. + logger.exception("Failed to remove partial conversion output %s after cancellation", md_path) + raise logger.info("Converted %s to markdown: %s (%d chars)", file_path.name, md_path.name, len(text)) return md_path diff --git a/backend/packages/harness/deerflow/utils/file_io.py b/backend/packages/harness/deerflow/utils/file_io.py index 771b0c9c52..d6f84363c7 100644 --- a/backend/packages/harness/deerflow/utils/file_io.py +++ b/backend/packages/harness/deerflow/utils/file_io.py @@ -8,6 +8,7 @@ import contextvars import functools import logging import os +import threading from collections.abc import Callable, Coroutine from concurrent.futures import ThreadPoolExecutor from typing import Any @@ -30,6 +31,10 @@ def _default_file_io_workers() -> int: _FILE_IO_EXECUTOR = ThreadPoolExecutor(max_workers=_default_file_io_workers(), thread_name_prefix="file-io") +# Marks the threads of ``_FILE_IO_EXECUTOR`` so ``run_file_io`` can detect +# (and run inline on) re-entrant calls. +_FILE_IO_WORKER = threading.local() + def _shutdown_file_io_executor() -> None: _FILE_IO_EXECUTOR.shutdown(wait=False, cancel_futures=True) @@ -38,6 +43,22 @@ def _shutdown_file_io_executor() -> None: atexit.register(_shutdown_file_io_executor) +def _worker_call(call: Callable[[], Any]) -> Any: + """Run *call* inside a file-IO worker, marking the thread as such. + + The marker makes ``run_file_io`` re-entrant: a nested offload issued from + code already running in a worker (e.g. ``asyncio.run`` inside a worker + body) runs inline instead of submitting to the same bounded executor, + which would deadlock once every worker is busy — deterministically so + with a single configured worker. + """ + _FILE_IO_WORKER.active = True + try: + return call() + finally: + _FILE_IO_WORKER.active = False + + async def run_file_io[**P, T](func: Callable[P, T], /, *args: P.args, **kwargs: P.kwargs) -> T: """Run blocking filesystem-oriented work on the dedicated file IO pool. @@ -45,11 +66,17 @@ async def run_file_io[**P, T](func: Callable[P, T], /, *args: P.args, **kwargs: ``loop.run_in_executor`` does not. Copy the current context explicitly so user-scoped helpers such as ``get_effective_user_id()`` keep working inside the worker thread. + + Re-entrant: called from inside a file-IO worker thread, *func* runs inline + rather than submitting to the pool again — the pool is bounded, so a + nested submit can deadlock when every worker is already busy. """ + call = functools.partial(func, *args, **kwargs) + if getattr(_FILE_IO_WORKER, "active", False): + return call() loop = asyncio.get_running_loop() ctx = contextvars.copy_context() - call = functools.partial(func, *args, **kwargs) - return await loop.run_in_executor(_FILE_IO_EXECUTOR, ctx.run, call) + return await loop.run_in_executor(_FILE_IO_EXECUTOR, ctx.run, _worker_call, call) async def await_drained[T](coro: Coroutine[Any, Any, T]) -> T: diff --git a/backend/tests/blocking_io/test_file_conversion_offloop.py b/backend/tests/blocking_io/test_file_conversion_offloop.py new file mode 100644 index 0000000000..9188b37bea --- /dev/null +++ b/backend/tests/blocking_io/test_file_conversion_offloop.py @@ -0,0 +1,95 @@ +"""Regression anchor: ``convert_file_to_markdown`` must not block the event loop. + +The converter itself is offloaded to a thread for files above 1 MB +(``_ASYNC_THRESHOLD_BYTES``), but the converted markdown was written back with +a synchronous ``Path.write_text`` on the event loop — a multi-megabyte blocking +write in the upload ingestion path (``app/gateway/upload_ingestion.py`` calls +this per uploaded document). This anchor drives the real +``convert_file_to_markdown`` under the strict Blockbuster gate with the +converter patched to return a large payload, so only the write-back is +exercised. + +If the write regresses back onto the event loop, Blockbuster raises +``BlockingError`` — which the function's broad ``except`` turns into a ``None`` +return, so this test fails on the missing output file. +""" + +from __future__ import annotations + +import asyncio +import threading +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.asyncio + + +async def test_convert_file_to_markdown_write_does_not_block_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from deerflow.utils import file_conversion + + large_text = "x" * (2 * 1024 * 1024) # 2 MB conversion result + monkeypatch.setattr(file_conversion, "_do_convert", lambda *_args, **_kwargs: large_text) + + source = tmp_path / "doc.txt" + source.write_text("source", encoding="utf-8") # test-side seeding (not in scanned_modules) + + md_path = await file_conversion.convert_file_to_markdown(source) + + assert md_path is not None, "conversion failed — see captured logs for the swallowed error" + assert md_path.read_text(encoding="utf-8") == large_text + + +async def test_cancelled_conversion_cleanup_failure_preserves_cancellation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """A failing cleanup ``unlink`` must not swallow the cancellation. + + The cleanup ran bare inside the ``CancelledError`` handler: an unlink + OSError escaped the handler, was converted into an ordinary conversion + failure (``None`` return) by the broad ``except``, and left both the + caller's cancellation and the staging output behind. Cleanup failures + are now caught and logged inside the handler, and the captured + cancellation is always re-raised. + """ + from deerflow.utils import file_conversion + + large_text = "x" * (2 * 1024 * 1024) + monkeypatch.setattr(file_conversion, "_do_convert", lambda *_args, **_kwargs: large_text) + + source = tmp_path / "doc.txt" + source.write_text("source", encoding="utf-8") + + def failing_unlink(self: Path, missing_ok: bool = False) -> None: + raise OSError("simulated unlink failure") + + monkeypatch.setattr(Path, "unlink", failing_unlink) + + real_run_file_io = file_conversion.run_file_io + started = threading.Event() + finish = threading.Event() + + async def gated_run_file_io(fn, *args, **kwargs): + if getattr(fn, "__name__", "") == "write_text": + # Hold the write-back worker start until the test has cancelled + # the task, so the cancellation arrives while the write is in + # flight (deterministic ordering, no sleeps). + started.set() + await asyncio.to_thread(finish.wait, 5) + return await real_run_file_io(fn, *args, **kwargs) + + monkeypatch.setattr(file_conversion, "run_file_io", gated_run_file_io) + + task = asyncio.create_task(file_conversion.convert_file_to_markdown(source)) + try: + await asyncio.to_thread(started.wait, 5) + task.cancel() + finish.set() + with pytest.raises(asyncio.CancelledError): + await task + finally: + finish.set() + + assert any("partial conversion output" in record.getMessage() for record in caplog.records), "cleanup failure should be logged inside the handler, not escaped" diff --git a/backend/tests/test_file_conversion.py b/backend/tests/test_file_conversion.py index 407b8ae120..d6cd74f661 100644 --- a/backend/tests/test_file_conversion.py +++ b/backend/tests/test_file_conversion.py @@ -485,3 +485,100 @@ class TestExtractOutline: assert len(outline) == 1 # Title must be clean — no ** ** artefacts assert outline[0]["title"] == "UNITED STATES SECURITIES AND EXCHANGE COMMISSION" + + +# --------------------------------------------------------------------------- +# Off-loop stat/write-back: cancellation drain and executor re-entrancy +# (review follow-ups on the uploads conversion PR) +# --------------------------------------------------------------------------- + + +def test_cancelled_conversion_drains_write_and_removes_partial(tmp_path, monkeypatch): + """Cancel must not abandon the in-flight write, and must clean the partial file. + + ``run_file_io`` awaits ``run_in_executor``, so a plain await would unwind + while the worker is still writing; the drained write delivers the + cancellation only after the worker settled, and the handler then removes + the partial output so repeated cancelled conversions cannot leave large + hidden staging files behind. + """ + import threading + from pathlib import Path + + started = threading.Event() + cancel_issued = threading.Event() + release = threading.Event() + done = threading.Event() + real_write_text = Path.write_text + + def _blocking_write_text(self, data, encoding=None): + started.set() + release.wait(timeout=10) + result = real_write_text(self, data, encoding=encoding) + done.set() + return result + + # Seed the source BEFORE installing the patch, so the setup write is not + # captured by _blocking_write_text (which would pre-set started/done). + src = tmp_path / "doc.txt" + out = tmp_path / "doc.md" + src.write_text("hello") + + monkeypatch.setattr(Path, "write_text", _blocking_write_text) + + async def _scenario(): + task = asyncio.create_task(convert_file_to_markdown(src, output_path=out)) + await asyncio.to_thread(started.wait, 10) + task.cancel() + # Release the worker only after the cancel is in flight, so the + # ordering (cancel first, then the write settles) is forced and the + # assertion below cannot pass by racing the releaser. + cancel_issued.set() + try: + await task + except asyncio.CancelledError: + # The drain invariant: the cancellation is delivered only after + # the in-flight write settled. A plain run_file_io await cancels + # immediately and leaves the worker running. + return done.is_set() + return "not-cancelled" + + releaser = threading.Thread(target=lambda: (started.wait(10), cancel_issued.wait(10), release.set()), daemon=True) + releaser.start() + try: + settled = asyncio.run(_scenario()) + finally: + release.set() + releaser.join(10) + + assert settled is True, "cancellation was delivered before the in-flight write drained" + assert done.wait(timeout=10), "the write worker never settled" + assert not out.exists(), "partial output left behind after a cancelled conversion" + + +def test_conversion_inside_file_io_worker_runs_nested_offload_inline(tmp_path, monkeypatch): + """A conversion started inside a file-IO worker must not re-enter the pool. + + Project-document conversion runs its worker body through ``run_file_io`` + and that body calls ``asyncio.run(convert_file_to_markdown(...))``, whose + own stat/write offloads used to submit to the same bounded executor — + deadlocking once every worker is busy (deterministically with a single + configured worker). ``run_file_io`` now runs nested calls inline in the + worker thread; this pins that behavior by thread identity. + """ + import threading + + from deerflow.utils import file_io + + observed: dict[str, object] = {} + + def _nested() -> str: + observed["nested_thread"] = threading.current_thread() + return "ok" + + def _worker_body() -> str: + observed["worker_thread"] = threading.current_thread() + return asyncio.run(file_io.run_file_io(_nested)) + + assert asyncio.run(file_io.run_file_io(_worker_body)) == "ok" + assert observed["nested_thread"] is observed["worker_thread"], "nested run_file_io from inside a file-io worker must run inline; submitting to the same bounded executor can deadlock conversion"