fix(client): keep a decided batch outcome through capture cleanup (#7517)

This commit is contained in:
llb555 2026-10-26 17:03:45 +08:00 committed by GitHub
parent d48ea2a9a3
commit 794be98704
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 119 additions and 3 deletions

View File

@ -1777,6 +1777,7 @@ class DeerFlowClient:
"""
validate_thread_id(thread_id)
from deerflow.uploads.companions import invalidate_overwritten_upload, register_companion
from deerflow.utils.atomic_write import cleanup_failed_atomic_write
from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown
# Validate all files upfront to avoid partial uploads.
@ -1824,7 +1825,7 @@ class DeerFlowClient:
conversion_source = None
if src_path.suffix.lower() in CONVERTIBLE_EXTENSIONS:
if conversion_sources is None:
conversion_sources = tempfile.TemporaryDirectory(prefix="deerflow-upload-source-")
conversion_sources = tempfile.TemporaryDirectory(prefix="deerflow-upload-source-", ignore_cleanup_errors=True)
conversion_source = Path(conversion_sources.name) / dest_name
if dest_name in existing_names:
if conversion_source is not None:
@ -1836,7 +1837,9 @@ class DeerFlowClient:
dest_name = dest.name
except UnsafeUploadPathError:
if conversion_source is not None:
conversion_source.unlink(missing_ok=True)
# Skipping one unsafe file must survive a failed discard,
# or a locked temp turns the skip into a batch failure.
cleanup_failed_atomic_write(conversion_source)
logger.warning("Skipping upload with unsafe destination: %s", dest_name)
skipped_files.append(dest_name)
continue
@ -1905,12 +1908,16 @@ class DeerFlowClient:
seen_names.discard(unique_md_name)
if conversion_source is not None:
conversion_source.unlink(missing_ok=True)
# The upload already succeeded; a capture we cannot discard
# must not replace that result or drop the later files.
cleanup_failed_atomic_write(conversion_source)
uploaded_files.append(info)
finally:
if conversion_pool is not None:
conversion_pool.shutdown(wait=True)
if conversion_sources is not None:
# ignore_cleanup_errors keeps a locked capture from turning a
# decided batch outcome into a cleanup error.
conversion_sources.cleanup()
message = f"Successfully uploaded {len(uploaded_files)} file(s)"

View File

@ -0,0 +1,109 @@
"""An unsafe upload must stay a skip even when its capture cannot be discarded.
``upload_files`` rejects a destination that resolves outside the uploads
directory and moves on to the next file. On Windows the staged conversion
capture can still be locked at that point, so an unguarded discard turned the
skip into a ``PermissionError`` for the whole batch.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from deerflow.client import DeerFlowClient
from deerflow.uploads.manager import UnsafeUploadPathError
@pytest.fixture
def upload_case(monkeypatch, tmp_path):
import deerflow.client as client_module
uploads = tmp_path / "thread/user-data/uploads"
uploads.mkdir(parents=True)
monkeypatch.setattr(client_module, "ensure_uploads_dir", lambda *_args, **_kwargs: uploads)
return DeerFlowClient.__new__(DeerFlowClient), uploads, client_module
def test_unsafe_upload_stays_a_skip_when_discarding_its_capture_fails(upload_case, tmp_path, monkeypatch):
client, uploads, client_module = upload_case
unsafe = tmp_path / "unsafe.pdf"
unsafe.write_bytes(b"%PDF unsafe")
keep = tmp_path / "keep.txt"
keep.write_bytes(b"keep me")
real_publish = client_module.publish_upload_copy
def publish(uploads_dir, dest_name, src_path, **_kwargs):
if Path(src_path).name == unsafe.name:
raise UnsafeUploadPathError("destination escapes the uploads directory")
return real_publish(uploads_dir, dest_name, src_path, **_kwargs)
monkeypatch.setattr(client_module, "publish_upload_copy", publish)
real_unlink = Path.unlink
def unlink(self, *args, **kwargs):
if "deerflow-upload-source-" in str(self):
raise PermissionError("capture is still open")
return real_unlink(self, *args, **kwargs)
monkeypatch.setattr(Path, "unlink", unlink)
result = client.upload_files("skip-thread", [unsafe, keep])
assert result["skipped_files"] == [unsafe.name]
assert [item["filename"] for item in result["files"]] == [keep.name]
assert (uploads / keep.name).read_bytes() == b"keep me"
def test_successful_upload_survives_a_capture_it_cannot_discard(upload_case, tmp_path, monkeypatch):
"""A locked capture must not replace a completed upload or drop later files."""
client, uploads, _client_module = upload_case
convertible = tmp_path / "report.pdf"
convertible.write_bytes(b"%PDF report")
keep = tmp_path / "keep.txt"
keep.write_bytes(b"keep me")
real_unlink = Path.unlink
def unlink(self, *args, **kwargs):
if "deerflow-upload-source-" in str(self):
raise PermissionError("capture is still open")
return real_unlink(self, *args, **kwargs)
monkeypatch.setattr(Path, "unlink", unlink)
result = client.upload_files("successful-thread", [convertible, keep])
assert result["skipped_files"] == []
assert [item["filename"] for item in result["files"]] == [convertible.name, keep.name]
assert (uploads / keep.name).read_bytes() == b"keep me"
def test_capture_directory_cleanup_failure_does_not_fail_the_batch(upload_case, tmp_path, monkeypatch):
"""The ``finally`` rmtree must not replace an outcome the batch already decided."""
import tempfile
client, _uploads, _client_module = upload_case
convertible = tmp_path / "report.pdf"
convertible.write_bytes(b"%PDF report")
real_rmtree = tempfile.TemporaryDirectory._rmtree.__func__
def rmtree(cls, name, ignore_errors=False, repeated=False):
# A locked capture: exactly what CPython's onexc re-raises when the
# directory's cleanup is not told to ignore errors.
if "deerflow-upload-source-" in str(name) and not ignore_errors:
raise PermissionError("capture directory is still open")
return real_rmtree(cls, name, ignore_errors=ignore_errors, repeated=repeated)
monkeypatch.setattr(tempfile.TemporaryDirectory, "_rmtree", classmethod(rmtree))
result = client.upload_files("cleanup-thread", [convertible])
assert [item["filename"] for item in result["files"]] == [convertible.name]