diff --git a/backend/README.md b/backend/README.md index dc6122e2b0..fc0d68b664 100644 --- a/backend/README.md +++ b/backend/README.md @@ -148,6 +148,11 @@ FastAPI application providing REST endpoints for frontend integration: | `DELETE /api/threads/{id}` | Delete DeerFlow-managed local thread data after LangGraph thread deletion; unexpected failures are logged server-side and return a generic 500 detail | | `GET /api/threads/{id}/artifacts/{path}` | Serve generated artifacts | +Cancelling an upload waits for an already-running document conversion worker to +finish before removing its temporary source. This prevents cleanup from deleting +a file that the converter is still reading; cancellation can therefore take as +long as that conversion. + Converted-upload ownership records live in each thread's `upload-companions/` directory, outside the sandbox-mounted `user-data/` tree. Older conversions without a record remain separate Markdown uploads and no longer provide an diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md index 1e540d9165..447abbe609 100644 --- a/backend/packages/harness/deerflow/utils/AGENTS.md +++ b/backend/packages/harness/deerflow/utils/AGENTS.md @@ -39,6 +39,14 @@ boundary for artifacts and project documents. Keep platform aliases such as Windows' `image/svg` aligned with their standard active type (`image/svg+xml`), 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 +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`. + ### Host Path Portability `host_paths.py` rejects Windows device names for host-visible creation paths on diff --git a/backend/packages/harness/deerflow/utils/file_conversion.py b/backend/packages/harness/deerflow/utils/file_conversion.py index ac7c74508f..b90c36f821 100644 --- a/backend/packages/harness/deerflow/utils/file_conversion.py +++ b/backend/packages/harness/deerflow/utils/file_conversion.py @@ -166,7 +166,9 @@ async def convert_file_to_markdown(file_path: Path, output_path: Path | None = N 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) + # 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) diff --git a/backend/tests/test_file_conversion_cancellation.py b/backend/tests/test_file_conversion_cancellation.py new file mode 100644 index 0000000000..b67c8935a3 --- /dev/null +++ b/backend/tests/test_file_conversion_cancellation.py @@ -0,0 +1,73 @@ +"""A cancelled conversion must retain its source until its worker exits.""" + +from __future__ import annotations + +import asyncio +import threading + +import pytest + +from deerflow.utils import file_conversion + + +@pytest.mark.asyncio +@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)) + output = tmp_path / "converted.md" + started = threading.Event() + release = threading.Event() + finished = threading.Event() + source_read = threading.Event() + cleanup_before_finish = [] + loop = asyncio.get_running_loop() + worker_started = asyncio.Event() + + def convert(path, _converter): + started.set() + loop.call_soon_threadsafe(worker_started.set) + try: + 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() + if conversion_fails: + raise ValueError("synthetic converter failure") + return "# Converted document" + finally: + finished.set() + + monkeypatch.setattr(file_conversion, "_get_pdf_converter", lambda: "auto") + monkeypatch.setattr(file_conversion, "_do_convert", convert) + + async def ingest(): + try: + return await file_conversion.convert_file_to_markdown(source, output_path=output) + finally: + # UploadIngestionSession removes its private source in a finally. + cleanup_before_finish.append(not finished.is_set()) + source.unlink(missing_ok=True) + + task = asyncio.create_task(ingest()) + try: + await asyncio.wait_for(worker_started.wait(), timeout=10) + assert started.is_set() + 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. + await asyncio.sleep(0) + task.cancel() + 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) + + assert cleanup_before_finish == [False] + assert source_read.is_set() + assert not source.exists() + assert not output.exists()