From 582b1f5a28b9415085d353664063499326441dde Mon Sep 17 00:00:00 2001 From: llb555 <3103086905@qq.com> Date: Thu, 22 Oct 2026 16:03:43 +0800 Subject: [PATCH] fix(uploads,integrations): keep atomic-write cleanup from replacing the caller's outcome (#7200) * fix(uploads,integrations): keep atomic-write cleanup from replacing the caller's outcome * fix(lark): keep the download error when archive cleanup fails * test(lark): keep the download fixtures inside the test directory --- .../deerflow/integrations/lark_broker.py | 15 +- .../harness/deerflow/integrations/lark_cli.py | 14 +- .../harness/deerflow/uploads/AGENTS.md | 9 + .../harness/deerflow/uploads/companions.py | 15 +- ..._atomic_cleanup_preserves_primary_error.py | 356 ++++++++++++++++++ 5 files changed, 393 insertions(+), 16 deletions(-) create mode 100644 backend/tests/test_atomic_cleanup_preserves_primary_error.py diff --git a/backend/packages/harness/deerflow/integrations/lark_broker.py b/backend/packages/harness/deerflow/integrations/lark_broker.py index e2b0f52993..d4eddfcbb8 100644 --- a/backend/packages/harness/deerflow/integrations/lark_broker.py +++ b/backend/packages/harness/deerflow/integrations/lark_broker.py @@ -29,8 +29,11 @@ import threading import time from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path from typing import Any +from deerflow.utils.atomic_write import cleanup_failed_atomic_write + logger = logging.getLogger(__name__) # ── Loopback wire contract ──────────────────────────────────────────────── @@ -471,19 +474,19 @@ def _write_text_atomically(path: str, content: str, *, mode: int) -> None: """ directory = os.path.dirname(os.path.abspath(path)) os.makedirs(directory, exist_ok=True) - fd, temporary = tempfile.mkstemp(dir=directory, prefix=f".{os.path.basename(path)}.", suffix=".tmp") + fd, temporary_name = tempfile.mkstemp(dir=directory, prefix=f".{os.path.basename(path)}.", suffix=".tmp") + temporary = Path(temporary_name) try: - with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle = os.fdopen(fd, "w", encoding="utf-8") + fd = None # The file object now owns the descriptor. + with handle: handle.write(content) handle.flush() os.chmod(temporary, mode) os.fsync(handle.fileno()) os.replace(temporary, path) except BaseException: - try: - os.unlink(temporary) - except OSError: - pass + cleanup_failed_atomic_write(temporary, fd) raise diff --git a/backend/packages/harness/deerflow/integrations/lark_cli.py b/backend/packages/harness/deerflow/integrations/lark_cli.py index 86aae5ddf2..9dfdfd8e06 100644 --- a/backend/packages/harness/deerflow/integrations/lark_cli.py +++ b/backend/packages/harness/deerflow/integrations/lark_cli.py @@ -91,6 +91,7 @@ from deerflow.skills.package_files import EXEC_SNIFF_BYTES, is_executable_binary from deerflow.skills.parser import parse_skill_file from deerflow.skills.permissions import make_skill_tree_sandbox_readable from deerflow.skills.types import SKILL_MD_FILE, SkillCategory +from deerflow.utils.atomic_write import cleanup_failed_atomic_write logger = logging.getLogger(__name__) @@ -1272,15 +1273,18 @@ def _publish_bytes_atomically(path: Path, content: bytes) -> None: ``make_skill_tree_sandbox_readable``. The parent directory is not fsynced, so this helper alone does not guarantee publication survives power loss. """ - fd, temporary = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp") + fd, temporary_name = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp") + temporary = Path(temporary_name) try: - with os.fdopen(fd, "wb") as handle: + handle = os.fdopen(fd, "wb") + fd = None # The file object now owns the descriptor. + with handle: handle.write(content) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) except BaseException: - Path(temporary).unlink(missing_ok=True) + cleanup_failed_atomic_write(temporary, fd) raise @@ -3468,10 +3472,10 @@ def _download_lark_archive(version: str) -> Path: raise ValueError("Lark CLI source archive is too large.") out.write(chunk) except ValueError: - archive_path.unlink(missing_ok=True) + cleanup_failed_atomic_write(archive_path) raise except Exception as exc: # noqa: BLE001 - network boundary - archive_path.unlink(missing_ok=True) + cleanup_failed_atomic_write(archive_path) raise ValueError(f"Could not download the Lark skill pack ({version}) from GitHub. Check the Gateway's internet access, or pre-stage the archive via {LARK_CLI_SOURCE_ARCHIVE_ENV}.") from exc return archive_path diff --git a/backend/packages/harness/deerflow/uploads/AGENTS.md b/backend/packages/harness/deerflow/uploads/AGENTS.md index 0885ed71ee..9e47dfff86 100644 --- a/backend/packages/harness/deerflow/uploads/AGENTS.md +++ b/backend/packages/harness/deerflow/uploads/AGENTS.md @@ -33,3 +33,12 @@ regular files and symlinks, and remove a mismatched publication before reporting failure. Cleanup errors are logged without masking success or the original error. Keep publication/fallback coverage in `backend/tests/test_upload_copy_publication.py` and capture-through-collision cases in `backend/tests/test_client_upload_snapshot.py`. + +`companions.py::register_companion` publishes its ownership record through +`utils.atomic_write.cleanup_failed_atomic_write`. Do not restore a bare +`finally: os.unlink(staged)`: when `json.dump` or `os.replace` failed, a failing +unlink replaced that primary error, and the old `with os.fdopen(fd, ...)` leaked +the descriptor whenever `fdopen` itself failed. Pass the raw descriptor only +until `fdopen` succeeds, and keep cleanup best-effort so a leftover staged +record is reported rather than becoming the caller's error. Coverage: +`backend/tests/test_atomic_cleanup_preserves_primary_error.py`. diff --git a/backend/packages/harness/deerflow/uploads/companions.py b/backend/packages/harness/deerflow/uploads/companions.py index cbfc783482..c82456ff46 100644 --- a/backend/packages/harness/deerflow/uploads/companions.py +++ b/backend/packages/harness/deerflow/uploads/companions.py @@ -13,6 +13,8 @@ import os import tempfile from pathlib import Path +from deerflow.utils.atomic_write import cleanup_failed_atomic_write + logger = logging.getLogger(__name__) @@ -38,8 +40,11 @@ def register_companion(original: Path, markdown: Path) -> None: source_stat = original.stat(follow_symlinks=False) markdown_stat = markdown.stat(follow_symlinks=False) fd, staged = tempfile.mkstemp(prefix=".companion-", dir=records_dir) + staged_path = Path(staged) try: - with os.fdopen(fd, "w", encoding="utf-8") as output: + output = os.fdopen(fd, "w", encoding="utf-8") + fd = None # The file object now owns the descriptor. + with output: json.dump( { "original": original.name, @@ -49,10 +54,10 @@ def register_companion(original: Path, markdown: Path) -> None: }, output, ) - os.replace(staged, record) - finally: - if os.path.exists(staged): - os.unlink(staged) + os.replace(staged_path, record) + except BaseException: + cleanup_failed_atomic_write(staged_path, fd) + raise def unregister_companion(original: Path) -> None: diff --git a/backend/tests/test_atomic_cleanup_preserves_primary_error.py b/backend/tests/test_atomic_cleanup_preserves_primary_error.py new file mode 100644 index 0000000000..87efe7eb49 --- /dev/null +++ b/backend/tests/test_atomic_cleanup_preserves_primary_error.py @@ -0,0 +1,356 @@ +# SPDX-License-Identifier: MIT +"""Failed publications report their own error, not the cleanup's. + +Three writers publish through a same-directory temporary and then clean it up +inside the handler for a failure they are already reporting. Left unguarded, +two distinct defects appear: + +* a failing ``unlink`` replaces the error the caller sees — the real cause + survives only as ``__context__``, which the surrounding + ``logger.warning("...: %s", exc)`` formats never print; +* a ``with os.fdopen(fd, ...)`` that never binds leaks ``fd``, because the + ``with`` cannot close a descriptor it never received. + +``uploads/companions.py`` is the strongest case: after a successful +``os.replace`` the staged name was already gone, so the unguarded ``finally`` +only ever ran on the failure path, where its own error replaced the +``json.dump``/``os.replace`` one the caller needed. + +Sites are pinned separately because they own different parts of the pipeline. +The channel publishers (#7106), project staging (#7141), the extension manager +and the Gateway skill-archive copy (#7164) are deliberately not repeated here. +""" + +from __future__ import annotations + +import logging +import os +from pathlib import Path + +import pytest + + +class _PrimaryError(Exception): + """Stand-in for the failure a publisher is already reporting.""" + + +class _Records(logging.Handler): + def __init__(self): + super().__init__() + self.records = [] + + def emit(self, record): + self.records.append(record) + + +def _fail_unlink_on(monkeypatch, predicate, real_unlink): + """Make ``Path.unlink`` (and ``os.unlink``) raise for the paths we own.""" + + def unlink(self, *args, **kwargs): + if predicate(self): + raise PermissionError("temporary unlink failed") + return real_unlink(self, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", unlink) + + +def _capture_atomic_write_warnings(monkeypatch): + """Point ``deerflow.utils.atomic_write`` diagnostics at a recording handler.""" + from deerflow.utils import atomic_write + + records = _Records() + monkeypatch.setattr(atomic_write.logger, "handlers", [records]) + monkeypatch.setattr(atomic_write.logger, "propagate", False) + monkeypatch.setattr(atomic_write.logger, "level", logging.WARNING) + return records + + +@pytest.fixture +def real_unlink(): + return Path.unlink + + +def _is_temporary_of(target): + """Match the ``.{name}..tmp`` sibling these helpers allocate.""" + + def predicate(path: Path) -> bool: + return path.parent == target.parent and path.name.startswith(f".{target.name}.") and path.suffix == ".tmp" + + return predicate + + +# -------------------------------------------------------------------------- +# integrations/lark_cli.py::_publish_bytes_atomically +# -------------------------------------------------------------------------- + + +def _publish_lark(target, content): + from deerflow.integrations import lark_cli + + lark_cli._publish_bytes_atomically(target, content) + + +def test_lark_cli_publish_preserves_primary_error_when_unlink_fails(tmp_path, monkeypatch, real_unlink): + target = tmp_path / "lark-cli" + target.write_bytes(b"previous runtime") + primary = _PrimaryError("publication failed") + + def replace(source, path): + raise primary + + monkeypatch.setattr(os, "replace", replace) + _fail_unlink_on(monkeypatch, _is_temporary_of(target), real_unlink) + records = _capture_atomic_write_warnings(monkeypatch) + + with pytest.raises(_PrimaryError) as raised: + _publish_lark(target, b"new runtime") + + assert raised.value is primary + assert raised.value.__context__ is None + assert target.read_bytes() == b"previous runtime" + # The leftover temporary is named in the diagnostic instead of silently + # becoming the reported failure. + assert len(records.records) == 1 + assert records.records[0].exc_info is not None + assert any(_is_temporary_of(target)(path) for path in tmp_path.iterdir()) + + +def test_lark_cli_publish_closes_the_descriptor_when_fdopen_fails(tmp_path, monkeypatch): + target = tmp_path / "lark-cli" + primary = _PrimaryError("fdopen failed") + allocated: list[int] = [] + + def fdopen(fd, *args, **kwargs): + allocated.append(fd) + raise primary + + monkeypatch.setattr(os, "fdopen", fdopen) + records = _capture_atomic_write_warnings(monkeypatch) + + with pytest.raises(_PrimaryError) as raised: + _publish_lark(target, b"new runtime") + + assert raised.value is primary + assert len(allocated) == 1 + # A leaked descriptor is still open, so fstat would succeed. + with pytest.raises(OSError): + os.fstat(allocated[0]) + assert not target.exists() + assert list(tmp_path.iterdir()) == [] + assert records.records == [] + + +def test_lark_cli_publish_success_is_unchanged(tmp_path): + target = tmp_path / "lark-cli" + + _publish_lark(target, b"#!/bin/sh\n") + + assert target.read_bytes() == b"#!/bin/sh\n" + assert list(tmp_path.iterdir()) == [target] + + +# -------------------------------------------------------------------------- +# integrations/lark_broker.py::_write_text_atomically +# -------------------------------------------------------------------------- + + +def _write_broker_launcher(path, content): + from deerflow.integrations import lark_broker + + lark_broker._write_text_atomically(str(path), content, mode=0o755) + + +def test_lark_broker_launcher_closes_the_descriptor_when_fdopen_fails(tmp_path, monkeypatch): + target = tmp_path / "bin" / "lark-cli" + target.parent.mkdir() + primary = _PrimaryError("fdopen failed") + allocated: list[int] = [] + + def fdopen(fd, *args, **kwargs): + allocated.append(fd) + raise primary + + monkeypatch.setattr(os, "fdopen", fdopen) + records = _capture_atomic_write_warnings(monkeypatch) + + with pytest.raises(_PrimaryError) as raised: + _write_broker_launcher(target, "#!/bin/sh\n") + + assert raised.value is primary + assert len(allocated) == 1 + with pytest.raises(OSError): + os.fstat(allocated[0]) + assert list(target.parent.iterdir()) == [] + assert records.records == [] + + +def test_lark_broker_launcher_preserves_primary_error_when_unlink_fails(tmp_path, monkeypatch, real_unlink): + target = tmp_path / "bin" / "lark-cli" + target.parent.mkdir() + target.write_text("old launcher", encoding="utf-8") + primary = _PrimaryError("chmod failed") + + def chmod(path, mode, *args, **kwargs): + raise primary + + monkeypatch.setattr(os, "chmod", chmod) + _fail_unlink_on(monkeypatch, _is_temporary_of(target), real_unlink) + records = _capture_atomic_write_warnings(monkeypatch) + + with pytest.raises(_PrimaryError) as raised: + _write_broker_launcher(target, "new launcher") + + assert raised.value is primary + assert raised.value.__context__ is None + assert target.read_text(encoding="utf-8") == "old launcher" + # Previously this cleanup failure was swallowed by `except OSError: pass`, + # leaving an unobservable leftover; now it is reported. + assert len(records.records) == 1 + assert any(_is_temporary_of(target)(path) for path in target.parent.iterdir()) + + +def test_lark_broker_launcher_success_is_unchanged(tmp_path): + target = tmp_path / "bin" / "lark-cli" + target.parent.mkdir() + + _write_broker_launcher(target, "#!/bin/sh\nexec lark-cli\n") + + assert target.read_text(encoding="utf-8") == "#!/bin/sh\nexec lark-cli\n" + assert list(target.parent.iterdir()) == [target] + + +# -------------------------------------------------------------------------- +# uploads/companions.py::register_companion +# -------------------------------------------------------------------------- + + +def _prepare_companion(tmp_path): + uploads = tmp_path / "thread" / "uploads" + uploads.mkdir(parents=True) + original = uploads / "report.pdf" + original.write_bytes(b"source") + markdown = uploads / "report.md" + markdown.write_text("# converted", encoding="utf-8") + return original, markdown + + +def _staged_predicate(uploads_dir): + """The staged record lives in ``/upload-companions/.companion-*``.""" + + def predicate(path: Path) -> bool: + return path.name.startswith(".companion-") and path.parent.name == "upload-companions" + + return predicate + + +def test_companion_registration_preserves_error_when_staged_unlink_fails(tmp_path, monkeypatch, real_unlink): + from deerflow.uploads import companions + + original, markdown = _prepare_companion(tmp_path) + primary = _PrimaryError("json dump failed") + + def dump(payload, stream, **kwargs): + stream.write("{") + raise primary + + monkeypatch.setattr(companions.json, "dump", dump) + _fail_unlink_on(monkeypatch, _staged_predicate(original.parent), real_unlink) + records = _capture_atomic_write_warnings(monkeypatch) + + with pytest.raises(_PrimaryError) as raised: + companions.register_companion(original, markdown) + + assert raised.value is primary + assert raised.value.__context__ is None + assert len(records.records) == 1 + leftovers = [p for p in (tmp_path / "thread" / "upload-companions").iterdir() if p.name.startswith(".companion-")] + assert leftovers, "the failed staged record is left behind but reported" + + +def test_companion_registration_success_is_unchanged(tmp_path): + from deerflow.uploads import companions + + original, markdown = _prepare_companion(tmp_path) + + companions.register_companion(original, markdown) + + records_dir = tmp_path / "thread" / "upload-companions" + assert [p.name for p in records_dir.iterdir() if p.name.endswith(".json")] or list(records_dir.glob("*.json")) + assert not list(records_dir.glob(".companion-*")) + assert companions.resolve_companion(original) == markdown + + +def test_companion_registration_still_succeeds_when_publish_cleanup_would_fail(tmp_path, monkeypatch, real_unlink): + """The success path must stay silent: nothing to clean up after a good replace.""" + from deerflow.uploads import companions + + original, markdown = _prepare_companion(tmp_path) + records = _capture_atomic_write_warnings(monkeypatch) + + companions.register_companion(original, markdown) + + assert companions.resolve_companion(original) == markdown + assert records.records == [] + + +class _ArchiveResponse: + """A download response that trips the archive size guard.""" + + def __init__(self, chunk: bytes) -> None: + self._chunk = chunk + + def __enter__(self) -> _ArchiveResponse: + return self + + def __exit__(self, *_exc: object) -> bool: + return False + + def read(self, _size: int) -> bytes: + return self._chunk + + +def _staged_archive_in(tmp_path): + """Keep the download's temporary archive inside the test's own directory.""" + import tempfile as tempfile_module + + # Bind the real allocator first: ``lark_cli.tempfile`` is this same module, + # so a late lookup would call the patched attribute and recurse forever. + real_mkstemp = tempfile_module.mkstemp + + def mkstemp(*_args, **_kwargs): + return real_mkstemp(dir=tmp_path, prefix="lark-cli-skills-", suffix=".zip") + + return mkstemp + + +def test_lark_archive_too_large_keeps_its_error_when_cleanup_fails(tmp_path, monkeypatch, real_unlink): + """The size guard's ValueError must survive a failing download cleanup.""" + from deerflow.integrations import lark_cli + + oversized = b"x" * (lark_cli.LARK_CLI_MAX_ARCHIVE_BYTES + 1) + monkeypatch.setattr( + lark_cli.urllib.request, + "urlopen", + lambda *_args, **_kwargs: _ArchiveResponse(oversized), + ) + monkeypatch.setattr(lark_cli.tempfile, "mkstemp", _staged_archive_in(tmp_path)) + _fail_unlink_on(monkeypatch, lambda path: path.suffix == ".zip", real_unlink) + + with pytest.raises(ValueError, match="too large"): + lark_cli._download_lark_archive("v1.2.3") + + +def test_lark_archive_download_error_keeps_its_message_when_cleanup_fails(tmp_path, monkeypatch, real_unlink): + """The network boundary's ValueError must survive a failing download cleanup.""" + + def _offline(*_args, **_kwargs): + raise OSError("network down") + + from deerflow.integrations import lark_cli + + monkeypatch.setattr(lark_cli.urllib.request, "urlopen", _offline) + monkeypatch.setattr(lark_cli.tempfile, "mkstemp", _staged_archive_in(tmp_path)) + _fail_unlink_on(monkeypatch, lambda path: path.suffix == ".zip", real_unlink) + + with pytest.raises(ValueError, match="Could not download the Lark skill pack"): + lark_cli._download_lark_archive("v1.2.3")