mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(config): keep atomic-write cleanup from replacing the caller's outcome (#7279)
* fix(config): keep atomic-write cleanup from replacing the caller's outcome * docs(config): state the cleanup invariant for every exit path
This commit is contained in:
parent
6b918bbe9a
commit
95ffb29e50
@ -25,6 +25,7 @@ from deerflow.config.managed_model_providers import get_managed_model_provider,
|
||||
from deerflow.config.model_config import ModelConfig, ReasoningCapabilities
|
||||
from deerflow.config.runtime_paths import runtime_home
|
||||
from deerflow.config.secret_file import read_or_create_secret_file
|
||||
from deerflow.utils.atomic_write import cleanup_failed_atomic_write
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from deerflow.config.app_config import AppConfig
|
||||
@ -224,7 +225,9 @@ class ManagedModelStore:
|
||||
os.replace(temporary, path)
|
||||
finally:
|
||||
if temporary is not None:
|
||||
temporary.unlink(missing_ok=True)
|
||||
# Discarding the temporary must not replace the outcome of the
|
||||
# write (or of a failed ``os.replace``) with a cleanup error.
|
||||
cleanup_failed_atomic_write(temporary)
|
||||
|
||||
|
||||
def merge_managed_models(config: AppConfig) -> AppConfig:
|
||||
|
||||
@ -36,6 +36,8 @@ import time
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
from deerflow.utils.atomic_write import cleanup_failed_atomic_write
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_POLL_INTERVAL_SECONDS = 0.05
|
||||
@ -147,7 +149,11 @@ def _create_exclusive(path: Path, generate: Callable[[], str]) -> bool:
|
||||
return _publish_under_claim(path, generate, replace_empty=False)
|
||||
return True
|
||||
finally:
|
||||
Path(temporary).unlink(missing_ok=True)
|
||||
# Removing our own temporary name is best-effort on every exit path —
|
||||
# whether the hard link published the value, a peer won the name, a
|
||||
# hard-link-less filesystem published through the replacement claim, or
|
||||
# an earlier step already failed — and must never mask that outcome.
|
||||
cleanup_failed_atomic_write(Path(temporary))
|
||||
|
||||
|
||||
def _claim_path(path: Path) -> Path:
|
||||
@ -186,4 +192,6 @@ def _publish_under_claim(path: Path, generate: Callable[[], str], *, replace_emp
|
||||
return True
|
||||
finally:
|
||||
if not published:
|
||||
claim.unlink(missing_ok=True)
|
||||
# Dropping our unpublished claim is best-effort: a failure here must
|
||||
# not replace the error that aborted the publication.
|
||||
cleanup_failed_atomic_write(claim)
|
||||
|
||||
92
backend/tests/test_config_atomic_cleanup.py
Normal file
92
backend/tests/test_config_atomic_cleanup.py
Normal file
@ -0,0 +1,92 @@
|
||||
"""Cleanup failures in the secret-file and managed-model writers.
|
||||
|
||||
Both writers stage a temporary file and then discard it. That removal runs on
|
||||
every exit path, not only after a successful publication: it also runs when the
|
||||
write, the close, or the replacement already failed. On Windows a scanner, a peer
|
||||
holding the handle, or a read-only mount can make the removal itself fail, so it
|
||||
has to stay best-effort — a cleanup error must never replace the outcome the
|
||||
caller needed, whether that outcome is a published value or an earlier error.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.config.managed_models import ManagedModelStore
|
||||
from deerflow.config.secret_file import read_or_create_secret_file
|
||||
|
||||
|
||||
def _fail_unlink_where(predicate: Callable[[Path], bool], monkeypatch: pytest.MonkeyPatch) -> list[Path]:
|
||||
"""Make ``Path.unlink`` fail for matching paths and record the attempts."""
|
||||
real_unlink = Path.unlink
|
||||
attempts: list[Path] = []
|
||||
|
||||
def unlink(self: Path, missing_ok: bool = False) -> None:
|
||||
if predicate(self):
|
||||
attempts.append(self)
|
||||
raise PermissionError(13, "The process cannot access the file", str(self))
|
||||
real_unlink(self, missing_ok=missing_ok)
|
||||
|
||||
monkeypatch.setattr(Path, "unlink", unlink)
|
||||
return attempts
|
||||
|
||||
|
||||
def test_secret_creation_survives_a_failing_temporary_removal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
path = tmp_path / ".secret"
|
||||
attempts = _fail_unlink_where(lambda candidate: candidate.suffix == ".tmp", monkeypatch)
|
||||
|
||||
value = read_or_create_secret_file(path, lambda: "generated-value")
|
||||
|
||||
assert value == "generated-value"
|
||||
assert path.read_text(encoding="utf-8") == "generated-value"
|
||||
assert attempts, "the temporary file should have been handed to the cleanup helper"
|
||||
|
||||
|
||||
def test_secret_creation_keeps_the_replace_error_when_cleanup_also_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
path = tmp_path / ".secret"
|
||||
path.write_text("", encoding="utf-8") # an abandoned empty file takes the claim path
|
||||
|
||||
def no_hard_links(*_args, **_kwargs):
|
||||
raise OSError(95, "Operation not supported")
|
||||
|
||||
def failing_replace(*_args, **_kwargs):
|
||||
raise OSError(5, "publish failed")
|
||||
|
||||
monkeypatch.setattr("deerflow.config.secret_file.os.link", no_hard_links)
|
||||
monkeypatch.setattr("deerflow.config.secret_file.os.replace", failing_replace)
|
||||
attempts = _fail_unlink_where(lambda candidate: candidate.suffix == ".replacing", monkeypatch)
|
||||
|
||||
with pytest.raises(OSError) as raised:
|
||||
read_or_create_secret_file(path, lambda: "generated-value")
|
||||
|
||||
assert "publish failed" in str(raised.value)
|
||||
assert attempts, "the unpublished claim should still have been cleaned up"
|
||||
|
||||
|
||||
def test_managed_model_write_survives_a_failing_temporary_removal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
target = tmp_path / "managed.json"
|
||||
attempts = _fail_unlink_where(lambda candidate: candidate.parent == tmp_path and candidate != target, monkeypatch)
|
||||
|
||||
ManagedModelStore._write(target, b"payload")
|
||||
|
||||
assert target.read_bytes() == b"payload"
|
||||
assert attempts
|
||||
|
||||
|
||||
def test_managed_model_write_keeps_the_replace_error_when_cleanup_also_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
target = tmp_path / "managed.json"
|
||||
|
||||
def failing_replace(*_args, **_kwargs):
|
||||
raise OSError(5, "publish failed")
|
||||
|
||||
monkeypatch.setattr("deerflow.config.managed_models.os.replace", failing_replace)
|
||||
attempts = _fail_unlink_where(lambda candidate: candidate.parent == tmp_path and candidate != target, monkeypatch)
|
||||
|
||||
with pytest.raises(OSError) as raised:
|
||||
ManagedModelStore._write(target, b"payload")
|
||||
|
||||
assert "publish failed" in str(raised.value)
|
||||
assert attempts
|
||||
Loading…
x
Reference in New Issue
Block a user