mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 05:06:59 +00:00
6ce385b963
* fix(uploads): stop Windows replacement uploads from being skipped by the multi-link guard NTFS never decrements st_nlink when one of a file's names is removed, so every upload published through os.link(staging, dest) + unlink keeps reading st_nlink == 2 on Windows. The non-exclusive replacement guards then reject that destination forever, and DeerFlowClient.upload_files turns the UnsafeUploadPathError into a log line - a replacement upload is silently skipped (Closes #6820). - Add _replacement_destination_is_hardlinked: POSIX-authoritative st_nlink > 1, skipped on Windows, used only on the not exclusive replacement path. The exclusive (first-claim) path keeps the strict count check, as does the POSIX O_NOFOLLOW branch. - Decide "published alias" in cleanup_stale_upload_staging_files by inode identity against siblings (st_dev/st_ino via os.lstat) instead of the link count; DirEntry.stat reports zeros for both on Windows, matching the existing _staging_entry_stat note. - Guard the POSIX-only st_nlink assertions in the affected tests and mark the three inbound-attachment mode-bit tests POSIX-only, mirroring the existing skipif in the same class. * fix(uploads): restore the exclusive post-open link guard and harden the staging sweep Address the three review findings on 5353715: - Restore the strict post-open st_nlink != 1 check for exclusive=True on the no-O_NOFOLLOW (Windows fallback) branch, which the previous commit dropped there against what the PR description claimed. Only the non-exclusive replacement path defers to the Windows-aware guard. - Never match a degenerate st_ino == 0 in the sibling-inode sweep: on FAT/exFAT or some SMB volumes every entry would look like a published alias and in-flight uploads would be deleted at any age. Such volumes fall back to the min_age guard only. - Build one {name: stat} map per uploads directory before the entry loop so the sweep scans each directory once instead of rescanning it for every staging file (O(N) instead of O(N^2) lstat calls). Tests: 407 passed / 27 skipped across test_uploads_manager.py, test_uploads_router.py, test_upload_copy_publication.py, test_upload_windows_staging_aliases.py, test_channel_file_attachments.py (Windows 11, Python 3.13 venv). The three new regression tests fail on 5353715 and pass with this change. * style(uploads): apply Ruff formatting * fix(uploads): report staging directory scan failures --------- Co-authored-by: sjr666666 <sjr666666@users.noreply.github.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
205 lines
7.6 KiB
Python
205 lines
7.6 KiB
Python
"""Upload publication rejects substituted staging paths and preserves outcomes."""
|
|
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from support.symlinks import symlink_or_skip
|
|
|
|
from deerflow.uploads.manager import UnsafeUploadPathError, copy_upload_file_no_replace
|
|
|
|
|
|
@pytest.fixture
|
|
def upload_paths(tmp_path):
|
|
uploads = tmp_path / "uploads"
|
|
uploads.mkdir()
|
|
source = tmp_path / "notes.txt"
|
|
source.write_bytes(b"uploaded content")
|
|
outside = tmp_path / "outside.txt"
|
|
outside.write_bytes(b"must not be published")
|
|
return uploads, source, outside
|
|
|
|
|
|
def replace_staging(staging, outside, replacement):
|
|
try:
|
|
staging.unlink()
|
|
except PermissionError:
|
|
if os.name == "nt":
|
|
pytest.skip("host prevents replacing an open staging file")
|
|
raise
|
|
if replacement == "symlink":
|
|
symlink_or_skip(staging, outside)
|
|
else:
|
|
staging.write_bytes(b"substituted content")
|
|
|
|
|
|
@pytest.mark.parametrize("replacement", ["symlink", "regular"])
|
|
@pytest.mark.parametrize("nofollow", [True, False])
|
|
def test_rejects_staging_replaced_during_copy(upload_paths, monkeypatch, replacement, nofollow):
|
|
uploads, source, outside = upload_paths
|
|
real_copy = shutil.copyfileobj
|
|
monkeypatch.setattr(os, "supports_follow_symlinks", {os.link} if nofollow else set())
|
|
|
|
def copy_and_swap(src, dst):
|
|
real_copy(src, dst)
|
|
staging = next(uploads.glob(".upload-*.part"))
|
|
replace_staging(staging, outside, replacement)
|
|
|
|
monkeypatch.setattr(shutil, "copyfileobj", copy_and_swap)
|
|
with pytest.raises(UnsafeUploadPathError):
|
|
copy_upload_file_no_replace(uploads, "notes.txt", source)
|
|
|
|
assert not os.path.lexists(uploads / "notes.txt")
|
|
assert outside.read_bytes() == b"must not be published"
|
|
assert outside.stat().st_nlink == 1
|
|
assert list(uploads.iterdir()) == []
|
|
|
|
|
|
@pytest.mark.parametrize("replacement", ["symlink", "regular"])
|
|
@pytest.mark.parametrize("nofollow", [True, False])
|
|
def test_rejects_staging_replaced_at_link(upload_paths, monkeypatch, replacement, nofollow):
|
|
uploads, source, outside = upload_paths
|
|
real_link = os.link
|
|
if nofollow and real_link not in os.supports_follow_symlinks:
|
|
pytest.skip("host does not support no-follow hard links")
|
|
calls = []
|
|
|
|
def swap_and_link(staging, dest, **kwargs):
|
|
replace_staging(staging, outside, replacement)
|
|
calls.append(kwargs)
|
|
real_link(staging, dest, **kwargs)
|
|
|
|
monkeypatch.setattr(os, "link", swap_and_link)
|
|
monkeypatch.setattr(os, "supports_follow_symlinks", {swap_and_link} if nofollow else set())
|
|
with pytest.raises(UnsafeUploadPathError):
|
|
copy_upload_file_no_replace(uploads, "notes.txt", source)
|
|
|
|
assert calls == [{"follow_symlinks": False} if nofollow else {}]
|
|
assert not os.path.lexists(uploads / "notes.txt")
|
|
assert outside.read_bytes() == b"must not be published"
|
|
assert outside.stat().st_nlink == 1
|
|
assert list(uploads.iterdir()) == []
|
|
|
|
|
|
@pytest.mark.parametrize("nofollow", [True, False])
|
|
def test_publishes_with_supported_link_options(upload_paths, monkeypatch, nofollow):
|
|
uploads, source, _outside = upload_paths
|
|
real_link = os.link
|
|
if nofollow and real_link not in os.supports_follow_symlinks:
|
|
pytest.skip("host does not support no-follow hard links")
|
|
calls = []
|
|
|
|
def record_link(staging, dest, **kwargs):
|
|
calls.append(kwargs)
|
|
real_link(staging, dest, **kwargs)
|
|
|
|
monkeypatch.setattr(os, "link", record_link)
|
|
monkeypatch.setattr(os, "supports_follow_symlinks", {record_link} if nofollow else set())
|
|
dest = copy_upload_file_no_replace(uploads, "notes.txt", source)
|
|
|
|
assert dest.read_bytes() == b"uploaded content"
|
|
if os.name != "nt":
|
|
# Windows never decrements the link count when the staged name is removed,
|
|
# so a freshly published upload reads back as ``st_nlink == 2`` there.
|
|
assert dest.stat().st_nlink == 1
|
|
assert calls == [{"follow_symlinks": False} if nofollow else {}]
|
|
|
|
|
|
def test_cleanup_directory_swap_does_not_mask_success(upload_paths, monkeypatch, caplog):
|
|
uploads, source, _outside = upload_paths
|
|
real_link = os.link
|
|
planted = []
|
|
|
|
def publish_and_swap(staging, dest, **kwargs):
|
|
real_link(staging, dest, **kwargs)
|
|
try:
|
|
staging.unlink()
|
|
except PermissionError:
|
|
if os.name == "nt":
|
|
pytest.skip("host prevents replacing an open staging file")
|
|
raise
|
|
staging.mkdir()
|
|
planted.append(staging)
|
|
|
|
monkeypatch.setattr(os, "link", publish_and_swap)
|
|
dest = copy_upload_file_no_replace(uploads, "notes.txt", source)
|
|
|
|
assert dest.read_bytes() == b"uploaded content"
|
|
assert planted[0].is_dir()
|
|
assert "Failed to remove upload staging file" in caplog.text
|
|
|
|
|
|
def test_cleanup_error_does_not_mask_copy_failure(upload_paths, monkeypatch, caplog):
|
|
uploads, source, _outside = upload_paths
|
|
real_unlink = Path.unlink
|
|
|
|
def fail_copy(src, dst):
|
|
dst.write(b"partial")
|
|
raise OSError("original copy failure")
|
|
|
|
def fail_cleanup(path, *args, **kwargs):
|
|
if path.name.startswith(".upload-"):
|
|
raise PermissionError("cleanup failure")
|
|
return real_unlink(path, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(shutil, "copyfileobj", fail_copy)
|
|
monkeypatch.setattr(Path, "unlink", fail_cleanup)
|
|
with pytest.raises(OSError, match="original copy failure"):
|
|
copy_upload_file_no_replace(uploads, "notes.txt", source)
|
|
|
|
assert not (uploads / "notes.txt").exists()
|
|
assert "Failed to remove upload staging file" in caplog.text
|
|
|
|
|
|
def test_cleanup_error_does_not_mask_collision(upload_paths, monkeypatch, caplog):
|
|
uploads, source, _outside = upload_paths
|
|
dest = uploads / "notes.txt"
|
|
dest.write_bytes(b"keep existing upload")
|
|
real_unlink = Path.unlink
|
|
|
|
def fail_cleanup(path, *args, **kwargs):
|
|
if path.name.startswith(".upload-"):
|
|
raise PermissionError("cleanup failure")
|
|
return real_unlink(path, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(Path, "unlink", fail_cleanup)
|
|
with pytest.raises(FileExistsError):
|
|
copy_upload_file_no_replace(uploads, "notes.txt", source)
|
|
|
|
assert dest.read_bytes() == b"keep existing upload"
|
|
assert "Failed to remove upload staging file" in caplog.text
|
|
|
|
|
|
@pytest.mark.parametrize("collisions", [0, 2])
|
|
def test_publication_retry_uses_original_seed(upload_paths, collisions):
|
|
from deerflow.uploads.manager import publish_upload_copy
|
|
|
|
uploads, source, _outside = upload_paths
|
|
seen = {"notes.txt", "notes_1.txt"}
|
|
for index in range(1, collisions + 1):
|
|
(uploads / f"notes_{index}.txt").write_bytes(f"existing-{index}".encode())
|
|
|
|
dest = publish_upload_copy(uploads, "notes_1.txt", source, claim_seed="notes.txt", seen_names=seen)
|
|
|
|
assert dest.name == f"notes_{collisions + 1}.txt"
|
|
assert dest.read_bytes() == b"uploaded content"
|
|
assert dest.name in seen
|
|
for index in range(1, collisions + 1):
|
|
assert (uploads / f"notes_{index}.txt").read_bytes() == f"existing-{index}".encode()
|
|
assert not list(uploads.glob(".upload-*.part"))
|
|
|
|
|
|
def test_publication_retry_preserves_noncollision_errors(upload_paths, monkeypatch):
|
|
from deerflow.uploads import manager
|
|
|
|
uploads, source, _outside = upload_paths
|
|
|
|
def fail_link(*args, **kwargs):
|
|
raise PermissionError("publication denied")
|
|
|
|
monkeypatch.setattr(manager.os, "link", fail_link)
|
|
with pytest.raises(PermissionError, match="publication denied"):
|
|
manager.publish_upload_copy(uploads, "notes.txt", source, claim_seed="notes.txt", seen_names={"notes.txt"})
|
|
assert list(uploads.iterdir()) == []
|