From a83b244a8c60f1ced096ca0c161348ad21800888 Mon Sep 17 00:00:00 2001 From: Nan Gao Date: Thu, 8 Oct 2026 07:45:29 +0800 Subject: [PATCH] fix(uploads): keep in-flight staging files out of the startup sweep (#6445) * fix(uploads): keep in-flight staging files out of the startup sweep `cleanup_stale_upload_staging_files` removed every `.upload-*.part` it found at Gateway startup. That assumed a single Gateway: nothing can be uploading while it boots, so every staging file is a crash leftover. With several replicas sharing one home volume (Helm RWX), a replica starting during a rolling update deleted the staging file of an upload another replica was still writing; that upload then failed at its `os.link` commit and lost its bytes. The sweep now skips staging files younger than `UPLOAD_STAGING_MIN_AGE` (24 hours, the same guard as the project-document `.staging` orphan sweep in `projects/trash.py`). Chunk writes refresh the mtime, so an upload in flight stays younger than the guard; an unreadable mtime keeps the file. The startup call site is unchanged and uses the default. Tests: the existing sweep tests backdate their orphans; new cases pin young-kept / old-removed, the configurable `min_age`, unreadable and vanished mtimes, and a router test that runs the sweep while an upload is blocked mid-write and asserts the upload still commits. * docs(changelog): link #6445 * fix(uploads): reclaim published staging aliases at any age Review follow-up (#6445): a Gateway that dies after the commit's `os.link` but before the staged-name removal leaves the final file and its `.upload-*.part` alias on one inode. With only the age guard, the next startup kept that alias for 24 hours, and because the sweep runs only at startup, `copy_upload_file_no_symlink` (embedded `DeerFlowClient.upload_files`) kept rejecting the destination for its multiple links. The sweep now stats each staging name with `os.lstat` (a `DirEntry` stat reports `st_nlink == 0` on Windows) and removes a name with `st_nlink > 1` at any age: a staged part only ever gains its second link through its own commit, so such a name is a published alias, never an upload in flight. Lone parts keep the 24 hour mtime guard. The destination hardlink check is unchanged. Tests cover the alias next to a young lone part, the replacement upload succeeding afterwards, and the crash reproduced through the real upload path with the staged-name removal patched out. --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 14 +++ backend/AGENTS.md | 2 +- .../harness/deerflow/uploads/manager.py | 44 ++++++- backend/tests/test_uploads_manager.py | 118 ++++++++++++++++++ backend/tests/test_uploads_router.py | 97 +++++++++++++- 5 files changed, 269 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 373e3b06cc..d8325efe90 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -804,6 +804,19 @@ This release closes that milestone with **439 merged pull requests**. ### Fixed +- **uploads:** The Gateway's startup sweep of orphaned `.upload-*.part` staging + files now skips files younger than 24 hours. The sweep removed every staging + file it found, which was right for one Gateway but not for several replicas + sharing a home volume: a replica starting during a rolling update deleted the + staging file of an upload another replica was still writing, and that upload + then failed at its atomic commit. Chunk writes refresh the staging file's + mtime, so an upload in flight stays younger than the guard; a crash leftover + is collected by the first startup more than 24 hours later, the same guard + project-document staging already uses. A staging file that already shares + its inode with the published upload (a crash between the atomic link and the + staged-name removal) is still reclaimed on the next startup at any age, so + that destination does not fail the multi-link safety check on its next + replacement. ([#6445]) - **gateway:** `GET /health/ready` now reports unready while the Redis stream bridge is unreachable. The bridge's Redis client connects lazily and nothing pinged it, so a gateway whose Redis was down started, answered `200 ready` to @@ -9144,4 +9157,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6407]: https://github.com/bytedance/deer-flow/pull/6407 [#6412]: https://github.com/bytedance/deer-flow/pull/6412 [#6426]: https://github.com/bytedance/deer-flow/pull/6426 +[#6445]: https://github.com/bytedance/deer-flow/pull/6445 [#6447]: https://github.com/bytedance/deer-flow/pull/6447 diff --git a/backend/AGENTS.md b/backend/AGENTS.md index a51ef13434..e77e931722 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -346,7 +346,7 @@ Outlines use ATX syntax (1–6 hashes, space/tab separator, ≤3 leading spaces) - 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()` - Per-thread `upload-companions/`: source mtime/ctime. - Duplicate filenames within one request get `_N` suffixes to prevent overwrites. -- 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 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. The sweep keeps lone `.part` files under 24h: they may be in flight on another replica; `st_nlink > 1` means a published alias, removed at any age so the multi-link check can still replace the file. - 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/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index 69c8f5f6b7..debef97233 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -9,6 +9,8 @@ import logging import os import shutil import stat +import time +from datetime import timedelta from pathlib import Path from urllib.parse import quote @@ -30,6 +32,9 @@ logger = logging.getLogger(__name__) UPLOAD_STAGING_PREFIX = ".upload-" UPLOAD_STAGING_SUFFIX = ".part" +# Staging files younger than this are treated as in flight by the startup sweep. +# Mirrors the project-document ``.staging`` orphan guard (``projects/trash.py``). +UPLOAD_STAGING_MIN_AGE = timedelta(hours=24) _MAX_FILENAME_BYTES = 255 @@ -181,9 +186,35 @@ def _iter_upload_dirs(base_dir: Path): yield from base_dir.glob("users/*/threads/*/user-data/uploads") -def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None) -> int: - """Remove orphaned Gateway upload staging files left by a hard crash.""" +def _staging_entry_stat(entry: os.DirEntry[str]) -> os.stat_result: + # ``os.lstat`` rather than ``entry.stat``: on Windows a ``DirEntry`` stat leaves + # ``st_nlink`` at zero, and the published-alias rule in the sweep depends on it. + return os.lstat(entry.path) + + +def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None, *, min_age: timedelta = UPLOAD_STAGING_MIN_AGE) -> int: + """Remove orphaned Gateway upload staging files left by a hard crash. + + A lone ``.upload-*.part`` (``st_nlink == 1``) is removed only once its mtime + is older than *min_age*. The uploads directories may live on a volume + shared by several Gateway replicas, so at startup such a file can belong to + an upload another replica is still writing; each chunk write refreshes its + mtime, which keeps it younger than the guard until it is committed or + abandoned. + + A staging name with ``st_nlink > 1`` is removed at any age: the commit's + ``os.link`` already published those bytes under their final name, so the + staged name is only an alias left behind by a crash (or a deferred removal + that never ran). Keeping it would make that destination fail the + multi-link safety check on the next replacement upload (embedded + ``DeerFlowClient.upload_files`` goes through ``copy_upload_file_no_symlink``). + Removing the alias cannot affect an upload in flight: a staged part gains + its second link only through its own commit. + + A file that cannot be stat'ed is kept, never removed on a guess. + """ root = Path(base_dir) if base_dir is not None else get_paths().base_dir + cutoff = time.time() - min_age.total_seconds() removed = 0 for uploads_dir in _iter_upload_dirs(root): if not uploads_dir.is_dir(): @@ -193,6 +224,15 @@ def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None) -> in for entry in entries: if not is_upload_staging_file(entry.name) or not entry.is_file(follow_symlinks=False): continue + try: + st = _staging_entry_stat(entry) + except FileNotFoundError: + continue + except OSError: + logger.warning("Could not stat upload staging file %s; keeping it", entry.path, exc_info=True) + continue + if st.st_nlink <= 1 and st.st_mtime >= cutoff: + continue # a lone, young part may still be in flight on another replica try: os.unlink(entry.path) removed += 1 diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py index 42a11bf061..54326fe98d 100644 --- a/backend/tests/test_uploads_manager.py +++ b/backend/tests/test_uploads_manager.py @@ -4,6 +4,8 @@ import errno import os import shutil import stat +import time +from datetime import timedelta from unittest.mock import patch import pytest @@ -19,6 +21,7 @@ from deerflow.uploads.manager import ( list_files_in_dir, normalize_filename, validate_path_traversal, + validate_upload_destination, write_upload_file_no_symlink, ) @@ -405,6 +408,12 @@ class TestListFilesInDir: # --------------------------------------------------------------------------- +def _set_age(path, age: timedelta) -> None: + """Backdate *path*'s mtime so the cleanup sees it as *age* old.""" + timestamp = time.time() - age.total_seconds() + os.utime(path, (timestamp, timestamp)) + + class TestCleanupStaleUploadStagingFiles: def test_removes_only_stale_staging_files_from_all_upload_layouts(self, tmp_path): legacy_uploads = tmp_path / "threads" / "thread-legacy" / "user-data" / "uploads" @@ -419,6 +428,15 @@ class TestCleanupStaleUploadStagingFiles: (legacy_uploads / ".env").write_text("intentional dotfile") (legacy_uploads / ".upload-note.txt").write_text("intentional upload") (legacy_uploads / "draft.part").write_text("intentional upload") + for path in ( + legacy_uploads / ".upload-old.part", + user_uploads / ".upload-new.part", + unrelated_uploads / ".upload-ignore.part", + legacy_uploads / ".env", + legacy_uploads / ".upload-note.txt", + legacy_uploads / "draft.part", + ): + _set_age(path, timedelta(days=2)) removed = cleanup_stale_upload_staging_files(tmp_path) @@ -430,6 +448,106 @@ class TestCleanupStaleUploadStagingFiles: assert (legacy_uploads / ".upload-note.txt").exists() assert (legacy_uploads / "draft.part").exists() + def test_keeps_staging_files_younger_than_the_default_guard(self, tmp_path): + """On a volume shared by several Gateway replicas, a young ``.part`` may be + an upload another replica is still writing; only old ones are orphans.""" + uploads_dir = tmp_path / "users" / "owner-1" / "threads" / "thread-1" / "user-data" / "uploads" + uploads_dir.mkdir(parents=True) + fresh = uploads_dir / ".upload-fresh.part" + recent = uploads_dir / ".upload-recent.part" + old = uploads_dir / ".upload-old.part" + for path in (fresh, recent, old): + path.write_text("partial") + _set_age(recent, timedelta(hours=23)) + _set_age(old, timedelta(hours=25)) + + removed = cleanup_stale_upload_staging_files(tmp_path) + + assert removed == 1 + assert fresh.exists() + assert recent.exists() + assert not old.exists() + + def test_min_age_is_configurable(self, tmp_path): + uploads_dir = tmp_path / "threads" / "thread-1" / "user-data" / "uploads" + uploads_dir.mkdir(parents=True) + young = uploads_dir / ".upload-young.part" + old = uploads_dir / ".upload-old.part" + for path in (young, old): + path.write_text("partial") + _set_age(young, timedelta(minutes=1)) + _set_age(old, timedelta(minutes=10)) + + removed = cleanup_stale_upload_staging_files(tmp_path, min_age=timedelta(minutes=5)) + + assert removed == 1 + assert young.exists() + assert not old.exists() + + def test_keeps_staging_file_when_its_age_cannot_be_read(self, tmp_path, caplog): + uploads_dir = tmp_path / "threads" / "thread-1" / "user-data" / "uploads" + uploads_dir.mkdir(parents=True) + old = uploads_dir / ".upload-old.part" + old.write_text("partial") + _set_age(old, timedelta(days=2)) + + with ( + patch("deerflow.uploads.manager._staging_entry_stat", side_effect=PermissionError(errno.EACCES, "denied")), + caplog.at_level("WARNING", logger="deerflow.uploads.manager"), + ): + removed = cleanup_stale_upload_staging_files(tmp_path) + + assert removed == 0 + assert old.exists() + assert any("keeping it" in record.getMessage() for record in caplog.records) + + def test_removes_published_alias_at_any_age_but_keeps_lone_young_part(self, tmp_path): + """A crash between the commit's ``os.link`` and the staged-name removal + leaves the final file and its ``.part`` alias sharing one inode. The + alias is reclaimed on the next startup regardless of age, or the + destination would fail the multi-link check on its next replacement; + a lone young part next to it is still treated as in flight.""" + uploads_dir = tmp_path / "users" / "owner-1" / "threads" / "thread-1" / "user-data" / "uploads" + uploads_dir.mkdir(parents=True) + published = uploads_dir / "notes.txt" + published.write_bytes(b"published bytes") + alias = uploads_dir / ".upload-abc123.part" + try: + os.link(published, alias) + except OSError as exc: # pragma: no cover - filesystems without hard links + pytest.skip(f"hard links unsupported here: {exc}") + in_flight = uploads_dir / ".upload-inflight.part" + in_flight.write_bytes(b"partial") + with pytest.raises(UnsafeUploadPathError, match="multiple links"): + validate_upload_destination(uploads_dir, "notes.txt") + + removed = cleanup_stale_upload_staging_files(tmp_path) + + assert removed == 1 + assert not alias.exists() + assert in_flight.exists() + assert published.read_bytes() == b"published bytes" + assert os.lstat(published).st_nlink == 1 + assert validate_upload_destination(uploads_dir, "notes.txt") == published + replacement = tmp_path / "replacement.txt" + replacement.write_bytes(b"replacement bytes") + assert copy_upload_file_no_symlink(uploads_dir, "notes.txt", replacement) == published + assert published.read_bytes() == b"replacement bytes" + + def test_skips_staging_file_that_vanished_before_its_age_was_read(self, tmp_path, caplog): + uploads_dir = tmp_path / "threads" / "thread-1" / "user-data" / "uploads" + uploads_dir.mkdir(parents=True) + (uploads_dir / ".upload-gone.part").write_text("partial") + + with ( + patch("deerflow.uploads.manager._staging_entry_stat", side_effect=FileNotFoundError), + caplog.at_level("WARNING", logger="deerflow.uploads.manager"), + ): + removed = cleanup_stale_upload_staging_files(tmp_path) + + assert removed == 0 + assert caplog.records == [] + # --------------------------------------------------------------------------- # delete_file_safe diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py index d26d3c4e3f..390bc740e8 100644 --- a/backend/tests/test_uploads_router.py +++ b/backend/tests/test_uploads_router.py @@ -3,6 +3,8 @@ import errno import os import stat import threading +import time +from datetime import timedelta from io import BytesIO from pathlib import Path, PurePosixPath from types import SimpleNamespace @@ -18,6 +20,7 @@ from app.gateway.deps import get_config from app.gateway.routers import uploads from deerflow.sandbox.lease import get_sandbox_lease_manager from deerflow.uploads.companions import resolve_companion +from deerflow.uploads.manager import cleanup_stale_upload_staging_files class ChunkedUpload: @@ -747,13 +750,14 @@ def test_upload_files_in_flight_upload_is_invisible_to_listings(tmp_path): def test_orphaned_staging_part_is_hidden_from_listings_and_swept(tmp_path): """A crashed upload leaves only a hidden ``.upload-*.part`` file: the - listing never exposes it and the startup cleanup removes it.""" - from deerflow.uploads.manager import cleanup_stale_upload_staging_files - + listing never exposes it and, once it is older than the in-flight guard, + the startup cleanup removes it.""" thread_uploads_dir = tmp_path / "threads" / "t1" / "user-data" / "uploads" thread_uploads_dir.mkdir(parents=True) orphan = thread_uploads_dir / ".upload-crashed.part" orphan.write_bytes(b"partial") + stale = time.time() - timedelta(days=2).total_seconds() + os.utime(orphan, (stale, stale)) visible = thread_uploads_dir / "kept.txt" visible.write_bytes(b"kept") @@ -764,6 +768,93 @@ def test_orphaned_staging_part_is_hidden_from_listings_and_swept(tmp_path): assert visible.read_bytes() == b"kept" +def test_startup_cleanup_reclaims_a_published_staging_alias_immediately(tmp_path): + """A Gateway that dies after the commit's ``os.link`` but before the staged + name is removed leaves ``notes.txt`` and its ``.upload-*.part`` alias on one + inode. The next startup must reclaim the alias at once: left in place, the + destination fails the multi-link check on its next replacement upload.""" + from deerflow.uploads.manager import validate_upload_destination + + thread_uploads_dir = tmp_path / "threads" / "thread-local" / "user-data" / "uploads" + thread_uploads_dir.mkdir(parents=True) + provider = _mounted_provider() + + 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), + # The crash: the staged name is never removed after the link publishes the bytes. + patch.object(uploads, "_remove_staged_file"), + ): + file = ChunkedUpload("notes.txt", [b"hello"]) + result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-local", request=MagicMock(), files=[file], config=SimpleNamespace())) + + assert result.success is True + names = sorted(p.name for p in thread_uploads_dir.iterdir()) + assert names[-1] == "notes.txt" and len(names) == 2 and names[0].startswith(".upload-") and names[0].endswith(".part") + assert os.lstat(thread_uploads_dir / "notes.txt").st_nlink == 2 + with pytest.raises(ValueError, match="multiple links"): + validate_upload_destination(thread_uploads_dir, "notes.txt") + + assert cleanup_stale_upload_staging_files(tmp_path) == 1 + + assert [p.name for p in thread_uploads_dir.iterdir()] == ["notes.txt"] + assert (thread_uploads_dir / "notes.txt").read_bytes() == b"hello" + assert validate_upload_destination(thread_uploads_dir, "notes.txt") == thread_uploads_dir / "notes.txt" + + +def test_startup_cleanup_keeps_an_in_flight_upload_on_a_shared_volume(tmp_path): + """Several Gateway replicas can share one home volume. A replica starting + while another one is still writing a ``.upload-*.part`` must not sweep it: + that upload would lose its staged bytes at the atomic link and fail.""" + thread_uploads_dir = tmp_path / "threads" / "thread-local" / "user-data" / "uploads" + thread_uploads_dir.mkdir(parents=True) + + provider = _mounted_provider() + real_write = uploads._write_upload_chunk + write_started = threading.Event() + release_write = threading.Event() + + def gated_write(upload_temp, chunk) -> None: + write_started.set() + assert release_write.wait(timeout=10) + real_write(upload_temp, chunk) + + results = [] + + def run_upload() -> None: + file = ChunkedUpload("notes.txt", [b"hello"]) + try: + results.append(asyncio.run(call_unwrapped(uploads.upload_files, "thread-local", request=MagicMock(), files=[file], config=SimpleNamespace()))) + except BaseException as exc: # noqa: BLE001 - surfaced through the assertions below + results.append(exc) + + 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, "_write_upload_chunk", side_effect=gated_write), + ): + upload_thread = threading.Thread(target=run_upload) + upload_thread.start() + assert write_started.wait(timeout=10) + staged = [p.name for p in thread_uploads_dir.iterdir()] + assert len(staged) == 1 and staged[0].startswith(".upload-") and staged[0].endswith(".part") + + # Another replica starts up mid-upload and runs its staging sweep. + assert cleanup_stale_upload_staging_files(tmp_path) == 0 + assert [p.name for p in thread_uploads_dir.iterdir()] == staged + + release_write.set() + upload_thread.join(timeout=15) + + assert not upload_thread.is_alive() + assert not isinstance(results[0], BaseException), results[0] + assert results[0].success is True + assert (thread_uploads_dir / "notes.txt").read_bytes() == b"hello" + assert [p.name for p in thread_uploads_dir.iterdir()] == ["notes.txt"] + + def test_upload_files_rejects_oversized_single_file_and_removes_partial_file(tmp_path): thread_uploads_dir = tmp_path / "uploads" thread_uploads_dir.mkdir(parents=True)