mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-27 13:40:37 +00:00
fix(gateway,integrations): keep temp cleanup from replacing the caller's outcome (#7505)
* fix(gateway,integrations): keep temp cleanup from replacing the caller's outcome * fix(gateway): adopt the family helper so a published edit stays silent
This commit is contained in:
parent
21fb080530
commit
885062e197
@ -25,6 +25,7 @@ from deerflow.config.paths import make_safe_user_id
|
||||
from deerflow.runtime import ConflictError, ThreadOperationKind
|
||||
from deerflow.runtime.user_context import get_effective_user_id
|
||||
from deerflow.sandbox.sandbox_provider import get_sandbox_provider
|
||||
from deerflow.utils.atomic_write import cleanup_failed_atomic_write
|
||||
from deerflow.utils.file_io import await_drained
|
||||
from deerflow.utils.text_detection import _is_active_content_mime_type, is_text_file_by_content
|
||||
from deerflow.utils.thread_id import ThreadId
|
||||
@ -142,12 +143,10 @@ def _replace_artifact_atomically(actual_path: Path, content: bytes, file_stat: o
|
||||
# clearing the whole 256-entry LRU costs nothing (see PR review).
|
||||
_sha256_of_file_cached.cache_clear()
|
||||
finally:
|
||||
if temp_fd >= 0:
|
||||
os.close(temp_fd)
|
||||
try:
|
||||
temp_path.unlink()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
# The helper keeps the published case silent (the temp name is already
|
||||
# gone), warns when a leftover temp cannot be removed, and never lets a
|
||||
# cleanup error replace the edit's outcome.
|
||||
cleanup_failed_atomic_write(temp_path, temp_fd if temp_fd >= 0 else None)
|
||||
|
||||
|
||||
def _sync_artifact_to_sandbox(sandbox, virtual_path: str, content: bytes) -> None:
|
||||
|
||||
@ -1445,7 +1445,7 @@ def _write_json_file_atomic(path: Path, payload: dict[str, Any]) -> None:
|
||||
temp_path.chmod(0o600)
|
||||
os.replace(temp_path, path)
|
||||
finally:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
cleanup_failed_atomic_write(temp_path)
|
||||
|
||||
|
||||
_LARK_FLOW_DEVICE_CODE_UNSET: Any = object()
|
||||
|
||||
94
backend/tests/test_atomic_cleanup_writers.py
Normal file
94
backend/tests/test_atomic_cleanup_writers.py
Normal file
@ -0,0 +1,94 @@
|
||||
"""Staged-temp cleanup in the artifact and Lark writers must keep the real outcome.
|
||||
|
||||
Both writers publish through a same-directory temporary and discard it on exit.
|
||||
A failing discard used to replace the error the caller needed — the artifact
|
||||
editor returned a cleanup ``OSError`` instead of the failed write, and the Lark
|
||||
flow-state writer did the same.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def real_unlink():
|
||||
return Path.unlink
|
||||
|
||||
|
||||
def _fail_unlink_on(monkeypatch, predicate, real_unlink):
|
||||
"""Make ``Path.unlink`` raise for the paths a test owns."""
|
||||
|
||||
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 test_artifact_edit_keeps_the_publish_error_when_cleanup_fails(tmp_path, monkeypatch, real_unlink):
|
||||
from app.gateway.routers import artifacts
|
||||
|
||||
target = tmp_path / "note.txt"
|
||||
target.write_text("original", encoding="utf-8")
|
||||
file_stat = os.stat(target)
|
||||
|
||||
def _failing_replace(*_args, **_kwargs):
|
||||
raise OSError("replace failed")
|
||||
|
||||
monkeypatch.setattr(artifacts.os, "replace", _failing_replace)
|
||||
_fail_unlink_on(monkeypatch, lambda path: path.name.startswith(".") and path.suffix == "", real_unlink)
|
||||
|
||||
with pytest.raises(OSError, match="replace failed"):
|
||||
artifacts._replace_artifact_atomically(target, b"replacement", file_stat)
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "original"
|
||||
|
||||
|
||||
def test_artifact_edit_success_is_quiet(tmp_path, caplog):
|
||||
"""The published path must not warn: its temp name is already gone."""
|
||||
from app.gateway.routers import artifacts
|
||||
|
||||
target = tmp_path / "note.txt"
|
||||
target.write_text("original", encoding="utf-8")
|
||||
file_stat = os.stat(target)
|
||||
caplog.clear()
|
||||
|
||||
with caplog.at_level("WARNING", logger="deerflow.utils.atomic_write"):
|
||||
artifacts._replace_artifact_atomically(target, b"replacement", file_stat)
|
||||
|
||||
assert target.read_bytes() == b"replacement"
|
||||
assert [p.name for p in tmp_path.iterdir() if p.name.startswith(artifacts._ARTIFACT_EDIT_TEMP_PREFIX)] == []
|
||||
assert caplog.text == ""
|
||||
|
||||
|
||||
def test_lark_flow_state_write_keeps_the_write_error_when_cleanup_fails(tmp_path, monkeypatch, real_unlink):
|
||||
from deerflow.integrations import lark_cli
|
||||
|
||||
target = tmp_path / "flow-state.json"
|
||||
|
||||
def _failing_replace(*_args, **_kwargs):
|
||||
raise OSError("replace failed")
|
||||
|
||||
monkeypatch.setattr(lark_cli.os, "replace", _failing_replace)
|
||||
_fail_unlink_on(monkeypatch, lambda path: path.name.startswith(".flow-state.json"), real_unlink)
|
||||
|
||||
with pytest.raises(OSError, match="replace failed"):
|
||||
lark_cli._write_json_file_atomic(target, {"state": "pending"})
|
||||
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
def test_lark_flow_state_write_still_succeeds(monkeypatch, tmp_path):
|
||||
"""The success path keeps its behaviour: payload written, no temp left."""
|
||||
from deerflow.integrations import lark_cli
|
||||
|
||||
target = tmp_path / "flow-state.json"
|
||||
lark_cli._write_json_file_atomic(target, {"state": "pending"})
|
||||
|
||||
assert target.read_text(encoding="utf-8") == '{"state": "pending"}\n'
|
||||
assert [p.name for p in tmp_path.iterdir() if p.name.startswith(".flow-state")] == []
|
||||
Loading…
x
Reference in New Issue
Block a user