mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(scanner): cache decoded text atomically (#6510)
* fix(scanner): cache decoded text atomically _cache_text_file keyed the cache on the virtual path and returned the path, and the caller serves that path as the file's content, so a write that failed part-way left a truncated entry that every later read handed back as the preview. Publish through a temporary file plus os.replace. * test(scanner): pin the cache entry against a failed publish The helper-level case cannot distinguish the call site, so this drives the real entry point with a failing publish and asserts the cached entry is untouched. * docs(sandbox): keep agent guidance within its size limit * fix(scanner): preserve publish errors during cleanup --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
5ae953fc2f
commit
b5d921c255
@ -96,7 +96,7 @@ cancellation before propagating it.
|
||||
Health probes require an exact `ok` line and reject `Error:`/`Exit Code:` diagnostic lines; other login-shell stdout/stderr is tolerated. Failures log sandbox id and output before eviction.
|
||||
|
||||
|
||||
**Shared warm-pool lifecycle:** community sandbox providers that keep released sandboxes alive for fast reuse share `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin`. The mixin owns the common `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, idle-checker loop, warm-pool expiry, oldest-warm eviction, replica counting, and soft-cap logging. Providers remain responsible for their own active registries, creation/discovery, health checks, and destroy hook (`_destroy_warm_entry`): AIO destroys `SandboxInfo` through its backend; Boxlite closes loop-affine `BoxliteBox` handles; Tenki closes the microVM session (`TenkiSandbox.close`, which terminates the remote sandbox). AIO keeps active-idle cleanup outside the mixin and delegates only warm-pool expiry to the shared helper. `_stop_idle_checker()` must confirm thread exit before provider teardown. A bounded-join timeout raises; reset the provider shutdown guard so shutdown can retry.
|
||||
**Shared warm-pool lifecycle:** `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin` owns `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, the idle-checker loop, warm expiry, oldest-warm eviction, replica counts and soft-cap logs. Providers own active registries, creation/discovery, health checks and `_destroy_warm_entry`: AIO destroys `SandboxInfo` via its backend; Boxlite closes loop-affine `BoxliteBox`; Tenki terminates the remote VM with `TenkiSandbox.close`. AIO delegates only warm expiry; active-idle cleanup stays outside the mixin. Shutdown must await the idle-checker thread before tearing down provider resources. `_stop_idle_checker()` uses a bounded join and raises on timeout; reset the provider's shutdown guard on failure so shutdown can retry.
|
||||
|
||||
**Virtual Path System**:
|
||||
- Agent sees: `/mnt/user-data/{workspace,uploads,outputs}`, `/mnt/skills`
|
||||
|
||||
@ -3,6 +3,7 @@ from __future__ import annotations
|
||||
import fnmatch
|
||||
import hashlib
|
||||
import os
|
||||
import tempfile
|
||||
from codecs import BOM_UTF16_BE, BOM_UTF16_LE, getincrementaldecoder
|
||||
from pathlib import Path
|
||||
|
||||
@ -323,10 +324,28 @@ def _snapshot_symlink(root: WorkspaceRoot, host_file: Path) -> FileSnapshot | No
|
||||
)
|
||||
|
||||
|
||||
def _publish_text_atomically(path: Path, content: str) -> None:
|
||||
"""Publish text through a temporary file so a partial write cannot survive."""
|
||||
fd, temporary = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp")
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
handle.write(content)
|
||||
handle.flush()
|
||||
os.fsync(handle.fileno())
|
||||
os.replace(temporary, path)
|
||||
except BaseException:
|
||||
try:
|
||||
Path(temporary).unlink(missing_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
|
||||
|
||||
def _cache_text_file(text: str, virtual_path: str, cache_dir: Path) -> Path:
|
||||
"""Cache decoded text atomically; the caller serves this path as file content."""
|
||||
cache_name = hashlib.sha256(virtual_path.encode("utf-8")).hexdigest()
|
||||
target = cache_dir / cache_name
|
||||
target.write_text(text, encoding="utf-8")
|
||||
_publish_text_atomically(target, text)
|
||||
return target
|
||||
|
||||
|
||||
|
||||
@ -199,3 +199,10 @@ or when the filesystem does not enforce those bits. Cover both persisted
|
||||
secrets, shell/Compose dotenv overrides, readable read-only secrets, and teardown
|
||||
without secret exports. `down` must not probe, read, or generate secrets.
|
||||
Writable-directory cases remain portable.
|
||||
|
||||
## Workspace text cache
|
||||
|
||||
`test_scanner_text_cache_atomic.py` verifies complete publication and preservation
|
||||
of an existing entry on failure. A cleanup failure must preserve the original
|
||||
publish exception, including interruption. Import the scanner during fixture
|
||||
setup, after the autouse fixtures initialize runtime, to avoid the package cycle.
|
||||
|
||||
115
backend/tests/test_scanner_text_cache_atomic.py
Normal file
115
backend/tests/test_scanner_text_cache_atomic.py
Normal file
@ -0,0 +1,115 @@
|
||||
# SPDX-License-Identifier: MIT
|
||||
"""A cached preview must never be served half written."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scanner():
|
||||
# Import after the autouse fixtures initialize the runtime package.
|
||||
from deerflow.workspace_changes import scanner
|
||||
|
||||
return scanner
|
||||
|
||||
|
||||
def test_publishes_the_text_and_leaves_no_temp_file(tmp_path, scanner):
|
||||
target = tmp_path / "entry"
|
||||
|
||||
scanner._publish_text_atomically(target, "decoded body")
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "decoded body"
|
||||
assert list(tmp_path.glob("*.tmp")) == []
|
||||
|
||||
|
||||
def test_a_failed_publish_keeps_the_previous_entry(tmp_path, monkeypatch, scanner):
|
||||
target = tmp_path / "entry"
|
||||
target.write_text("previous body", encoding="utf-8")
|
||||
|
||||
def _boom(*args, **kwargs):
|
||||
raise OSError("disk full")
|
||||
|
||||
monkeypatch.setattr(os, "replace", _boom)
|
||||
|
||||
with pytest.raises(OSError):
|
||||
scanner._publish_text_atomically(target, "new body")
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "previous body"
|
||||
assert list(tmp_path.glob("*.tmp")) == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("publish_error", [OSError("publish failed"), KeyboardInterrupt("publish interrupted")], ids=["io-error", "interrupt"])
|
||||
def test_cleanup_failure_preserves_the_publish_error(tmp_path, monkeypatch, scanner, publish_error):
|
||||
target = tmp_path / "entry"
|
||||
target.write_text("previous body", encoding="utf-8")
|
||||
|
||||
def _fail_publish(*args, **kwargs):
|
||||
raise publish_error
|
||||
|
||||
def _fail_cleanup(*args, **kwargs):
|
||||
raise PermissionError("cleanup failed")
|
||||
|
||||
with monkeypatch.context() as patches:
|
||||
patches.setattr(os, "replace", _fail_publish)
|
||||
patches.setattr(Path, "unlink", _fail_cleanup)
|
||||
|
||||
with pytest.raises(type(publish_error)) as caught:
|
||||
scanner._publish_text_atomically(target, "new body")
|
||||
|
||||
assert caught.value is publish_error
|
||||
assert target.read_text(encoding="utf-8") == "previous body"
|
||||
|
||||
|
||||
def test_truncating_write_would_have_destroyed_it(tmp_path):
|
||||
"""Control: the previous implementation loses the entry in the same failure."""
|
||||
target = tmp_path / "entry"
|
||||
target.write_text("previous body", encoding="utf-8")
|
||||
|
||||
try:
|
||||
with open(target, "w", encoding="utf-8") as handle:
|
||||
handle.write("new ")
|
||||
raise OSError("disk full")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
assert target.read_text(encoding="utf-8") != "previous body"
|
||||
|
||||
|
||||
def test_cache_text_file_returns_a_complete_entry(tmp_path, scanner):
|
||||
cache_dir = tmp_path / "cache"
|
||||
cache_dir.mkdir()
|
||||
virtual_path = "/workspace/report.csv"
|
||||
|
||||
returned = scanner._cache_text_file("a,b\n1,2\n", virtual_path, cache_dir)
|
||||
|
||||
assert Path(returned).read_text(encoding="utf-8") == "a,b\n1,2\n"
|
||||
# The key is derived from the virtual path, so a partial entry would be served
|
||||
# as the file's content on every later read.
|
||||
assert Path(returned).name == hashlib.sha256(virtual_path.encode()).hexdigest()
|
||||
|
||||
|
||||
def test_cache_entry_survives_a_failed_publish(tmp_path, monkeypatch, scanner):
|
||||
"""The real entry point: a failed publish must not damage the cached entry."""
|
||||
cache_dir = tmp_path / "cache"
|
||||
cache_dir.mkdir()
|
||||
virtual_path = "/workspace/report.csv"
|
||||
name = hashlib.sha256(virtual_path.encode()).hexdigest()
|
||||
entry = cache_dir / name
|
||||
entry.write_text("previous body", encoding="utf-8")
|
||||
|
||||
def _boom(*args, **kwargs):
|
||||
raise OSError("disk full")
|
||||
|
||||
monkeypatch.setattr(os, "replace", _boom)
|
||||
|
||||
with pytest.raises(OSError):
|
||||
scanner._cache_text_file("new body", virtual_path, cache_dir)
|
||||
|
||||
# A truncating write would have emptied the entry before failing, and the
|
||||
# caller serves this path as the file's content on every later read.
|
||||
assert entry.read_text(encoding="utf-8") == "previous body"
|
||||
Loading…
x
Reference in New Issue
Block a user