diff --git a/backend/packages/harness/deerflow/uploads/__init__.py b/backend/packages/harness/deerflow/uploads/__init__.py index 028be2c46..6b360d618 100644 --- a/backend/packages/harness/deerflow/uploads/__init__.py +++ b/backend/packages/harness/deerflow/uploads/__init__.py @@ -1,4 +1,3 @@ -from .conversion import convert_uploaded_file_to_markdown from .errors import AtomicUploadPublishError, PathTraversalError, UnsafeUploadPathError from .layout import ( UPLOAD_CONVERSIONS_DIRNAME, @@ -6,6 +5,7 @@ from .layout import ( UnsafeConversionPathError, artifact_url_for_virtual_path, conversion_dir_for_uploads, + conversion_filename_for_upload, conversion_path_for_upload, conversion_virtual_path, ensure_conversion_dir, @@ -44,12 +44,22 @@ from .manager import ( validate_thread_id, ) + +def __getattr__(name: str): + if name == "convert_uploaded_file_to_markdown": + from .conversion import convert_uploaded_file_to_markdown + + return convert_uploaded_file_to_markdown + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + + __all__ = [ "get_uploads_dir", "UPLOAD_CONVERSIONS_DIRNAME", "UPLOAD_LOCKS_DIRNAME", "UnsafeConversionPathError", "conversion_dir_for_uploads", + "conversion_filename_for_upload", "conversion_path_for_upload", "conversion_virtual_path", "artifact_url_for_virtual_path", diff --git a/backend/packages/harness/deerflow/uploads/conversion.py b/backend/packages/harness/deerflow/uploads/conversion.py index a12074ae6..b49d2995a 100644 --- a/backend/packages/harness/deerflow/uploads/conversion.py +++ b/backend/packages/harness/deerflow/uploads/conversion.py @@ -1,34 +1,178 @@ """Safe publication of Markdown generated from primary uploads.""" +import asyncio +import logging +import os +import stat +from dataclasses import dataclass from pathlib import Path +from deerflow.uploads.errors import UnsafeUploadPathError from deerflow.uploads.layout import ( UnsafeConversionPathError, conversion_path_for_upload, ensure_conversion_dir, ) +from deerflow.uploads.lease import UploadIdentity, UploadNameLease from deerflow.uploads.manager import ( + PublishedUpload, + StagedUpload, abort_staged_upload, create_upload_staging_file, replace_system_owned_staged_file, ) from deerflow.utils.file_conversion import convert_file_to_markdown +logger = logging.getLogger(__name__) -async def convert_uploaded_file_to_markdown(upload_path: Path) -> Path | None: - """Convert one primary upload and atomically publish its owned Markdown.""" - conversion_dir = ensure_conversion_dir(upload_path.parent) - target = conversion_path_for_upload(upload_path) - staged = create_upload_staging_file(conversion_dir) - staged.handle.close() + +@dataclass(slots=True) +class _PreparedConversion: + publication: PublishedUpload + staged: StagedUpload + target: Path + release_publication: bool + + +def _abort_stage_without_masking(staged: StagedUpload) -> None: try: - result = await convert_file_to_markdown(upload_path, output_path=staged.path) - if result is None: - abort_staged_upload(staged) - return None - if Path(result) != staged.path: - raise UnsafeConversionPathError("Converter returned an unexpected output path") - return replace_system_owned_staged_file(staged, target.name) - except Exception: abort_staged_upload(staged) + except BaseException: + logger.warning("Failed to clean up upload conversion staging file: %s", staged.path, exc_info=True) + + +def _release_publication_without_masking(publication: PublishedUpload) -> None: + try: + publication.release() + except BaseException: + logger.warning("Failed to release upload conversion lease: %s", publication.lease.lock_path, exc_info=True) + + +def _validate_publication(upload_path: Path, publication: PublishedUpload) -> None: + if publication.path != upload_path: + raise UnsafeUploadPathError("Upload publication does not match the conversion source") + if publication.lease.filename != upload_path.name or publication.lease.uploads_dir != upload_path.parent: + raise UnsafeUploadPathError("Upload publication lease does not match the conversion source") + if not publication.is_active: + raise UnsafeUploadPathError("Upload publication lease was already released") + if not publication.identity.matches(upload_path): + raise UnsafeUploadPathError("Upload generation changed before conversion") + upload_stat = os.lstat(upload_path) + if not stat.S_ISREG(upload_stat.st_mode) or upload_stat.st_nlink != 1: + raise UnsafeUploadPathError("Upload conversion source is not an exclusive regular file") + + +def _discard_prepared_conversion(prepared: _PreparedConversion) -> None: + _abort_stage_without_masking(prepared.staged) + if prepared.release_publication: + _release_publication_without_masking(prepared.publication) + + +def _prepare_conversion(upload_path: Path, publication: PublishedUpload | None) -> _PreparedConversion: + release_publication = publication is None + if publication is None: + lease = UploadNameLease.acquire(upload_path.parent, upload_path.name) + try: + publication = PublishedUpload( + path=upload_path, + identity=UploadIdentity.from_path(upload_path), + lease=lease, + ) + except BaseException: + try: + lease.release() + except BaseException: + logger.warning("Failed to release upload conversion lease: %s", lease.lock_path, exc_info=True) + raise + + staged: StagedUpload | None = None + try: + _validate_publication(upload_path, publication) + conversion_dir = ensure_conversion_dir(upload_path.parent) + target = conversion_path_for_upload(upload_path) + staged = create_upload_staging_file(conversion_dir) + staged.handle.close() + return _PreparedConversion( + publication=publication, + staged=staged, + target=target, + release_publication=release_publication, + ) + except BaseException: + if staged is not None: + _abort_stage_without_masking(staged) + if release_publication: + _release_publication_without_masking(publication) raise + + +def _publish_prepared_conversion(prepared: _PreparedConversion, result: Path) -> Path: + _validate_publication(prepared.publication.path, prepared.publication) + if result != prepared.staged.path: + raise UnsafeConversionPathError("Converter returned an unexpected output path") + return replace_system_owned_staged_file(prepared.staged, prepared.target.name) + + +async def _prepare_conversion_cancellation_safe( + upload_path: Path, + publication: PublishedUpload | None, +) -> _PreparedConversion: + prepare_task = asyncio.create_task( + asyncio.to_thread(_prepare_conversion, upload_path, publication), + name=f"prepare-upload-conversion:{upload_path.name}", + ) + try: + return await asyncio.shield(prepare_task) + except asyncio.CancelledError: + while not prepare_task.done(): + try: + await asyncio.shield(prepare_task) + except asyncio.CancelledError: + continue + except Exception: + break + if not prepare_task.cancelled() and prepare_task.exception() is None: + cleanup_task = asyncio.create_task( + asyncio.to_thread(_discard_prepared_conversion, prepare_task.result()), + name=f"discard-upload-conversion:{upload_path.name}", + ) + while not cleanup_task.done(): + try: + await asyncio.shield(cleanup_task) + except asyncio.CancelledError: + continue + raise + + +async def _run_cleanup_off_thread(function, *args) -> None: + cleanup_task = asyncio.create_task(asyncio.to_thread(function, *args)) + while not cleanup_task.done(): + try: + await asyncio.shield(cleanup_task) + except asyncio.CancelledError: + continue + cleanup_task.result() + + +async def convert_uploaded_file_to_markdown( + upload_path: Path, + *, + publication: PublishedUpload | None = None, +) -> Path | None: + """Convert one primary generation and atomically publish its owned Markdown.""" + prepared = await _prepare_conversion_cancellation_safe(upload_path, publication) + stage_consumed = False + try: + result = await convert_file_to_markdown(upload_path, output_path=prepared.staged.path) + if result is None: + await _run_cleanup_off_thread(_abort_stage_without_masking, prepared.staged) + stage_consumed = True + return None + converted = await asyncio.to_thread(_publish_prepared_conversion, prepared, Path(result)) + stage_consumed = True + return converted + finally: + if not stage_consumed: + await _run_cleanup_off_thread(_abort_stage_without_masking, prepared.staged) + if prepared.release_publication: + await _run_cleanup_off_thread(_release_publication_without_masking, prepared.publication) diff --git a/backend/packages/harness/deerflow/uploads/layout.py b/backend/packages/harness/deerflow/uploads/layout.py index 788bf994f..0f008d092 100644 --- a/backend/packages/harness/deerflow/uploads/layout.py +++ b/backend/packages/harness/deerflow/uploads/layout.py @@ -1,5 +1,6 @@ """Path and URL layout helpers for primary uploads and generated assets.""" +import hashlib import os import stat from pathlib import Path @@ -20,9 +21,27 @@ def conversion_dir_for_uploads(uploads_dir: Path) -> Path: return uploads_dir.parent / UPLOAD_CONVERSIONS_DIRNAME +def _truncate_utf8(value: str, max_bytes: int) -> str: + encoded = value.encode("utf-8") + if len(encoded) <= max_bytes: + return value + return encoded[:max_bytes].decode("utf-8", errors="ignore") + + +def conversion_filename_for_upload(filename: str) -> str: + """Return a deterministic generated-Markdown component within 255 bytes.""" + desired = f"{filename}.md" + if len(desired.encode("utf-8")) <= 255: + return desired + digest = hashlib.sha256(filename.encode("utf-8")).hexdigest() + marker = f".{digest}.md" + prefix = _truncate_utf8(filename, 255 - len(marker.encode("utf-8"))) + return f"{prefix}{marker}" + + def conversion_path_for_upload(upload_path: Path) -> Path: """Return the generated Markdown path owned by one primary upload.""" - return conversion_dir_for_uploads(upload_path.parent) / f"{upload_path.name}.md" + return conversion_dir_for_uploads(upload_path.parent) / conversion_filename_for_upload(upload_path.name) def validate_conversion_dir(uploads_dir: Path) -> Path | None: @@ -93,7 +112,7 @@ def upload_virtual_path(filename: str) -> str: def conversion_virtual_path(filename: str) -> str: """Build the sandbox virtual path for an upload's generated Markdown.""" - return f"{VIRTUAL_PATH_PREFIX}/{UPLOAD_CONVERSIONS_DIRNAME}/{filename}.md" + return f"{VIRTUAL_PATH_PREFIX}/{UPLOAD_CONVERSIONS_DIRNAME}/{conversion_filename_for_upload(filename)}" def artifact_url_for_virtual_path(thread_id: str, virtual_path: str) -> str: diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index f5c05ab58..e592cf720 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -20,6 +20,7 @@ from deerflow.runtime.user_context import get_effective_user_id from deerflow.uploads.errors import AtomicUploadPublishError, PathTraversalError, UnsafeUploadPathError from deerflow.uploads.layout import ( UPLOAD_CONVERSIONS_DIRNAME, + _truncate_utf8, artifact_url_for_virtual_path, existing_conversion_path_for_upload, upload_virtual_path, @@ -192,13 +193,6 @@ def _validate_staged_upload(staged: StagedUpload) -> None: raise UnsafeUploadPathError("Upload staging path is not an exclusive regular file") -def _truncate_utf8(value: str, max_bytes: int) -> str: - encoded = value.encode("utf-8") - if len(encoded) <= max_bytes: - return value - return encoded[:max_bytes].decode("utf-8", errors="ignore") - - def _filename_candidates(name: str) -> Iterator[str]: """Yield collision candidates that stay within the filename byte limit.""" yield name diff --git a/backend/packages/harness/deerflow/utils/file_conversion.py b/backend/packages/harness/deerflow/utils/file_conversion.py index 53cec1a9a..a54e5f981 100644 --- a/backend/packages/harness/deerflow/utils/file_conversion.py +++ b/backend/packages/harness/deerflow/utils/file_conversion.py @@ -8,8 +8,8 @@ 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). +Parser work and generated-Markdown writes run in a worker thread so document +conversion never blocks an async caller's event loop. No FastAPI or HTTP dependencies — pure utility functions. """ @@ -39,11 +39,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 @@ -140,12 +135,18 @@ def _do_convert(file_path: Path, pdf_converter: str) -> str: return _convert_with_markitdown(file_path) +def _convert_file_to_markdown_sync(file_path: Path, output_path: Path | None) -> Path: + pdf_converter = _get_pdf_converter() + 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") + return md_path + + async def convert_file_to_markdown(file_path: Path, output_path: Path | None = None) -> Path | None: - """Convert a supported document file to Markdown. + """Convert a supported document file to Markdown off the event loop. 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. Args: file_path: Path to the file to convert. @@ -158,18 +159,9 @@ async def convert_file_to_markdown(file_path: Path, output_path: Path | None = N Path to the generated .md file, or None if conversion failed. """ try: - pdf_converter = _get_pdf_converter() - file_size = file_path.stat().st_size + md_path = await asyncio.to_thread(_convert_file_to_markdown_sync, file_path, output_path) - if file_size > _ASYNC_THRESHOLD_BYTES: - text = await asyncio.to_thread(_do_convert, file_path, pdf_converter) - else: - 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") - - logger.info("Converted %s to markdown: %s (%d chars)", file_path.name, md_path.name, len(text)) + logger.info("Converted %s to markdown: %s", file_path.name, md_path.name) return md_path except Exception as e: logger.error("Failed to convert %s to markdown: %s", file_path.name, e) diff --git a/backend/tests/blocking_io/test_upload_conversion.py b/backend/tests/blocking_io/test_upload_conversion.py new file mode 100644 index 000000000..dfd34f675 --- /dev/null +++ b/backend/tests/blocking_io/test_upload_conversion.py @@ -0,0 +1,27 @@ +"""Regression anchor: upload conversion lifecycle must not block the event loop.""" + +from __future__ import annotations + +import asyncio + +import pytest + +from deerflow.uploads.conversion import convert_uploaded_file_to_markdown + +pytestmark = pytest.mark.asyncio + + +async def test_real_upload_conversion_lifecycle_does_not_block_event_loop(tmp_path, monkeypatch): + uploads = tmp_path / "user-data" / "uploads" + await asyncio.to_thread(uploads.mkdir, parents=True) + source = uploads / "report.pdf" + await asyncio.to_thread(source.write_bytes, b"PDF") + monkeypatch.setattr( + "deerflow.utils.file_conversion._do_convert", + lambda path, converter: "# converted", + ) + + result = await convert_uploaded_file_to_markdown(source) + + assert result is not None + assert await asyncio.to_thread(result.read_text, encoding="utf-8") == "# converted" diff --git a/backend/tests/test_file_conversion.py b/backend/tests/test_file_conversion.py index 407b8ae12..96702e605 100644 --- a/backend/tests/test_file_conversion.py +++ b/backend/tests/test_file_conversion.py @@ -8,7 +8,6 @@ from types import ModuleType from unittest.mock import MagicMock, patch from deerflow.utils.file_conversion import ( - _ASYNC_THRESHOLD_BYTES, _MIN_CHARS_PER_PAGE, MAX_OUTLINE_ENTRIES, _do_convert, @@ -239,23 +238,25 @@ class TestGetPdfConverter: class TestConvertFileToMarkdown: - def test_small_file_runs_synchronously(self, tmp_path): - """Small files (< 1 MB) are converted in the event loop thread.""" + def test_small_file_is_offloaded_to_thread(self, tmp_path): + """Small files are offloaded so parser and write IO cannot block the event loop.""" pdf = tmp_path / "small.pdf" pdf.write_bytes(b"%PDF-1.4 " + b"x" * 100) # well under 1 MB + 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="# Small PDF", ) as mock_convert, - patch("asyncio.to_thread") as mock_thread, + patch("asyncio.to_thread", side_effect=fake_to_thread) as mock_thread, ): md_path = _run(convert_file_to_markdown(pdf)) - # asyncio.to_thread must NOT have been called - mock_thread.assert_not_called() + mock_thread.assert_called_once() mock_convert.assert_called_once() assert md_path == pdf.with_suffix(".md") assert md_path.read_text() == "# Small PDF" @@ -263,8 +264,7 @@ class TestConvertFileToMarkdown: 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)) + pdf.write_bytes(b"%PDF-1.4 " + b"x" * (1024 * 1024 + 1)) async def fake_to_thread(fn, *args, **kwargs): return fn(*args, **kwargs) diff --git a/backend/tests/test_upload_conversion.py b/backend/tests/test_upload_conversion.py index 75ca7f344..694a4f333 100644 --- a/backend/tests/test_upload_conversion.py +++ b/backend/tests/test_upload_conversion.py @@ -1,12 +1,100 @@ """Tests for publication of system-owned Markdown upload conversions.""" +import asyncio +import hashlib from pathlib import Path from unittest.mock import AsyncMock, patch import pytest from deerflow.uploads.conversion import convert_uploaded_file_to_markdown -from deerflow.uploads.layout import conversion_path_for_upload +from deerflow.uploads.layout import ( + conversion_filename_for_upload, + conversion_path_for_upload, + conversion_virtual_path, + existing_conversion_path_for_upload, +) +from deerflow.uploads.manager import delete_file_safe, publish_upload_bytes, publish_upload_bytes_leased + + +@pytest.mark.asyncio +async def test_delete_and_reupload_cannot_receive_old_conversion(tmp_path): + uploads = tmp_path / "user-data" / "uploads" + uploads.mkdir(parents=True) + publication = publish_upload_bytes_leased(uploads, "report.pdf", b"OLD") + converter_started = asyncio.Event() + allow_converter = asyncio.Event() + + async def paused_convert(source, output_path=None): + converter_started.set() + await allow_converter.wait() + output_path.write_text("FROM OLD", encoding="utf-8") + return output_path + + with patch("deerflow.uploads.conversion.convert_file_to_markdown", side_effect=paused_convert): + conversion = asyncio.create_task(convert_uploaded_file_to_markdown(publication.path, publication=publication)) + await converter_started.wait() + deletion = asyncio.create_task(asyncio.to_thread(delete_file_safe, uploads, "report.pdf")) + await asyncio.sleep(0.05) + assert not deletion.done() + allow_converter.set() + await conversion + publication.release() + await deletion + + replacement = publish_upload_bytes(uploads, "report.pdf", b"NEW") + assert replacement.read_bytes() == b"NEW" + assert existing_conversion_path_for_upload(replacement) is None + + +@pytest.mark.asyncio +async def test_cancelled_waiter_does_not_leak_name_lease_or_conversion_stage(tmp_path): + uploads = tmp_path / "user-data" / "uploads" + uploads.mkdir(parents=True) + publication = publish_upload_bytes_leased(uploads, "report.pdf", b"OLD") + conversion = asyncio.create_task(convert_uploaded_file_to_markdown(publication.path)) + await asyncio.sleep(0.05) + + conversion.cancel() + await asyncio.sleep(0.05) + assert not conversion.done() + publication.release() + + with pytest.raises(asyncio.CancelledError): + await conversion + + delete_file_safe(uploads, "report.pdf") + replacement = publish_upload_bytes(uploads, "report.pdf", b"NEW") + assert replacement.read_bytes() == b"NEW" + conversion_dir = uploads.parent / ".upload-conversions" + assert not list(conversion_dir.glob(".upload-*.part")) + + +@pytest.mark.parametrize("byte_length", [252, 253, 254, 255]) +def test_long_conversion_filename_fits_component_limit(byte_length, tmp_path): + filename = "a" * (byte_length - 4) + ".pdf" + upload = tmp_path / "uploads" / filename + target = conversion_path_for_upload(upload) + + assert len(target.name.encode("utf-8")) <= 255 + assert target.name == conversion_filename_for_upload(filename) + assert conversion_virtual_path(filename).endswith(f"/{target.name}") + assert target.name.endswith(".md") + + +def test_255_byte_conversion_name_uses_full_digest(): + filename = "a" * 251 + ".pdf" + digest = hashlib.sha256(filename.encode("utf-8")).hexdigest() + + assert conversion_filename_for_upload(filename) == f"{'a' * 187}.{digest}.md" + + +def test_multibyte_long_conversion_name_is_utf8_safe(): + filename = "é" * 125 + ".pdf" + converted = conversion_filename_for_upload(filename) + + assert len(converted.encode("utf-8")) <= 255 + assert converted.endswith(".md") @pytest.mark.asyncio diff --git a/docs/superpowers/plans/2026-08-06-upload-review-remediation.md b/docs/superpowers/plans/2026-08-06-upload-review-remediation.md index d6da5224a..e867be285 100644 --- a/docs/superpowers/plans/2026-08-06-upload-review-remediation.md +++ b/docs/superpowers/plans/2026-08-06-upload-review-remediation.md @@ -322,7 +322,7 @@ def test_long_conversion_filename_fits_component_limit(byte_length, tmp_path): assert target.name.endswith(".md") ``` -Use literal expected output for the 255-byte case: 186 `a` bytes, a dot, the full SHA-256 +Use literal expected output for the 255-byte case: 187 `a` bytes, a dot, the full SHA-256 of the original filename, and `.md`. - [ ] **Step 2: Write a strict blocking-I/O regression**