diff --git a/README.md b/README.md index 91a616a603..fe2db48627 100644 --- a/README.md +++ b/README.md @@ -3159,6 +3159,9 @@ from the project page's Documents section: - **Upload** a file (button or drag-drop, one file per request). Shelf size limits reuse `uploads.max_file_size` (default 50 MiB); re-uploading identical content returns the existing entry instead of creating a duplicate. + Oversized uploads return `413`; deduplicated uploads return `200`. Staging + cleanup failures are logged without replacing these outcomes or a missing, + foreign, or archived project's `404` response. - **List** entries with name, size, modified time, and provenance (uploaded vs. saved from a conversation), and preview or download any entry. - **Save to project** from a thread file: the read-only conversation-files diff --git a/backend/packages/harness/deerflow/projects/documents.py b/backend/packages/harness/deerflow/projects/documents.py index d82d95d821..ab29684ed8 100644 --- a/backend/packages/harness/deerflow/projects/documents.py +++ b/backend/packages/harness/deerflow/projects/documents.py @@ -32,12 +32,14 @@ import threading import uuid from collections import OrderedDict from collections.abc import AsyncIterator, Awaitable, Callable +from contextlib import suppress from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Any from deerflow.config.paths import Paths from deerflow.uploads.manager import normalize_filename +from deerflow.utils.atomic_write import cleanup_failed_atomic_write from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown from deerflow.utils.file_io import await_drained, run_file_io from deerflow.utils.text_detection import is_text_file_by_content @@ -143,8 +145,11 @@ def _open_staging(staging_dir: Path, staging_path: Path): def _close_and_unlink(handle: Any, staging_path: Path) -> None: try: handle.close() + except OSError: + with suppress(Exception): + logger.warning("Failed to close shelf staging file %s", staging_path, exc_info=True) finally: - staging_path.unlink(missing_ok=True) + cleanup_failed_atomic_write(staging_path) async def stage_document_bytes(paths: Paths, *, user_id: str, project_id: str, chunks: AsyncIterator[bytes] | list[bytes], max_bytes: int) -> StagedDocument: @@ -192,7 +197,7 @@ def _place_staging(staging_path: Path, final_path: Path) -> None: try: os.replace(staging_path, final_path) except Exception: - staging_path.unlink(missing_ok=True) + cleanup_failed_atomic_write(staging_path) raise @@ -263,7 +268,7 @@ async def add_staged_document( # row raced into trash (or a trashing project deletion) before the # re-check still owns the placed bytes — trash is recoverable, so # deleting the namespace would destroy a recoverable document. - await run_file_io(_remove_staging, staged.staging_path) + await run_file_io(cleanup_failed_atomic_write, staged.staging_path) committed: dict | None try: committed = await repo.get(document_id, include_trashed=True, user_id=user_id) @@ -292,11 +297,11 @@ async def add_staged_document( raise if row is None: # Missing/foreign/archived project: nothing was placed or inserted. - await run_file_io(_remove_staging, staged.staging_path) + await run_file_io(cleanup_failed_atomic_write, staged.staging_path) return None if row["id"] != document_id: # Dedup hit: the existing row wins; discard the staged duplicate. - await run_file_io(_remove_staging, staged.staging_path) + await run_file_io(cleanup_failed_atomic_write, staged.staging_path) return row, False return row, True diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md index b3a308bc6b..5f56e0241a 100644 --- a/backend/packages/harness/deerflow/utils/AGENTS.md +++ b/backend/packages/harness/deerflow/utils/AGENTS.md @@ -6,9 +6,11 @@ capture/fetch publication. It writes and fsyncs a hidden exclusive sibling befor umask; replacements preserve existing permission bits before publication. Never read the process umask by changing it in this threaded runtime. Behavioral tests for both tool imports live in `tests/test_browser_outputs_atomic.py`. -`cleanup_failed_atomic_write` shares failure cleanup between this publisher and -`skills/projection.py`'s private manifest writer. Call it while handling the -original `BaseException`; pass a raw descriptor only until `fdopen` succeeds. +`cleanup_failed_atomic_write` shares best-effort cleanup across this publisher, +`skills/projection.py`'s private manifest writer, and project document staging. +Use it on publication failure or after deciding to discard an owned +temporary; pass a raw descriptor only until `fdopen` succeeds. Shelf discards +must preserve both a rejected project's 404 and a dedup hit's 200 response. Callers retain their own serialization, permissions and publication policy. Failure cleanup must preserve the original publication exception, including interruptions. Descriptor close and sibling unlink are separate best-effort diff --git a/backend/packages/harness/deerflow/utils/atomic_write.py b/backend/packages/harness/deerflow/utils/atomic_write.py index 05638ebf18..42fbb0f424 100644 --- a/backend/packages/harness/deerflow/utils/atomic_write.py +++ b/backend/packages/harness/deerflow/utils/atomic_write.py @@ -13,11 +13,12 @@ logger = logging.getLogger(__name__) def cleanup_failed_atomic_write(temporary: Path, fd: int | None = None) -> None: - """Release an owned temporary without replacing the caller's active error. + """Release an owned temporary without replacing the caller's outcome. Pass *fd* only while the raw descriptor is still owned by the caller; successful ``fdopen`` transfers that ownership to the file object. - Filesystem and logging-handler errors remain independent best-effort steps. + Also use this to discard staging after a publication decision. Filesystem + and logging-handler errors remain independent best-effort steps. """ # Cleanup and diagnostics are best-effort: keep the original error # and attempt unlink even if closing an untransferred descriptor fails. diff --git a/backend/tests/blocking_io/test_project_documents.py b/backend/tests/blocking_io/test_project_documents.py index fb08765e5a..230f92c8e4 100644 --- a/backend/tests/blocking_io/test_project_documents.py +++ b/backend/tests/blocking_io/test_project_documents.py @@ -126,20 +126,27 @@ async def test_list_content_missing_check_dispatches_off_the_loop(tmp_path, monk await close_engine() -async def test_dedup_hit_staging_cleanup_dispatches_off_the_loop(tmp_path, monkeypatch) -> None: - """The dedup-hit branch removes the staged duplicate via the offload.""" +@pytest.mark.parametrize("outcome", ["dedup", "archived"]) +async def test_staging_discard_dispatches_off_the_loop(tmp_path, monkeypatch, outcome) -> None: + """Both discarded-staging branches perform best-effort cleanup off-loop.""" calls = _spy_offload(monkeypatch) env = await _make_env(tmp_path, monkeypatch) try: - first = await stage_document_bytes(env.paths, user_id=_USER, project_id=env.project_id, chunks=[b"same"], max_bytes=1 << 20) - await add_staged_document(env.docs, env.paths, user_id=_USER, project_id=env.project_id, name="first.txt", staged=first) + if outcome == "dedup": + first = await stage_document_bytes(env.paths, user_id=_USER, project_id=env.project_id, chunks=[b"same"], max_bytes=1 << 20) + await add_staged_document(env.docs, env.paths, user_id=_USER, project_id=env.project_id, name="first.txt", staged=first) + else: + await ProjectRepository(get_session_factory()).set_status(env.project_id, "archived", user_id=_USER) calls.clear() - duplicate = await stage_document_bytes(env.paths, user_id=_USER, project_id=env.project_id, chunks=[b"same"], max_bytes=1 << 20) - result = await add_staged_document(env.docs, env.paths, user_id=_USER, project_id=env.project_id, name="second.txt", staged=duplicate) - assert result is not None and result[1] is False - assert not duplicate.staging_path.exists() - assert "_remove_staging" in calls - # No placement happened for the discarded duplicate. + staged = await stage_document_bytes(env.paths, user_id=_USER, project_id=env.project_id, chunks=[b"same"], max_bytes=1 << 20) + result = await add_staged_document(env.docs, env.paths, user_id=_USER, project_id=env.project_id, name="second.txt", staged=staged) + if outcome == "dedup": + assert result is not None and result[1] is False + else: + assert result is None + assert not staged.staging_path.exists() + assert "cleanup_failed_atomic_write" in calls + # No placement happened for either discarded upload. assert "_place_staging" not in calls finally: await close_engine() diff --git a/backend/tests/test_project_documents_router.py b/backend/tests/test_project_documents_router.py index e853e652ad..7be6d55b32 100644 --- a/backend/tests/test_project_documents_router.py +++ b/backend/tests/test_project_documents_router.py @@ -639,3 +639,188 @@ class TestDeleteScopesToTheUrlProject: assert client.delete(f"/api/projects/{project_a['id']}/documents/{doc['id']}").status_code == 404 assert client.get(f"/api/projects/{project_b['id']}/documents/{doc['id']}/content").status_code == 200 assert client.delete(f"/api/projects/{project_b['id']}/documents/{doc['id']}").status_code == 204 + + +class TestStagingFailureCleanup: + @pytest.mark.parametrize("failure", ["none", "close", "unlink", "logging"]) + def test_oversize_upload_keeps_413_when_cleanup_fails(self, tmp_path, monkeypatch, caplog, failure): + import logging + from pathlib import Path + + from deerflow.projects import documents + from deerflow.utils import atomic_write + + app = _build_app(tmp_path) + app.dependency_overrides[get_config] = lambda: SimpleNamespace(uploads={"max_file_size": 3}) + with TestClient(app) as client: + pid = _create_project(client)["id"] + staging = get_paths().project_documents_dir("user-a", pid) / ".staging" + if failure == "close": + open_staging = documents._open_staging + + def open_with_close_error(*args): + handle = open_staging(*args) + close = handle.close + + def fail_close(): + close() + raise OSError("staging close failed") + + handle.close = fail_close + return handle + + monkeypatch.setattr(documents, "_open_staging", open_with_close_error) + elif failure in {"unlink", "logging"}: + unlink = Path.unlink + + def fail_unlink(path, *args, **kwargs): + if path.parent == staging: + raise OSError("staging unlink failed") + return unlink(path, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", fail_unlink) + if failure == "logging": + + def fail_warning(*args, **kwargs): + raise RuntimeError("logging handler failed") + + monkeypatch.setattr(atomic_write.logger, "warning", fail_warning) + with caplog.at_level(logging.WARNING): + response = _upload(client, pid, "large.txt", b"1234") + assert response.status_code == 413 + assert response.json()["detail"] == "File too large: large.txt" + assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 0 + remaining = list(staging.iterdir()) + if failure in {"unlink", "logging"}: + assert len(remaining) == 1 + if failure == "unlink": + assert str(remaining[0]) in caplog.text + assert "staging unlink failed" in caplog.text + else: + assert remaining == [] + if failure == "close": + assert str(staging) in caplog.text + assert "staging close failed" in caplog.text + + @pytest.mark.parametrize("after_commit", [False, True]) + def test_insert_failure_cleanup_preserves_error_or_committed_document(self, tmp_path, monkeypatch, caplog, after_commit): + from pathlib import Path + + from deerflow.projects import documents + + app = _build_app(tmp_path) + primary = OSError("original shelf insert failure") + insert = app.state.project_document_repo.insert_active + if after_commit: + + async def insert_then_fail(*args, **kwargs): + await insert(*args, **kwargs) + raise primary + + monkeypatch.setattr(app.state.project_document_repo, "insert_active", insert_then_fail) + else: + + def fail_replace(*args, **kwargs): + raise primary + + monkeypatch.setattr(documents.os, "replace", fail_replace) + with TestClient(app) as client: + pid = _create_project(client)["id"] + staging = get_paths().project_documents_dir("user-a", pid) / ".staging" + unlink = Path.unlink + + def fail_unlink(path, *args, **kwargs): + if path.parent == staging: + raise OSError("staging unlink failed") + return unlink(path, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", fail_unlink) + if after_commit: + response = _upload(client, pid, "saved.txt", b"saved bytes") + assert response.status_code == 201 + doc = response.json()["document"] + assert client.get(f"/api/projects/{pid}/documents/{doc['id']}/content").content == b"saved bytes" + assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 1 + else: + with pytest.raises(OSError) as caught: + _upload(client, pid, "saved.txt", b"saved bytes") + assert caught.value is primary + assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 0 + assert "staging unlink failed" in caplog.text + + @pytest.mark.parametrize("outcome", ["dedup", "archived"]) + @pytest.mark.parametrize("failure", ["none", "unlink", "logging"]) + def test_staging_discard_preserves_upload_outcome(self, tmp_path, monkeypatch, caplog, outcome, failure): + import logging + from pathlib import Path + + from deerflow.utils import atomic_write + + app = _build_app(tmp_path) + payload = b"same shelf bytes" + with TestClient(app) as client: + pid = _create_project(client)["id"] + staging = get_paths().project_documents_dir("user-a", pid) / ".staging" + if outcome == "dedup": + first = _upload(client, pid, "first.txt", payload) + assert first.status_code == 201 + original = first.json()["document"] + else: + real_stage = project_documents.stage_document_bytes + + async def archive_then_stage(*args, **kwargs): + # The locked insert rejects a project archived after admission. + await app.state.project_repo.set_status(kwargs["project_id"], "archived") + return await real_stage(*args, **kwargs) + + monkeypatch.setattr(project_documents, "stage_document_bytes", archive_then_stage) + + unlink = Path.unlink + attempts = [] + + def discard(path, *args, **kwargs): + if path.parent == staging: + attempts.append(path) + if failure != "none": + raise PermissionError("staging discard failed") + return unlink(path, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", discard) + warning = MagicMock(wraps=atomic_write.logger.warning) + monkeypatch.setattr(atomic_write.logger, "warning", warning) + # This stdlib handler raises from emit when its delayed open fails. + handler = logging.FileHandler(tmp_path / "missing-logs" / "gateway.log", delay=True) + handler.emit = MagicMock(wraps=handler.emit) + if failure == "logging": + atomic_write.logger.addHandler(handler) + try: + with caplog.at_level(logging.WARNING, logger=atomic_write.__name__): + response = _upload(client, pid, "duplicate.txt", payload) + if outcome == "dedup": + assert response.status_code == 200 + body = response.json() + assert body["deduplicated"] is True + assert body["document"]["id"] == original["id"] + assert body["document"]["name"] == "first.txt" + assert client.get(f"/api/projects/{pid}/documents/{original['id']}/content").content == payload + assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 1 + else: + assert response.status_code == 404 + assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 0 + assert len(attempts) == 1 + if failure == "none": + assert list(staging.iterdir()) == [] + warning.assert_not_called() + else: + assert list(staging.iterdir()) == attempts + assert attempts[0].read_bytes() == payload + warning.assert_called_once() + assert attempts[0] in warning.call_args.args + assert warning.call_args.kwargs["exc_info"] is True + if failure == "unlink": + assert str(attempts[0]) in caplog.text + assert "staging discard failed" in caplog.text + assert handler.emit.call_count == int(failure == "logging") + finally: + atomic_write.logger.removeHandler(handler) + handler.close()