diff --git a/backend/app/gateway/routers/uploads.py b/backend/app/gateway/routers/uploads.py index 3e152e272..641e58039 100644 --- a/backend/app/gateway/routers/uploads.py +++ b/backend/app/gateway/routers/uploads.py @@ -233,7 +233,13 @@ def _prepare_upload_destination(uploads_dir: os.PathLike[str] | str, display_fil return _UploadTempFile(file_path=file_path, temp_path=temp_path, handle=handle) -def _link_staged_no_overwrite(staged_path: Path, uploads_dir: os.PathLike[str] | str, display_filename: str) -> Path: +def _link_staged_no_overwrite( + staged_path: Path, + uploads_dir: os.PathLike[str] | str, + display_filename: str, + *, + unlink_staged: bool = True, +) -> Path: """Worker: publish *staged_path* under *display_filename* atomically, never overwriting. The ``os.link`` itself is the whole no-overwrite guard: it fails with @@ -250,6 +256,12 @@ def _link_staged_no_overwrite(staged_path: Path, uploads_dir: os.PathLike[str] | plain collision to retry. Any other failure removes the staged file and propagates; success unlinks it. Staging and destination are co-located in the uploads dir, so the hard link is always same-filesystem. + + ``unlink_staged=False`` publishes the link but leaves the staged name in + place, for a caller that still holds a descriptor on the staged inode and + therefore must remove it itself. Windows refuses to remove a file that + has an open handle, so the removal cannot happen here in that case; the + caller owns the staged path from the moment this returns. """ file_path = _pure_destination(uploads_dir, display_filename) try: @@ -267,20 +279,44 @@ def _link_staged_no_overwrite(staged_path: Path, uploads_dir: os.PathLike[str] | except FileNotFoundError: pass raise - os.unlink(staged_path) + if unlink_staged: + os.unlink(staged_path) return file_path -def _commit_upload_temp_no_overwrite(upload_temp: _UploadTempFile, uploads_dir: os.PathLike[str] | str, display_filename: str) -> Path: +def _commit_upload_temp_no_overwrite( + upload_temp: _UploadTempFile, + uploads_dir: os.PathLike[str] | str, + display_filename: str, + *, + unlink_staged: bool = True, +) -> Path: """Worker: close the staged handle and publish the ``.part`` atomically via ``os.link``. Same no-overwrite contract as :func:`_link_staged_no_overwrite`: :class:`FileExistsError` leaves the staged part in place for a next-suffix retry (the handle's second ``close`` is idempotent); any - other failure removes it. + other failure removes it. ``unlink_staged=False`` hands the staged name + back to the caller, which still holds a descriptor on its inode. """ upload_temp.handle.close() - return _link_staged_no_overwrite(upload_temp.temp_path, uploads_dir, display_filename) + return _link_staged_no_overwrite(upload_temp.temp_path, uploads_dir, display_filename, unlink_staged=unlink_staged) + + +def _remove_staged_file(staged_path: os.PathLike[str] | str) -> None: + """Worker: remove a staged ``.part`` name whose inode is no longer held open. + + The deferred half of ``unlink_staged=False``. Removal is best-effort: the + name is already unreachable through the uploads listing, and failing an + otherwise-committed upload over leftover staging bytes would be worse than + leaving them for the startup sweep. + """ + try: + os.unlink(staged_path) + except FileNotFoundError: + pass + except OSError: + logger.warning("Failed to remove staged upload file: %s", staged_path, exc_info=True) def _write_upload_chunk(upload_temp: _UploadTempFile, chunk: bytes) -> None: @@ -291,10 +327,11 @@ def _abort_upload_temp(upload_temp: _UploadTempFile) -> None: try: upload_temp.handle.close() finally: - try: - os.unlink(upload_temp.temp_path) - except FileNotFoundError: - pass + # Best-effort, not ``os.unlink``: an abandoned duplication worker can + # still hold this inode open (Windows refuses to remove it then), and + # raising here would replace the caller's cancellation or original + # failure with a secondary permission error. + _remove_staged_file(upload_temp.temp_path) def _make_uploaded_paths_sandbox_readable(paths: list[os.PathLike[str] | str]) -> None: diff --git a/backend/app/gateway/upload_ingestion.py b/backend/app/gateway/upload_ingestion.py index 00715852d..292bc0c54 100644 --- a/backend/app/gateway/upload_ingestion.py +++ b/backend/app/gateway/upload_ingestion.py @@ -249,6 +249,7 @@ class ThreadUploadIngestionService: file_size = 0 upload_temp = None convert_source_fd: int | None = None + deferred_staged_path: Path | None = None try: upload_temp = await run_file_io(uploads._prepare_upload_destination, self._uploads_dir, safe_filename) async for chunk in chunks: @@ -269,12 +270,26 @@ class ThreadUploadIngestionService: # Link-commit with collision retry: the FileExistsError arm # leaves the staged part in place for the retry under the next # suffix (the handle's second close is idempotent). + # + # The conversion descriptor above keeps the staged inode open, and + # Windows refuses to remove a file that still has an open handle. + # Publishing is therefore split from removing the staged name: + # this request keeps ownership of the staged path and removes it + # once the descriptor is released below. while True: try: - file_path = await run_file_io(uploads._commit_upload_temp_no_overwrite, upload_temp, self._uploads_dir, safe_filename) + file_path = await run_file_io( + uploads._commit_upload_temp_no_overwrite, + upload_temp, + self._uploads_dir, + safe_filename, + unlink_staged=convert_source_fd is None, + ) break except FileExistsError: safe_filename = uploads.claim_unique_filename(safe_filename, self._seen_filenames) + if convert_source_fd is not None: + deferred_staged_path = upload_temp.temp_path upload_temp = None except uploads.UnsafeUploadPathError as exc: _close_fd(convert_source_fd) @@ -343,6 +358,10 @@ class ThreadUploadIngestionService: raise finally: _close_fd(convert_source_fd) + if deferred_staged_path is not None: + # The descriptor that pinned the staged inode is released + # by now, so the deferred staged-name removal can land. + await run_file_io(uploads._remove_staged_file, deferred_staged_path) if private_dir is not None: await run_file_io(shutil.rmtree, private_dir, True) if not md_staged: diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py index 0bc1fbf48..eef55c68b 100644 --- a/backend/tests/test_uploads_router.py +++ b/backend/tests/test_uploads_router.py @@ -372,6 +372,77 @@ def test_upload_files_makes_non_local_files_sandbox_writable(tmp_path): make_writable.assert_any_call(thread_uploads_dir / "report.md") +def test_upload_files_removes_the_staged_part_when_the_platform_blocks_an_open_unlink(tmp_path, monkeypatch): + """A staged ``.part`` that is still open must neither fail the upload nor leak. + + Windows refuses to remove a file that still has an open handle, and the + conversion path deliberately holds a descriptor on the staged inode across + the link commit (that descriptor is what keeps conversion reading the bytes + this request wrote). Reproduce the resulting sharing violation portably by + pinning the staged name for as long as the descriptor lives: the commit's + own removal attempt would raise ``PermissionError``, and the deferred one + after the descriptor is released succeeds. + """ + thread_uploads_dir = tmp_path / "uploads" + thread_uploads_dir.mkdir(parents=True) + + provider = MagicMock() + provider.uses_thread_data_mounts = False + provider.acquire.side_effect = AssertionError("upload route should use acquire_async") + provider.acquire_async = AsyncMock(return_value="aio-1") + sandbox = MagicMock() + provider.get.return_value = sandbox + + async def fake_convert(file_path: Path, output_path: Path | None = None) -> Path: + md_path = output_path if output_path is not None else file_path.with_suffix(".md") + md_path.write_text("converted", encoding="utf-8") + return md_path + + pinned: set[str] = set() + real_unlink = os.unlink + real_prepare_upload_destination = uploads._prepare_upload_destination + real_copy_fd_to_path = upload_ingestion._copy_fd_to_path + + def prepare_upload_destination(uploads_dir, display_filename): + upload_temp = real_prepare_upload_destination(uploads_dir, display_filename) + pinned.add(str(upload_temp.temp_path)) + return upload_temp + + def copy_fd_to_path(fd: int, dest: Path) -> None: + try: + real_copy_fd_to_path(fd, dest) + finally: + # The descriptor is gone: the staged inode is removable again. + pinned.clear() + + def unlink(path, *args, **kwargs): + if str(path) in pinned: + raise PermissionError(32, "The process cannot access the file because it is being used by another process") + return real_unlink(path, *args, **kwargs) + + monkeypatch.setattr(uploads, "_prepare_upload_destination", prepare_upload_destination) + monkeypatch.setattr(upload_ingestion, "_copy_fd_to_path", copy_fd_to_path) + monkeypatch.setattr(os, "unlink", unlink) + + with ( + patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir), + patch.object(uploads, "get_sandbox_provider", return_value=provider), + patch.object(uploads, "_auto_convert_documents_enabled", return_value=True), + patch.object(uploads, "convert_file_to_markdown", AsyncMock(side_effect=fake_convert)), + patch.object(uploads, "_make_file_sandbox_writable"), + ): + file = UploadFile(filename="report.pdf", file=BytesIO(b"pdf-bytes")) + result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-aio", request=MagicMock(), files=[file], config=SimpleNamespace())) + + assert result.success is True + assert result.files[0].filename == "report.pdf" + assert result.files[0].markdown_file == "report.md" + assert (thread_uploads_dir / "report.pdf").read_bytes() == b"pdf-bytes" + assert (thread_uploads_dir / "report.md").read_text(encoding="utf-8") == "converted" + assert not list(thread_uploads_dir.glob(f"{uploads.UPLOAD_STAGING_PREFIX}*{uploads.UPLOAD_STAGING_SUFFIX}")) + + def test_upload_files_does_not_adjust_permissions_for_local_sandbox(tmp_path): thread_uploads_dir = tmp_path / "uploads" thread_uploads_dir.mkdir(parents=True)