From 708f16766aed5c590a60752f1a8d2f90d475d1ce Mon Sep 17 00:00:00 2001 From: EastonJiang <3465192960@qq.com> Date: Sat, 17 Oct 2026 19:58:50 +0800 Subject: [PATCH] fix(uploads): offload document conversion for every file size (#6924) * fix(uploads): offload document conversion for every file size * test(uploads): clarify cancellation regression failures --- README.md | 5 + backend/docs/FILE_UPLOAD.md | 2 + .../harness/deerflow/projects/documents.py | 6 +- .../packages/harness/deerflow/utils/AGENTS.md | 5 +- .../harness/deerflow/utils/file_conversion.py | 31 ++----- .../test_file_conversion_offloop.py | 20 ++-- backend/tests/test_file_conversion.py | 91 +++++++++---------- .../test_file_conversion_cancellation.py | 23 ++++- 8 files changed, 96 insertions(+), 87 deletions(-) diff --git a/README.md b/README.md index 11f770982f..9535f09e76 100644 --- a/README.md +++ b/README.md @@ -2490,6 +2490,11 @@ Probe errors are logged at debug level. Sizes of 1 TB or more that can be render safely display as `>= 1 TB` to keep the prompt compact; stored byte counts remain unchanged. +When `uploads.auto_convert_documents: true` is enabled, document parsing runs +outside the Gateway event loop for every file size, including small compressed +Office files. Cancelling an upload waits for its conversion worker before +removing the worker's temporary source. Automatic conversion remains off by default. + Uploaded Markdown outlines recognize ATX heading syntax, clean closing markers with a linear suffix scan, and skip HTML comment blocks and fenced and indented code examples, so hashtags and code comments do not crowd out real document sections from the agent's heading preview. Indented bold examples are also excluded; PDF-style bold headings with up to three leading spaces remain supported. Split-bold numeric table rows, including parenthesized years, signed values, and diff --git a/backend/docs/FILE_UPLOAD.md b/backend/docs/FILE_UPLOAD.md index ebdd1b0090..d39a9d6928 100644 --- a/backend/docs/FILE_UPLOAD.md +++ b/backend/docs/FILE_UPLOAD.md @@ -116,6 +116,8 @@ DELETE /api/threads/{thread_id}/uploads/{filename} - Excel (`.xls`, `.xlsx`) - Word (`.doc`, `.docx`) +所有大小的文档都在线程池中解析，避免压缩后较小的 Office 文件长时间占用网关事件循环。取消上传时，会等待转换线程退出后再清理其临时原文件。 + 转换后的 Markdown 文件会保存在同一目录下，通常使用原文件的主干名加 `.md`；若该名称已被占用，则追加 `_N` 后缀。上传响应中的 `markdown_file` 给出实际名称。 新转换会在沙箱挂载目录之外保存原文件与转换文件的归属记录。Agent 的历史文件列表只隐藏归属已验证的转换文件；文档大纲也只读取这份记录指定的 Markdown。原文件校验包含修改时间戳，即使修改前后大小相同，也会拒绝使用旧转换结果。旧版本没有归属记录或记录缺少原文件时间戳的转换文件，升级后会作为独立 Markdown 显示，原文件也不会再从这些文件读取大纲。系统不会自动补建记录或填入当前时间戳，以免将过期转换结果认定为有效；需要大纲时，可在启用 `uploads.auto_convert_documents: true` 后重新上传原文件。原文件和转换文件都会保留。 diff --git a/backend/packages/harness/deerflow/projects/documents.py b/backend/packages/harness/deerflow/projects/documents.py index 3a99a692a9..960fa6df09 100644 --- a/backend/packages/harness/deerflow/projects/documents.py +++ b/backend/packages/harness/deerflow/projects/documents.py @@ -303,9 +303,9 @@ async def add_staged_document( def _convert_in_thread(original: Path, output_path: Path) -> Path | None: """Run the async conversion engine on a private loop inside the worker. - ``convert_file_to_markdown`` offloads its own heavy work (>1 MiB) but - stats and writes on its caller's loop; running the whole coroutine here - keeps every byte of the conversion path off the serving event loop. + Keep the shelf conversion lifecycle on its worker's private loop. + ``convert_file_to_markdown`` offloads parsing for every source size and + drains that work before this caller can clean up its staging files. """ return asyncio.run(convert_file_to_markdown(original, output_path=output_path)) diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md index eccbc8c4b9..a67b21d725 100644 --- a/backend/packages/harness/deerflow/utils/AGENTS.md +++ b/backend/packages/harness/deerflow/utils/AGENTS.md @@ -106,8 +106,9 @@ and preserve the generic `+xml` rule. ### Conversion Worker Ownership -`convert_file_to_markdown` drains large-file conversion workers across caller -cancellation before returning control to source-file cleanup. Keep conversion +`convert_file_to_markdown` offloads parsing for every source size: compressed +Office files can require seconds of work below 1 MiB. It drains conversion workers +across caller cancellation before returning control to source-file cleanup. Keep conversion inside `await_drained`, including when the worker fails or cancellation repeats; upload ingestion owns a temporary source that it removes in `finally`. Coverage lives in `tests/test_file_conversion_cancellation.py`. diff --git a/backend/packages/harness/deerflow/utils/file_conversion.py b/backend/packages/harness/deerflow/utils/file_conversion.py index b90c36f821..2334ed3ac6 100644 --- a/backend/packages/harness/deerflow/utils/file_conversion.py +++ b/backend/packages/harness/deerflow/utils/file_conversion.py @@ -8,8 +8,9 @@ PDF conversion strategy (auto mode): total when page count is unavailable), treat as image-based and fall back to MarkItDown. 3. If pymupdf4llm is not installed, use MarkItDown directly (existing behaviour). -Large files (> ASYNC_THRESHOLD_BYTES) are converted in a thread pool via -asyncio.to_thread() to avoid blocking the event loop (fixes #1569). +All document parsing runs in a thread pool via asyncio.to_thread() so compressed +Office files and small PDFs cannot block the serving event loop. Source byte +size does not bound parsing time. No FastAPI or HTTP dependencies — pure utility functions. """ @@ -40,11 +41,6 @@ CONVERTIBLE_EXTENSIONS = { ".docx", } -# Files larger than this threshold are converted in a background thread. -# Small files complete in < 1s synchronously; spawning a thread adds unnecessary -# scheduling overhead for them. -_ASYNC_THRESHOLD_BYTES = 1 * 1024 * 1024 # 1 MB - # If pymupdf4llm produces fewer characters *per page* than this threshold, # the PDF is likely image-based or encrypted — fall back to MarkItDown. # Rationale: normal text PDFs yield 200-2000 chars/page; image-based PDFs @@ -109,7 +105,7 @@ def _convert_with_markitdown(file_path: Path) -> str: def _do_convert(file_path: Path, pdf_converter: str) -> str: - """Synchronous conversion — called directly or via asyncio.to_thread. + """Synchronous conversion — run outside the calling event loop. Args: file_path: Path to the file. @@ -145,8 +141,8 @@ async def convert_file_to_markdown(file_path: Path, output_path: Path | None = N """Convert a supported document file to Markdown. PDF files are handled with a two-converter strategy (see module docstring). - Large files (> 1 MB) are offloaded to a thread pool to avoid blocking the - event loop. + Parsing is offloaded for every source size: a small compressed document + can still require seconds of synchronous work. Args: file_path: Path to the file to convert. @@ -160,17 +156,10 @@ async def convert_file_to_markdown(file_path: Path, output_path: Path | None = N """ try: pdf_converter = _get_pdf_converter() - # 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: - # Cancellation must not release the caller's source-file ownership - # while the converter is still reading it in the worker thread. - text = await await_drained(asyncio.to_thread(_do_convert, file_path, pdf_converter)) - else: - text = _do_convert(file_path, pdf_converter) + # Compressed size cannot predict parsing cost (e.g. a small XLSX can + # contain many cells). Always offload, and drain across cancellation so + # the caller retains its source until the converter finishes reading. + text = await await_drained(asyncio.to_thread(_do_convert, file_path, pdf_converter)) md_path = output_path if output_path is not None else file_path.with_suffix(".md") # Drain the write-back across caller cancellation: run_file_io awaits diff --git a/backend/tests/blocking_io/test_file_conversion_offloop.py b/backend/tests/blocking_io/test_file_conversion_offloop.py index 9188b37bea..6793674880 100644 --- a/backend/tests/blocking_io/test_file_conversion_offloop.py +++ b/backend/tests/blocking_io/test_file_conversion_offloop.py @@ -1,13 +1,13 @@ """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 +Parsing and Markdown write-back must stay off-loop for every source size. +The converted Markdown used to be 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. +converter patched to read its tiny source and return a large payload, so both +parsing IO and write-back are exercised. If the write regresses back onto the event loop, Blockbuster raises ``BlockingError`` — which the function's broad ``except`` turns into a ``None`` @@ -25,11 +25,11 @@ 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: +async def test_convert_file_to_markdown_parse_and_write_do_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) + monkeypatch.setattr(file_conversion, "_do_convert", lambda path, _converter: path.read_text(encoding="utf-8") + large_text) source = tmp_path / "doc.txt" source.write_text("source", encoding="utf-8") # test-side seeding (not in scanned_modules) @@ -37,7 +37,7 @@ async def test_convert_file_to_markdown_write_does_not_block_event_loop(tmp_path 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 + assert md_path.read_text(encoding="utf-8") == "source" + large_text async def test_cancelled_conversion_cleanup_failure_preserves_cancellation( diff --git a/backend/tests/test_file_conversion.py b/backend/tests/test_file_conversion.py index d6cd74f661..049dd795b3 100644 --- a/backend/tests/test_file_conversion.py +++ b/backend/tests/test_file_conversion.py @@ -3,12 +3,15 @@ from __future__ import annotations import asyncio +import contextvars import sys +import threading from types import ModuleType from unittest.mock import MagicMock, patch +import pytest + from deerflow.utils.file_conversion import ( - _ASYNC_THRESHOLD_BYTES, _MIN_CHARS_PER_PAGE, MAX_OUTLINE_ENTRIES, _do_convert, @@ -239,65 +242,59 @@ class TestGetPdfConverter: class TestConvertFileToMarkdown: - def test_small_file_runs_synchronously(self, tmp_path): - """Small files (< 1 MB) are converted in the event loop thread.""" - pdf = tmp_path / "small.pdf" - pdf.write_bytes(b"%PDF-1.4 " + b"x" * 100) # well under 1 MB + @pytest.mark.parametrize("file_size", [100, 1024 * 1024, 1024 * 1024 + 1], ids=["small", "former-threshold", "large"]) + def test_conversion_runs_off_loop_for_every_file_size(self, tmp_path, file_size): + """Compressed byte size cannot bound a document's parsing time.""" + source = tmp_path / "report.xlsx" + source.write_bytes(b"x" * file_size) + loop_thread = threading.get_ident() + worker_threads = [] + request_context = contextvars.ContextVar("conversion_request_context") + token = request_context.set("upload-request") - with ( - patch("deerflow.utils.file_conversion._get_pdf_converter", return_value="auto"), - patch( - "deerflow.utils.file_conversion._do_convert", - return_value="# Small PDF", - ) as mock_convert, - patch("asyncio.to_thread") as mock_thread, - ): - md_path = _run(convert_file_to_markdown(pdf)) + def convert(path, converter): + worker_threads.append(threading.get_ident()) + assert path == source + assert converter == "auto" + assert request_context.get() == "upload-request" + return "# Converted document" - # asyncio.to_thread must NOT have been called - mock_thread.assert_not_called() - mock_convert.assert_called_once() - assert md_path == pdf.with_suffix(".md") - assert md_path.read_text() == "# Small PDF" + try: + with ( + patch("deerflow.utils.file_conversion._get_pdf_converter", return_value="auto"), + patch("deerflow.utils.file_conversion._do_convert", side_effect=convert), + ): + md_path = _run(convert_file_to_markdown(source)) + finally: + request_context.reset(token) - def test_large_file_offloaded_to_thread(self, tmp_path): - """Large files (> 1 MB) are offloaded via asyncio.to_thread.""" - pdf = tmp_path / "large.pdf" - # Write slightly more than the threshold - pdf.write_bytes(b"%PDF-1.4 " + b"x" * (_ASYNC_THRESHOLD_BYTES + 1)) - - async def fake_to_thread(fn, *args, **kwargs): - return fn(*args, **kwargs) - - with ( - patch("deerflow.utils.file_conversion._get_pdf_converter", return_value="auto"), - patch( - "deerflow.utils.file_conversion._do_convert", - return_value="# Large PDF", - ), - patch("asyncio.to_thread", side_effect=fake_to_thread) as mock_thread, - ): - md_path = _run(convert_file_to_markdown(pdf)) - - mock_thread.assert_called_once() - assert md_path == pdf.with_suffix(".md") - assert md_path.read_text() == "# Large PDF" + assert len(worker_threads) == 1 + assert worker_threads[0] != loop_thread, "document parsing occupied the calling event-loop thread" + assert md_path == source.with_suffix(".md") + assert md_path.read_text(encoding="utf-8") == "# Converted document" def test_returns_none_on_conversion_error(self, tmp_path): """If conversion raises, return None without propagating the exception.""" pdf = tmp_path / "broken.pdf" pdf.write_bytes(b"%PDF-1.4 fake") + worker_threads = [] + loop_thread = threading.get_ident() + + def fail_conversion(*_args): + worker_threads.append(threading.get_ident()) + raise RuntimeError("conversion failed") + with ( patch("deerflow.utils.file_conversion._get_pdf_converter", return_value="auto"), - patch( - "deerflow.utils.file_conversion._do_convert", - side_effect=RuntimeError("conversion failed"), - ), + patch("deerflow.utils.file_conversion._do_convert", side_effect=fail_conversion), ): result = _run(convert_file_to_markdown(pdf)) assert result is None + assert len(worker_threads) == 1 + assert worker_threads[0] != loop_thread + assert not pdf.with_suffix(".md").exists() def test_writes_utf8_markdown_file(self, tmp_path): """Generated .md file is written with UTF-8 encoding.""" @@ -524,11 +521,13 @@ def test_cancelled_conversion_drains_write_and_removes_partial(tmp_path, monkeyp out = tmp_path / "doc.md" src.write_text("hello") + # Isolate write cancellation from parser imports and conversion latency. + monkeypatch.setattr("deerflow.utils.file_conversion._do_convert", lambda _path, _converter: "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) + assert await asyncio.to_thread(started.wait, 10), "the write worker never started" 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 diff --git a/backend/tests/test_file_conversion_cancellation.py b/backend/tests/test_file_conversion_cancellation.py index b67c8935a3..6df9c4f9e8 100644 --- a/backend/tests/test_file_conversion_cancellation.py +++ b/backend/tests/test_file_conversion_cancellation.py @@ -11,10 +11,11 @@ from deerflow.utils import file_conversion @pytest.mark.asyncio +@pytest.mark.parametrize("file_size", [100, 1024 * 1024 + 1], ids=["small", "large"]) @pytest.mark.parametrize("conversion_fails", [False, True], ids=["success", "failure"]) -async def test_cancelled_conversion_drains_worker_before_source_cleanup(tmp_path, monkeypatch, conversion_fails): - source = tmp_path / "large.pdf" - source.write_bytes(b"x" * (file_conversion._ASYNC_THRESHOLD_BYTES + 1)) +async def test_cancelled_conversion_drains_worker_before_source_cleanup(tmp_path, monkeypatch, conversion_fails, file_size): + source = tmp_path / "document.pdf" + source.write_bytes(b"x" * file_size) output = tmp_path / "converted.md" started = threading.Event() release = threading.Event() @@ -24,11 +25,18 @@ async def test_cancelled_conversion_drains_worker_before_source_cleanup(tmp_path loop = asyncio.get_running_loop() worker_started = asyncio.Event() + loop_thread = threading.get_ident() + worker_threads = [] + def convert(path, _converter): + worker_threads.append(threading.get_ident()) started.set() loop.call_soon_threadsafe(worker_started.set) try: - assert release.wait(timeout=10), "test did not release conversion worker" + # An inline regression must reach the test body's diagnostic + # instead of parking the event loop waiting for its own release. + if worker_threads[0] != loop_thread: + assert release.wait(timeout=10), "test did not release conversion worker" # A converter may reopen/read the source throughout conversion. assert path.read_bytes().startswith(b"x") source_read.set() @@ -53,6 +61,8 @@ async def test_cancelled_conversion_drains_worker_before_source_cleanup(tmp_path try: await asyncio.wait_for(worker_started.wait(), timeout=10) assert started.is_set() + assert len(worker_threads) == 1 + assert worker_threads[0] != loop_thread, "converter must not block the event loop" task.cancel() # Yield scheduler turns, not a wall-clock delay, so cancellation reaches # the suspended await while the worker is held by the release event. @@ -61,11 +71,14 @@ async def test_cancelled_conversion_drains_worker_before_source_cleanup(tmp_path await asyncio.sleep(0) assert not cleanup_before_finish, "caller cleaned the source while the converter was still running" assert not task.done() - finally: release.set() with pytest.raises(asyncio.CancelledError): await task assert await asyncio.to_thread(finished.wait, 10) + finally: + release.set() + task.cancel() + await asyncio.gather(task, return_exceptions=True) assert cleanup_before_finish == [False] assert source_read.is_set()