From 4a92cb579acfb6a7614fe17670abf9315ade81c5 Mon Sep 17 00:00:00 2001 From: yetuge <182123929+yetuge@users.noreply.github.com> Date: Fri, 25 Sep 2026 14:22:34 +0800 Subject: [PATCH] fix(uploads): keep document-conversion stat and write-back off the event loop (#5830) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(uploads): keep document-conversion stat and write-back off the event loop convert_file_to_markdown offloads the conversion itself for files above 1 MB (documented in its docstring), but the two bookend filesystem calls still ran on the event loop: the size stat before the threshold check and the write of the converted markdown afterwards — the latter a potentially multi-megabyte write in the per-upload ingestion path. Route both through run_file_io (the dedicated file IO pool that also keeps the caller's context), matching the off-the-loop fixes already applied to the sibling upload/document paths. Add a tests/blocking_io anchor: with the converter patched to a 2 MB payload it fails on the previous head (Blockbuster catches the blocking os.stat first; the broad except swallows it into a None return) and passes with this change. * fix(uploads): address review — drain conversion write-back, make run_file_io re-entrant Review follow-up on the off-loop conversion stat/write PR: 1. The write-back now goes through await_drained(): a plain run_file_io await cancels immediately while the worker keeps writing, so the ingestion scope could remove its private directory over an in-flight write and leave a large hidden partial file. The drain delivers the cancellation only after the worker settled, and the handler removes the partial output before re-raising. 2. run_file_io is now re-entrant: a nested offload issued from code already running inside a file-io worker (project-document conversion runs asyncio.run(convert_file_to_markdown(...)) inside a run_file_io worker) executes inline instead of submitting to the same bounded executor, which deadlocked with a single configured worker. Both fixes carry deterministic regressions: the cancellation test pins the drain ordering (cancel issued before the worker is released) and the partial-file cleanup; the re-entrancy test pins inline execution by thread identity. Both fail on the previous head. Full blocking_io + project-document + conversion suites show an identical failure set to the previous head (pre-existing Windows platform failures only). * fix(uploads): keep cancellation authoritative when conversion cleanup fails If the post-cancellation unlink of the staging output raised, the new exception escaped the CancelledError handler, was swallowed by the broad except as an ordinary conversion failure (None return), and left both the caller's cancellation and the partial file behind. Catch and log cleanup failures inside the handler and always re-raise the captured cancellation; a second cancellation still passes through untouched (CancelledError is not an Exception). Red/green: tests/blocking_io/test_file_conversion_offloop.py gains a regression where unlink fails while the write-back is in flight — on the reviewed head the task settled on None (DID NOT RAISE); with the change the CancelledError propagates and the failure is logged. Signed-off-by: yetuge <2219677952@qq.com> --------- Signed-off-by: yetuge <2219677952@qq.com> --- .../harness/deerflow/utils/file_conversion.py | 24 ++++- .../harness/deerflow/utils/file_io.py | 31 +++++- .../test_file_conversion_offloop.py | 95 ++++++++++++++++++ backend/tests/test_file_conversion.py | 97 +++++++++++++++++++ 4 files changed, 243 insertions(+), 4 deletions(-) create mode 100644 backend/tests/blocking_io/test_file_conversion_offloop.py 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"