diff --git a/backend/packages/harness/deerflow/storage/backends/local_fs/local_fs_store.py b/backend/packages/harness/deerflow/storage/backends/local_fs/local_fs_store.py index 04bce9a331..5e88776c85 100644 --- a/backend/packages/harness/deerflow/storage/backends/local_fs/local_fs_store.py +++ b/backend/packages/harness/deerflow/storage/backends/local_fs/local_fs_store.py @@ -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) diff --git a/backend/tests/test_blob_store.py b/backend/tests/test_blob_store.py index c4ea6af00e..eaf22e2a6f 100644 --- a/backend/tests/test_blob_store.py +++ b/backend/tests/test_blob_store.py @@ -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 # ---------------------------------------------------------------------------