From 2c227dc1936ce89fc0731c86bf88db1f90d03bc2 Mon Sep 17 00:00:00 2001 From: llb555 <3103086905@qq.com> Date: Sat, 24 Oct 2026 11:07:09 +0800 Subject: [PATCH] fix(projects): keep shelf staging cleanup best-effort across the routes (#7229) --- .../app/gateway/routers/project_documents.py | 23 +- .../test_project_document_attach_cleanup.py | 232 ++++++++++++++++++ 2 files changed, 252 insertions(+), 3 deletions(-) create mode 100644 backend/tests/test_project_document_attach_cleanup.py diff --git a/backend/app/gateway/routers/project_documents.py b/backend/app/gateway/routers/project_documents.py index 453c06d4fa..e9f4730b98 100644 --- a/backend/app/gateway/routers/project_documents.py +++ b/backend/app/gateway/routers/project_documents.py @@ -156,6 +156,20 @@ async def _stream_upload(file: UploadFile) -> AsyncIterator[bytes]: yield chunk +async def _discard_staged_copy(staging_path: Path) -> None: + """Remove a staged shelf copy that is no longer needed. + + Best-effort, mirroring ``routers/uploads.py::_remove_staged_file``: the caller + is about to return or raise its real outcome, and a removal failure (a Windows + handle still held elsewhere, a read-only mount) must neither replace that + outcome nor skip later teardown. The leftover is what the staging sweep owns. + """ + try: + await run_file_io(staging_path.unlink, True) + except OSError: + logger.warning("Failed to remove staged shelf copy: %s", staging_path, exc_info=True) + + @router.get("", response_model=ProjectDocumentListResponse) @require_permission("projects", "read") async def list_project_documents( @@ -216,7 +230,7 @@ async def upload_project_document( # Unsafe project_id charset — fail closed, indistinguishable from missing. raise _not_found() if staged.size_bytes == 0: - await run_file_io(staged.staging_path.unlink, True) + await _discard_staged_copy(staged.staging_path) raise HTTPException(status_code=400, detail="Empty file") result = await add_staged_document( @@ -332,7 +346,7 @@ async def promote_thread_file_to_shelf( # Unsafe project_id charset — fail closed, indistinguishable from missing. raise _not_found() if staged.size_bytes == 0: - await run_file_io(staged.staging_path.unlink, True) + await _discard_staged_copy(staged.staging_path) raise HTTPException(status_code=400, detail="Empty file") result = await add_staged_document( @@ -424,7 +438,10 @@ async def attach_project_document_to_thread( # uploads: they propagate and the host files stay in place. await service.finalize() finally: - await run_file_io(staged_path.unlink, True) + # The staged copy is best-effort, so its removal can never replace the + # caller's outcome; ``aclose()`` is this endpoint's only teardown and must + # still run when the removal fails. + await _discard_staged_copy(staged_path) await service.aclose() return AttachDocumentResponse( filename=file_info["filename"], diff --git a/backend/tests/test_project_document_attach_cleanup.py b/backend/tests/test_project_document_attach_cleanup.py new file mode 100644 index 0000000000..bb96bca57c --- /dev/null +++ b/backend/tests/test_project_document_attach_cleanup.py @@ -0,0 +1,232 @@ +"""Attach-to-thread staging cleanup must not replace the caller's outcome. + +``attach_project_document_to_thread`` stages a copy of a shelf document and +removes it in a ``finally``. When that removal fails — a worker still holds the +handle on Windows, an antivirus scanner intercepts it, the filesystem is +read-only — the raised ``PermissionError`` would replace whatever the endpoint +was about to return, and it would also skip ``service.aclose()`` because the +call sits after the removal in the same ``finally``. +""" + +from __future__ import annotations + +import asyncio +from typing import Any + +import pytest +from fastapi import HTTPException + +from app.gateway.routers import project_documents as mod + + +class _Request: + """Minimal request stand-in; the permission decorators honor this bypass.""" + + _deerflow_test_bypass_auth = True + + +class _StagedPath: + """A staged path whose removal always fails.""" + + def __init__(self) -> None: + self.unlink_attempts = 0 + + def unlink(self, missing_ok: bool = False) -> None: + self.unlink_attempts += 1 + raise PermissionError(13, "The process cannot access the file", str(self)) + + +class _FakeService: + """Records the lifecycle calls the endpoint is expected to make.""" + + instances: list[_FakeService] = [] + + def __init__(self, *_args: Any, **_kwargs: Any) -> None: + self.opened = False + self.aclosed = False + self.cleaned = False + self.ingest_error: BaseException | None = None + _FakeService.instances.append(self) + + async def open(self) -> None: + self.opened = True + + async def ingest_chunks(self, *_args: Any, **_kwargs: Any) -> dict[str, Any]: + if self.ingest_error is not None: + raise self.ingest_error + return { + "filename": "report.txt", + "size": 4, + "virtual_path": "/threads/t1/report.txt", + "artifact_url": "/artifacts/report.txt", + } + + async def finalize(self) -> None: + return None + + async def cleanup_written(self) -> None: + self.cleaned = True + + async def aclose(self) -> None: + self.aclosed = True + + +def _patch_endpoint(monkeypatch: pytest.MonkeyPatch, staged_path: _StagedPath) -> type[_FakeService]: + _FakeService.instances = [] + + async def _require_project(_request: Any, _project_id: str) -> dict[str, Any]: + return {"id": "p1"} + + async def _stage(*_args: Any, **_kwargs: Any) -> tuple[dict[str, Any], Any]: + return ({"name": "report.txt"}, staged_path) + + async def _chunks(*_args: Any, **_kwargs: Any): + if False: # pragma: no cover - keeps this an async generator + yield b"" + + async def _run_inline(func, *args, **kwargs): + return func(*args, **kwargs) + + monkeypatch.setattr(mod, "_require_project", _require_project) + monkeypatch.setattr(mod, "get_project_document_repo", lambda _request: object()) + monkeypatch.setattr(mod, "get_effective_user_id", lambda: "u1") + monkeypatch.setattr(mod, "get_paths", lambda: object()) + monkeypatch.setattr(mod, "stage_document_copy_for_attach", _stage) + monkeypatch.setattr(mod, "ThreadUploadIngestionService", _FakeService) + monkeypatch.setattr(mod, "read_file_chunks", _chunks) + monkeypatch.setattr(mod, "run_file_io", _run_inline) + return _FakeService + + +def _call_endpoint() -> Any: + return mod.attach_project_document_to_thread( + project_id="p1", + document_id="d1", + thread_id="t1", + request=_Request(), + config=object(), + ) + + +def test_staging_cleanup_failure_keeps_the_response_and_closes_the_service( + monkeypatch: pytest.MonkeyPatch, +) -> None: + staged_path = _StagedPath() + service_cls = _patch_endpoint(monkeypatch, staged_path) + + response = asyncio.run(_call_endpoint()) + + assert response.filename == "report.txt" + assert staged_path.unlink_attempts == 1 + assert len(service_cls.instances) == 1 + assert service_cls.instances[0].aclosed is True + + +def test_staging_cleanup_failure_keeps_the_original_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + staged_path = _StagedPath() + service_cls = _patch_endpoint(monkeypatch, staged_path) + original = HTTPException(status_code=409, detail="content_missing") + + def _make_service(*args: Any, **kwargs: Any) -> Any: + service = _FakeService(*args, **kwargs) + service.ingest_error = original + return service + + monkeypatch.setattr(mod, "ThreadUploadIngestionService", _make_service) + + with pytest.raises(HTTPException) as raised: + asyncio.run(_call_endpoint()) + + assert raised.value is original + assert service_cls.instances[0].aclosed is True + assert service_cls.instances[0].cleaned is True + + +class _ZeroSizeStaged: + """A staged upload that turned out to be empty.""" + + def __init__(self, staging_path: Any) -> None: + self.size_bytes = 0 + self.staging_path = staging_path + + +def _patch_rejection_path(monkeypatch: pytest.MonkeyPatch, staged_path: _StagedPath) -> None: + """Make ``stage_document_bytes`` return an empty staged upload.""" + + async def _require_active_project(_request: Any, _project_id: str) -> dict[str, Any]: + return {"id": "p1", "status": "active"} + + async def _stage(*_args: Any, **_kwargs: Any) -> _ZeroSizeStaged: + return _ZeroSizeStaged(staged_path) + + async def _run_inline(func, *args, **kwargs): + return func(*args, **kwargs) + + monkeypatch.setattr(mod, "_require_active_project", _require_active_project) + monkeypatch.setattr(mod, "_require_project", _require_active_project) + monkeypatch.setattr(mod, "get_effective_user_id", lambda: "u1") + monkeypatch.setattr(mod, "get_paths", lambda: object()) + monkeypatch.setattr(mod, "_get_upload_limit", lambda *_args, **_kwargs: 1024) + monkeypatch.setattr(mod, "stage_document_bytes", _stage) + monkeypatch.setattr(mod, "run_file_io", _run_inline) + + +def test_empty_upload_rejection_survives_a_failing_staging_removal( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A cleanup failure must not turn the mapped 400 into a raw 500.""" + + staged_path = _StagedPath() + _patch_rejection_path(monkeypatch, staged_path) + + with pytest.raises(HTTPException) as raised: + asyncio.run( + mod.upload_project_document( + project_id="p1", + request=_Request(), + response=object(), + file=object(), + name="report.txt", + config=object(), + ) + ) + + assert raised.value.status_code == 400 + assert raised.value.detail == "Empty file" + assert staged_path.unlink_attempts == 1 + + +def test_empty_promotion_rejection_survives_a_failing_staging_removal( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Same guarantee on the save-to-shelf path.""" + + staged_path = _StagedPath() + _patch_rejection_path(monkeypatch, staged_path) + + class _ThreadStore: + async def get(self, _thread_id: str) -> dict[str, Any]: + return {"thread_id": "t1"} + + monkeypatch.setattr(mod, "get_thread_store", lambda _request: _ThreadStore()) + monkeypatch.setattr(mod, "get_project_document_repo", lambda _request: object()) + monkeypatch.setattr(mod, "_resolve_thread_source", lambda *_args, **_kwargs: object()) + + body = mod.PromoteThreadFileRequest(thread_id="t1", kind="upload", name="report.txt") + + with pytest.raises(HTTPException) as raised: + asyncio.run( + mod.promote_thread_file_to_shelf( + project_id="p1", + body=body, + request=_Request(), + response=object(), + config=object(), + ) + ) + + assert raised.value.status_code == 400 + assert raised.value.detail == "Empty file" + assert staged_path.unlink_attempts == 1