mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(storage): tolerate blob sidecar allocation failures (#6538)
* test(storage): cover sidecar allocation failures * fix(storage): tolerate sidecar allocation failures --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
0c5b25b4ab
commit
a8250eaa4b
@ -239,8 +239,9 @@ class LocalFsBlobStore(BlobStore):
|
||||
"writer_thread_id": thread_id,
|
||||
"created_at": time.time(),
|
||||
}
|
||||
fd, tmp_name = tempfile.mkstemp(prefix=".meta-", suffix=".tmp", dir=meta_path.parent)
|
||||
tmp_name: str | None = None
|
||||
try:
|
||||
fd, tmp_name = tempfile.mkstemp(prefix=".meta-", suffix=".tmp", dir=meta_path.parent)
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
json.dump(payload, handle)
|
||||
os.replace(tmp_name, meta_path)
|
||||
@ -248,10 +249,11 @@ class LocalFsBlobStore(BlobStore):
|
||||
# The sidecar is an optimization for later GC, not a read
|
||||
# dependency: losing it must not fail a put that already
|
||||
# succeeded.
|
||||
try:
|
||||
os.unlink(tmp_name)
|
||||
except OSError:
|
||||
pass
|
||||
if tmp_name is not None:
|
||||
try:
|
||||
os.unlink(tmp_name)
|
||||
except OSError:
|
||||
pass
|
||||
logger.warning("Could not write blob sidecar %s", meta_path, exc_info=True)
|
||||
|
||||
|
||||
|
||||
@ -7,6 +7,7 @@ wiring PR for issue #4189 item 2.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
import hashlib
|
||||
import json
|
||||
import threading
|
||||
@ -32,6 +33,7 @@ from deerflow.storage import (
|
||||
)
|
||||
from deerflow.storage import contract as contract_module
|
||||
from deerflow.storage.backends.local_fs import LocalFsBlobStore
|
||||
from deerflow.storage.backends.local_fs import local_fs_store as local_fs_module
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
@ -273,6 +275,63 @@ def test_put_survives_a_lost_sidecar(store: LocalFsBlobStore):
|
||||
assert store.get_bytes(ref) == data
|
||||
|
||||
|
||||
@pytest.mark.parametrize("existing", [False, True], ids=["new-put", "retry-put"])
|
||||
def test_put_survives_sidecar_allocation_failure(store: LocalFsBlobStore, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, existing: bool):
|
||||
"""Even creating the advisory sidecar may fail after publishing content."""
|
||||
data = b"content survives an unavailable sidecar"
|
||||
if existing:
|
||||
previous = store.put_bytes(data, kind="tool-output")
|
||||
store._paths_for(previous)[1].unlink()
|
||||
original_mkstemp = local_fs_module.tempfile.mkstemp
|
||||
|
||||
def fail_sidecar_allocation(*args, **kwargs):
|
||||
if kwargs.get("prefix") == ".meta-":
|
||||
raise OSError(errno.ENOSPC, "sidecar allocation failed")
|
||||
return original_mkstemp(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(local_fs_module.tempfile, "mkstemp", fail_sidecar_allocation)
|
||||
ref = store.put_bytes(data, kind="tool-output")
|
||||
|
||||
assert store.get_bytes(ref) == data
|
||||
assert not store._paths_for(ref)[1].exists()
|
||||
assert not list(store._root.rglob("*.tmp"))
|
||||
assert "Could not write blob sidecar" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.parametrize("existing", [False, True], ids=["new-put", "retry-put"])
|
||||
def test_put_survives_sidecar_publish_failure(store: LocalFsBlobStore, monkeypatch: pytest.MonkeyPatch, existing: bool):
|
||||
"""A failed sidecar replacement still removes its temporary file."""
|
||||
data = b"content survives a failed sidecar replacement"
|
||||
if existing:
|
||||
previous = store.put_bytes(data, kind="tool-output")
|
||||
store._paths_for(previous)[1].unlink()
|
||||
original_replace = local_fs_module.os.replace
|
||||
|
||||
def fail_sidecar_publish(source, destination):
|
||||
if Path(destination).suffix == ".json":
|
||||
raise OSError(errno.ENOSPC, "sidecar replacement failed")
|
||||
return original_replace(source, destination)
|
||||
|
||||
monkeypatch.setattr(local_fs_module.os, "replace", fail_sidecar_publish)
|
||||
ref = store.put_bytes(data, kind="tool-output")
|
||||
|
||||
assert store.get_bytes(ref) == data
|
||||
assert not store._paths_for(ref)[1].exists()
|
||||
assert not list(store._root.rglob("*.tmp"))
|
||||
|
||||
|
||||
def test_content_allocation_failure_still_fails_put(store: LocalFsBlobStore, monkeypatch: pytest.MonkeyPatch):
|
||||
"""Only metadata failures are advisory; missing content remains an error."""
|
||||
|
||||
def fail_allocation(*args, **kwargs):
|
||||
raise OSError(errno.ENOSPC, "content allocation failed")
|
||||
|
||||
monkeypatch.setattr(local_fs_module.tempfile, "mkstemp", fail_allocation)
|
||||
with pytest.raises(BlobWriteError, match="content allocation failed"):
|
||||
store.put_bytes(b"not published", kind="tool-output")
|
||||
assert not any(path.is_file() for path in store._root.rglob("*"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Factory + config wiring
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user