diff --git a/README.md b/README.md index 61696958f0..aff33b3795 100644 --- a/README.md +++ b/README.md @@ -1784,6 +1784,11 @@ can set `sandbox.thread_data_mounts: true` to skip that per-upload sandbox acquire and sync. Leave the field unset for automatic detection; setting it incorrectly can make uploaded files unavailable inside the sandbox. +Failed upload commits report the original error even if temporary-file cleanup +also fails, for example because of a Windows sharing violation. +Once the file is published, a temporary-file cleanup failure is logged without +failing the upload; hidden staging files are left for the startup sweep. + This is the difference between a chatbot with tool access and an agent with an actual execution environment. ``` diff --git a/backend/AGENTS.md b/backend/AGENTS.md index d7449f1910..5d029c431c 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -338,7 +338,7 @@ Outlines use ATX syntax (1–6 hashes, space/tab separator, ≤3 leading spaces) - One conversion worker per request when called from an active event loop - Files stored in thread-isolated directories under the resolving user's bucket (`users/{user_id}/threads/{thread_id}/user-data/uploads`). For IM channels the owner is threaded explicitly via the `user_id=` kwarg (see IM Channels → Owner-scoped file storage); HTTP/embedded callers resolve it from `get_effective_user_id()` - Duplicate filenames within one request get `_N` suffixes to prevent overwrites. -- Gateway HTTP uploads stage bytes as `.upload-*.part` files and atomically replace the destination only after size validation. These staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools, and swept on Gateway startup if a hard crash leaves one behind. +- Gateway HTTP uploads stage `.upload-*.part` files, hidden from upload listings, agent context, and sandbox listings/searches. After size validation, publication is atomic; staged-name cleanup logs errors and leaves leftovers for startup sweep. - Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor. Non-mounted sandbox uploads acquire sandboxes with `SandboxProvider.acquire_async()` and offload `read_bytes()` plus `sandbox.update_file()` together. - Mounted uploads skip sandbox acquire/sync. AIO remote/provisioner requires accurate `sandbox.thread_data_mounts: true`; omission keeps backend auto-detection. - `UploadsMiddleware` caps outline titles at 200 characters and previews at 2000 including markers. Titles use `original_user_content`, not upload-prefixed content; attachment-only titles use a sanitized, bounded filename or count. diff --git a/backend/app/gateway/routers/uploads.py b/backend/app/gateway/routers/uploads.py index 641e580394..c3fa0d022d 100644 --- a/backend/app/gateway/routers/uploads.py +++ b/backend/app/gateway/routers/uploads.py @@ -253,15 +253,15 @@ def _link_staged_no_overwrite( happens AFTER the atomic failure: an existing non-regular file (a planted symlink — symlinks are excluded from the seeded listing, so one could only come from outside) stays an unsafe destination; anything else is a - 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. + plain collision to retry. Any other failure removes the staged file + best-effort and propagates; success also removes it best-effort. 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. + caller retains ownership of the staged path on both success and failure. """ file_path = _pure_destination(uploads_dir, display_filename) try: @@ -274,13 +274,11 @@ def _link_staged_no_overwrite( pass # The winner vanished between link and lstat — plain retry. raise except Exception: - try: - os.unlink(staged_path) - except FileNotFoundError: - pass + if unlink_staged: + _remove_staged_file(staged_path) raise if unlink_staged: - os.unlink(staged_path) + _remove_staged_file(staged_path) return file_path @@ -296,8 +294,9 @@ def _commit_upload_temp_no_overwrite( 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. ``unlink_staged=False`` hands the staged name - back to the caller, which still holds a descriptor on its inode. + other failure removes it best-effort. With ``unlink_staged=False``, the + caller retains the staged name on success and failure until it releases + its descriptor on the inode. """ upload_temp.handle.close() return _link_staged_no_overwrite(upload_temp.temp_path, uploads_dir, display_filename, unlink_staged=unlink_staged) diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py index eef55c68b8..d729fe9911 100644 --- a/backend/tests/test_uploads_router.py +++ b/backend/tests/test_uploads_router.py @@ -1,4 +1,5 @@ import asyncio +import errno import os import stat import threading @@ -443,6 +444,104 @@ def test_upload_files_removes_the_staged_part_when_the_platform_blocks_an_open_u assert not list(thread_uploads_dir.glob(f"{uploads.UPLOAD_STAGING_PREFIX}*{uploads.UPLOAD_STAGING_SUFFIX}")) +@pytest.mark.parametrize("filename", ["notes.txt", "report.pdf"]) +@pytest.mark.parametrize("simulate_sharing_violation", [False, True]) +def test_upload_files_preserves_link_error_and_cleans_staging_after_closing_conversion_fd(tmp_path, monkeypatch, filename, simulate_sharing_violation): + thread_uploads_dir = tmp_path / "uploads" + thread_uploads_dir.mkdir() + conversion_fds: set[int] = set() + duplicated_fds: list[int] = [] + blocked_unlinks: list[str] = [] + real_dup = upload_ingestion._dup_for_conversion + real_close = upload_ingestion._close_fd + real_unlink = os.unlink + + async def dup_for_conversion(fd): + duplicate = await real_dup(fd) + conversion_fds.add(duplicate) + duplicated_fds.append(duplicate) + return duplicate + + def close_fd(fd): + real_close(fd) + conversion_fds.discard(fd) + + def unlink(path, *args, **kwargs): + # Simulate Windows sharing violations on every platform, while keeping + # the real conversion descriptor and the production abort path. + if simulate_sharing_violation and conversion_fds and str(path).endswith(uploads.UPLOAD_STAGING_SUFFIX): + blocked_unlinks.append(str(path)) + raise PermissionError("conversion descriptor is still open") + return real_unlink(path, *args, **kwargs) + + monkeypatch.setattr(upload_ingestion, "_dup_for_conversion", dup_for_conversion) + monkeypatch.setattr(upload_ingestion, "_close_fd", close_fd) + 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=_mounted_provider()), + patch.object(uploads, "_auto_convert_documents_enabled", return_value=True), + patch.object(os, "link", side_effect=OSError(errno.EIO, "original link failure")), + ): + file = UploadFile(filename=filename, file=BytesIO(b"upload bytes")) + with pytest.raises(HTTPException) as exc_info: + asyncio.run(call_unwrapped(uploads.upload_files, "thread-link-error", request=MagicMock(), files=[file], config=SimpleNamespace())) + + assert exc_info.value.status_code == 500 + assert "original link failure" in exc_info.value.detail + assert not blocked_unlinks + assert not conversion_fds + assert len(duplicated_fds) == (1 if filename == "report.pdf" else 0) + assert list(thread_uploads_dir.iterdir()) == [] + + +def test_upload_files_succeeds_when_published_staging_cleanup_fails(tmp_path, monkeypatch, caplog): + thread_uploads_dir = tmp_path / "uploads" + thread_uploads_dir.mkdir() + real_unlink = os.unlink + + def unlink(path, *args, **kwargs): + if str(path).endswith(uploads.UPLOAD_STAGING_SUFFIX): + raise PermissionError("staged cleanup failure") + return real_unlink(path, *args, **kwargs) + + 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=_mounted_provider()), + patch.object(uploads, "_auto_convert_documents_enabled", return_value=False), + ): + file = UploadFile(filename="notes.txt", file=BytesIO(b"published bytes")) + result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-cleanup-error", request=MagicMock(), files=[file], config=SimpleNamespace())) + + assert result.success + assert (thread_uploads_dir / "notes.txt").read_bytes() == b"published bytes" + # The hidden staging name is left for the startup sweep, not reported as a + # failed upload after the destination has already been published. + staged_paths = list(thread_uploads_dir.glob(".upload-*.part")) + assert len(staged_paths) == 1 + assert "Failed to remove staged upload file" in caplog.text + + +def test_link_staged_preserves_original_error_when_cleanup_fails(tmp_path): + staged_path = tmp_path / "upload.part" + staged_path.write_bytes(b"upload bytes") + original_error = OSError(errno.EIO, "original link failure") + with ( + patch.object(os, "link", side_effect=original_error), + patch.object(os, "unlink", side_effect=PermissionError("cleanup failure")) as unlink, + pytest.raises(OSError) as exc_info, + ): + uploads._link_staged_no_overwrite(staged_path, tmp_path, "report.pdf") + + assert exc_info.value is original_error + unlink.assert_called_once_with(staged_path) + assert staged_path.read_bytes() == b"upload bytes" + assert not (tmp_path / "report.pdf").exists() + + 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)