mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
* feat(storage): add a content-addressed blob store contract (#4189 item 2) Addresses the review on the previous head: * the retention/dedup claim in the doc now states the deletion rule (liveness from the durable references) instead of promising refcount-free deletion, and the sidecar records the first writer as advisory provenance only; * one validate_blob_kind() owns the kind grammar, so the message a producer sees is the pattern that is enforced; * adds the missing storage/AGENTS.md (registry entry included) that both docstrings already pointed at; * drops the dead _import_backend_module() by routing the folder scan through it; * exports BlobNotConfiguredError from the package; * renames _KIND_MAX_BYTES to _MAX_BLOB_BYTES (it caps one put, not a kind); * delete failures raise the neutral BlobStoreError instead of BlobReadError. Rebased onto current main so the guidance-budget test matches upstream. Signed-off-by: zeng-bohan <zengbh1@gmail.com> * fix(storage): restore the config module and backend packages dropped by the squash The force-push to e72b435 silently lost four items the surviving code imports, breaking `import deerflow.storage` and the test suite at collection: - deerflow/config/blob_storage_config.py (BlobStorageConfig + get/set_blob_storage_config + load_blob_storage_config_from_dict, mirroring memory_config's host-shared schema and hot-reload pattern) - app_config.py wiring: the blob_storage AppConfig field, the import, and the _apply_singleton_configs -> load_blob_storage_config_from_dict call (memory stayed wired; blob did not) - storage/backends/__init__.py (drop-in contract docstring) - storage/backends/local_fs/__init__.py (LocalFsBlobStore re-export + STORE_CLASS for _scan_backends) Verified: pytest tests/test_blob_store.py tests/test_app_config_reload.py tests/test_app_config_name_indexes.py -> 67 passed; ruff format/check clean on all four files. * fix(storage): name the failing digest check; document the 64 MiB per-put cap - get_bytes: split the corruption message into size mismatch (truncation / partial write) vs digest mismatch (same-size bit rot, reporting what the content hashes to vs what the ref names) — the combined message reported 'got 1024 bytes, expected 1024' on same-size corruption and sent the debugger in the wrong direction. - docs/blob-storage.md + storage/AGENTS.md: state the _MAX_BLOB_BYTES cap (64 MiB) and that it is backend-level defense-in-depth rather than part of the BlobStore contract, so the externalized-tool-results follow-up sees the limit before the first production put fails. * test(storage): pin both get_bytes corruption branches separately The pre-split test corrupted the blob with different-size bytes, so only the size-mismatch branch ever ran and neither new message was asserted. Now: different-size corruption asserts 'size mismatch', and a same-size reversal (bit rot) asserts 'digest mismatch' — the branch the split was actually about. * docs(config): document blob_storage in config.example.yaml Add the blob_storage section (enabled/backend/backend_config) mirroring the memory section's shape, matching the Configuration section of backend/docs/blob-storage.md, and bump config_version 46 -> 47 per the example-file convention for schema additions. Default state is unchanged: enabled: false, purely opt-in. * fix(helm): align chart config version with example * fix(storage): refresh blob store on config changes --------- Signed-off-by: zeng-bohan <zengbh1@gmail.com> Co-authored-by: zengbohan1 <310902929+zengbohan1@users.noreply.github.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com> Co-authored-by: zeng-bohan <310902929+zeng-bohan@users.noreply.github.com>
This commit is contained in:
zeng-bohan
GitHub
parent
45cf6e86b5
commit
82ac1612e3
112
backend/docs/blob-storage.md
Normal file
112
backend/docs/blob-storage.md
Normal file
@ -0,0 +1,112 @@
|
||||
# Blob storage (content-addressed, cross-instance)
|
||||
|
||||
Resolves the multi-instance half of [#4189](https://github.com/bytedance/deer-flow/issues/4189) item 2: two producers persist blob-shaped data outside the checkpoint payload and address it with a **server-local filesystem path**, which only resolves on the instance that wrote it.
|
||||
|
||||
| Producer | Write | Reads |
|
||||
|---|---|---|
|
||||
| Viewed images | `view_image_tool` → `ViewedImageData.actual_path` (`deerflow/agents/thread_state.py:52`) | `ViewImageMiddleware._read_image_as_data_url`, gateway artifact routes, IM channels, `present_file_tool` |
|
||||
| Externalized tool results | `ToolOutputBudgetMiddleware._externalize` → virtual path under the thread `outputs` tree | model `read_file` via the thread-data mount |
|
||||
|
||||
On a single gateway both are correct. Behind a load balancer, the instance handling the read is frequently not the instance that wrote the file. The blob store replaces **"where on this machine"** with **"which content"**, so every instance that can reach the backing store resolves the same bytes.
|
||||
|
||||
## Contract
|
||||
|
||||
`deerflow/storage/contract.py`
|
||||
|
||||
- `BlobRef` — `{sha256, size, kind, content_type}`. The digest is the address, so writes are idempotent and dedup is free.
|
||||
- `BlobStore` — plain ABC, tiered like `MemoryStorage`:
|
||||
- **abstract** — `put_bytes(data, *, kind, content_type=None, thread_id=None) -> BlobRef` (`thread_id` is advisory provenance, see the deletion rule below), `get_bytes(ref) -> bytes`
|
||||
- **default** — `exists` (probes via `get_bytes`), `delete` (raises; deleting an absent blob is not an error), `close`
|
||||
- Errors — `BlobStoreError` → `BlobNotConfiguredError` / `BlobWriteError` / `BlobReadError` → `BlobNotFoundError`. A failed `delete` raises the neutral base `BlobStoreError`, not a read error.
|
||||
|
||||
Reads **verify the digest**. A content-addressed store that silently returns wrong bytes is indistinguishable from a corrupt checkpoint, so it fails loudly instead.
|
||||
|
||||
`kind` is validated by `validate_blob_kind` against `^[a-z0-9]([a-z0-9-]*[a-z0-9])?$` (1-64 chars, no leading or trailing hyphen) at the contract level because it is also a path segment in the `local_fs` backend — traversal and separator surprises are ruled out once, not per backend, and the single validator is what a rejected producer sees.
|
||||
|
||||
## Backend
|
||||
|
||||
`deerflow/storage/backends/local_fs/` (default)
|
||||
|
||||
```
|
||||
<root>/<kind>/<sha256[:2]>/<sha256> the bytes
|
||||
<root>/<kind>/<sha256[:2]>/<sha256>.json sidecar: content_type, writer_thread_id, created_at
|
||||
```
|
||||
|
||||
- Writes go to a unique temp file in the destination directory then `os.replace` — atomic within a volume, so a concurrent reader never sees a partial file and two instances racing the same blob converge on identical content.
|
||||
- Single-put size cap: `_MAX_BLOB_BYTES` (64 MiB) in `local_fs_store.py` rejects larger puts with `BlobWriteError`. This is a backend-level defense-in-depth limit, **not** part of the `BlobStore` contract — other backends set their own policy — but the externalized-tool-results follow-up should assume it and stream or split oversized payloads before they reach `put_bytes`.
|
||||
- The sidecar is GC metadata, **not a read dependency and not a reference count**: losing it must not make content unreadable, and `writer_thread_id` is advisory provenance that cannot stand in for liveness (see the deletion rule above).
|
||||
|
||||
This backend already delivers multi-instance resolution when `root` points at a shared volume (NFS / EFS / a `ReadWriteMany` PVC). The S3/MinIO backend is a later, optional extra implementing the same contract — it is deliberately not part of this change, because it needs a dependency decision (`[tool.uv.sources]`, optional extra) that belongs in its own PR.
|
||||
|
||||
## Configuration
|
||||
|
||||
```yaml
|
||||
blob_storage:
|
||||
enabled: true # default false
|
||||
backend: local_fs # folder name under storage/backends/, or a dotted import path
|
||||
backend_config:
|
||||
root: /mnt/shared/deerflow-blobs # default: {runtime_home}/blobs, absolute
|
||||
```
|
||||
|
||||
Fail-fast on an unresolvable backend (`ValueError`), mirroring `MemoryConfig.manager_class`: blobs are persistent state, so silently substituting a different backend would strand previously written content.
|
||||
|
||||
The factory checks the effective backend and backend configuration on each access. After a `config.yaml` edit, new accesses use the new store; disabling storage makes the optional accessor return `None` and the required accessor raise `BlobNotConfiguredError`. A store already handed to an in-flight caller remains usable and is closed by `reset_blob_store()` rather than during the switch. When changing a storage root in a multi-instance deployment, coordinate the rollout and keep previously written blobs available until existing references have been migrated.
|
||||
|
||||
## What this PR deliberately does not do
|
||||
|
||||
**No producer is migrated.** `blob_storage.enabled` defaults to `false`, no existing call site changed, and the diff is purely additive — a deployment that never sets the key behaves exactly as before.
|
||||
|
||||
Migration happens in two follow-up PRs, each independently revertible:
|
||||
|
||||
1. **Viewed images.** `ViewedImageData` gains an optional `blob_ref` (`actual_path` is kept); `view_image_tool` writes the blob when the store is enabled; `ViewImageMiddleware._read_image_as_data_url` resolves blob-first, path-second. The gateway artifact routes and IM channels keep their local-path reads until they can be exercised against a multi-instance deployment.
|
||||
2. **Externalized tool results.** `ToolOutputBudgetMiddleware._externalize` records a ref alongside the virtual path; the sandbox variant (`_externalize_to_sandbox`, issue #3416) stays as-is — sandbox-resident content is a different failure mode one layer down.
|
||||
|
||||
## Interaction with checkpoint retention (#5255)
|
||||
|
||||
Blobs are content-addressed, and `kind` plus the sidecar's `writer_thread_id`
|
||||
make a sweep *addressable* by kind and thread. That is the whole claim this seam
|
||||
supports — it is **not** that retention can stop reasoning about references:
|
||||
|
||||
> **Deletion rule.** A blob may be unlinked only after confirming that no
|
||||
> surviving durable reference names it. `kind` / `writer_thread_id` narrow the
|
||||
> candidate set; on their own they are never sufficient.
|
||||
|
||||
The narrower rule is the honest one, because identical content collapses to one
|
||||
object. Two threads that externalize the same bytes share a single file, and the
|
||||
sidecar can only record whichever writer landed first (`writer_thread_id` is set
|
||||
by the first write; the idempotent path does not rewrite it). Concretely: thread
|
||||
A and thread B both view the same uploaded image — same bytes, same sha256, one
|
||||
file under `viewed-image/`. A sweep for A that trusted the sidecar would delete
|
||||
bytes B still references, and B's own sweep would never see the blob at all,
|
||||
because the sidecar names A.
|
||||
|
||||
Recording *every* writer thread id would not repair that. Appending to one
|
||||
sidecar from two gateway instances is a read-modify-write race with lost
|
||||
updates, so the list could be missing precisely the writer that still needs the
|
||||
content — and a reference count that is wrong under concurrency is worse than
|
||||
none. Hence the contract states the liveness rule instead of a refcount.
|
||||
|
||||
What this means for the two migration PRs:
|
||||
|
||||
- **Viewed images** — an image is shareable by construction (any two threads can
|
||||
upload the same bytes), so the sweep must check liveness against the
|
||||
checkpoint references before unlinking; `kind` and `writer_thread_id` bound
|
||||
the search.
|
||||
- **Externalized tool results** — per-thread by construction, so a thread-scoped
|
||||
sweep is safe while that remains true; `kind` is what makes the "while"
|
||||
checkable.
|
||||
|
||||
This is also the seam [#5188](https://github.com/bytedance/deer-flow/issues/5188) needs: a thread-scoped blob sweep keyed by thread incarnation, rather than by the reusable thread id.
|
||||
|
||||
## Adding a backend
|
||||
|
||||
`packages/harness/deerflow/storage/AGENTS.md` owns the depth — the drop-in
|
||||
folder contract, the portability rule and the durability rules a new backend has
|
||||
to keep. In short:
|
||||
|
||||
1. Copy `backends/local_fs/` to `backends/<name>/`.
|
||||
2. Implement `from_config` + `put_bytes` + `get_bytes`; override `delete` if you can support it.
|
||||
3. Export `STORE_CLASS = <YourStore>` from `backends/<name>/__init__.py`.
|
||||
4. Set `blob_storage.backend: <name>`; backend knobs go under `blob_storage.backend_config`.
|
||||
|
||||
If the backend needs external libs (boto3, minio), declare them in `packages/harness/pyproject.toml` with `[tool.uv.sources]` — otherwise `uv sync` purges them.
|
||||
@ -14,6 +14,7 @@ from deerflow.config.agent_storage_config import AgentStorageConfig
|
||||
from deerflow.config.agents_api_config import AgentsApiConfig, load_agents_api_config_from_dict
|
||||
from deerflow.config.auth_config import AuthAppConfig
|
||||
from deerflow.config.authorization_config import AuthorizationConfig, load_authorization_config_from_dict
|
||||
from deerflow.config.blob_storage_config import BlobStorageConfig, load_blob_storage_config_from_dict
|
||||
from deerflow.config.channel_connections_config import ChannelConnectionsConfig
|
||||
from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
|
||||
from deerflow.config.database_config import DatabaseConfig
|
||||
@ -247,6 +248,7 @@ class AppConfig(BaseModel):
|
||||
summarization: SummarizationConfig = Field(default_factory=SummarizationConfig, description="Conversation summarization configuration")
|
||||
task_continuity: TaskContinuityConfig = Field(default_factory=TaskContinuityConfig, description="Thread-local notes and compacted-source recall")
|
||||
memory: MemoryConfig = Field(default_factory=MemoryConfig, description="Memory subsystem configuration")
|
||||
blob_storage: BlobStorageConfig = Field(default_factory=BlobStorageConfig, description="Content-addressed blob store configuration")
|
||||
knowledge_base: KnowledgeBaseConfig = Field(
|
||||
default_factory=KnowledgeBaseConfig,
|
||||
description="Provider-agnostic knowledge capability and custom-agent scope-selection configuration",
|
||||
@ -484,6 +486,7 @@ class AppConfig(BaseModel):
|
||||
load_title_config_from_dict(config.title.model_dump())
|
||||
load_summarization_config_from_dict(config.summarization.model_dump())
|
||||
load_memory_config_from_dict(config.memory.model_dump())
|
||||
load_blob_storage_config_from_dict(config.blob_storage.model_dump())
|
||||
load_agents_api_config_from_dict(config.agents_api.model_dump())
|
||||
load_subagents_config_from_dict(config.subagents.model_dump())
|
||||
load_tool_search_config_from_dict(config.tool_search.model_dump())
|
||||
|
||||
120
backend/packages/harness/deerflow/config/blob_storage_config.py
Normal file
120
backend/packages/harness/deerflow/config/blob_storage_config.py
Normal file
@ -0,0 +1,120 @@
|
||||
"""Configuration for the blob store (host-shared fields only).
|
||||
|
||||
Backend-private fields live in each backend's own config (parsed from
|
||||
``backend_config``, a dict the factory passes verbatim to the backend). This
|
||||
module holds ONLY the host-shared fields every backend / call site / factory
|
||||
reads: ``enabled`` / ``backend`` / ``backend_config``. Keeping the shared
|
||||
schema slim is what makes backends swappable and portable (a backend's knobs
|
||||
do not leak onto the shared contract). Mirrors ``memory_config``: blobs are
|
||||
persistent state, so the factory fails fast on an unresolvable ``backend``
|
||||
instead of silently substituting a different storage backend.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Host-shared BlobStorageConfig fields (read by every backend / call site / factory).
|
||||
_SHARED_FIELDS = frozenset({"enabled", "backend", "backend_config"})
|
||||
|
||||
|
||||
class BlobStorageConfig(BaseModel):
|
||||
"""Host-shared blob store configuration (backend-agnostic)."""
|
||||
|
||||
enabled: bool = Field(
|
||||
default=False,
|
||||
description=("Whether to enable the content-addressed blob store. Default False: no producer is migrated yet, so a deployment that never sets this key behaves exactly as it does today (see docs/blob-storage.md)."),
|
||||
)
|
||||
backend: str = Field(
|
||||
default="local_fs",
|
||||
description=(
|
||||
"Blob store backend selector. Either a registered backend name "
|
||||
"(matching a `storage/backends/<name>/` folder that exposes "
|
||||
"`STORE_CLASS`, e.g. `local_fs`) or a dotted import path to a "
|
||||
"`BlobStore` subclass. The factory resolves this at "
|
||||
"`get_blob_store()` time and raises `ValueError` on failure "
|
||||
"(fail-fast: blobs are persistent state, so an unresolved backend "
|
||||
"must not be silently substituted with a different one)."
|
||||
),
|
||||
)
|
||||
backend_config: dict[str, Any] = Field(
|
||||
default_factory=dict,
|
||||
description=(
|
||||
"Backend-private config (a dict), passed verbatim to the backend by "
|
||||
"the factory. Each backend self-interprets it (local_fs reads "
|
||||
"`root`; an empty root defaults to `{runtime_home}/blobs`). Values "
|
||||
"live in the host config file (`config.yaml` "
|
||||
"`blob_storage.backend_config`); they do not belong on the shared "
|
||||
"schema."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# Global configuration instance
|
||||
_blob_storage_config: BlobStorageConfig = BlobStorageConfig()
|
||||
|
||||
|
||||
def get_blob_storage_config() -> BlobStorageConfig:
|
||||
"""Get the current blob store configuration.
|
||||
|
||||
``_blob_storage_config`` is only refreshed as a side effect of
|
||||
``get_app_config()`` reloading (via ``_apply_singleton_configs`` ->
|
||||
``load_blob_storage_config_from_dict``). A reader that reaches blob config
|
||||
without going through ``get_app_config()`` first -- e.g. a middleware
|
||||
deciding whether to externalize a tool result -- would otherwise see a
|
||||
stale ``blob_storage.enabled`` after a ``config.yaml`` edit, even though
|
||||
``blob_storage.*`` is documented as hot-reloadable. Trigger the same
|
||||
signature-checked reload here so the singleton follows the config file.
|
||||
|
||||
If ``get_app_config()`` has never been called (``_app_config`` is
|
||||
``None``), there is no stale config to refresh, so we keep the
|
||||
pre-existing behaviour of returning the in-memory singleton. This avoids
|
||||
picking up a config file as a side effect of the first access to
|
||||
``get_blob_storage_config()``, which would break callers that expect
|
||||
module-level defaults (e.g. unit tests).
|
||||
"""
|
||||
# Lazy import: app_config imports this module, so a top-level import cycles.
|
||||
from .app_config import _app_config, get_app_config
|
||||
|
||||
if _app_config is not None:
|
||||
try:
|
||||
get_app_config()
|
||||
except Exception:
|
||||
# If the config file is transiently broken (invalid YAML, schema
|
||||
# violation, missing env var, etc.), keep the last-good singleton
|
||||
# so an in-flight turn completes normally instead of crashing.
|
||||
logger.warning(
|
||||
"Failed to reload app config from get_blob_storage_config(); falling back to cached blob storage config.",
|
||||
exc_info=True,
|
||||
)
|
||||
return _blob_storage_config
|
||||
|
||||
|
||||
def set_blob_storage_config(config: BlobStorageConfig) -> None:
|
||||
"""Set the blob store configuration (tests / programmatic setup)."""
|
||||
global _blob_storage_config
|
||||
_blob_storage_config = config
|
||||
|
||||
|
||||
def load_blob_storage_config_from_dict(config_dict: dict) -> None:
|
||||
"""Load blob store configuration from a dictionary.
|
||||
|
||||
Host-shared fields (``enabled`` / ``backend`` / ``backend_config``) are
|
||||
read directly. Unknown top-level keys (likely typos) are warned and
|
||||
ignored -- there are no legacy pre-abstraction fields to migrate (the
|
||||
blob store is new in this contract).
|
||||
"""
|
||||
global _blob_storage_config
|
||||
config_dict = dict(config_dict or {})
|
||||
unknown = sorted(key for key in config_dict if key not in _SHARED_FIELDS)
|
||||
if unknown:
|
||||
logger.warning(
|
||||
"Unknown blob_storage config key(s) %r at top level (shared fields: %s); ignored.",
|
||||
unknown,
|
||||
sorted(_SHARED_FIELDS),
|
||||
)
|
||||
config_dict = {key: value for key, value in config_dict.items() if key in _SHARED_FIELDS}
|
||||
_blob_storage_config = BlobStorageConfig(**config_dict)
|
||||
79
backend/packages/harness/deerflow/storage/AGENTS.md
Normal file
79
backend/packages/harness/deerflow/storage/AGENTS.md
Normal file
@ -0,0 +1,79 @@
|
||||
### Blob Store (`packages/harness/deerflow/storage/`)
|
||||
|
||||
**What it owns**: a content-addressed store for bytes that must resolve on every
|
||||
instance, not just the one that wrote them (issue #4189, item 2). The address is
|
||||
the SHA-256 of the content, so writes are idempotent, deduplication is free, and
|
||||
a reader can verify what it got back. Contract: `storage/contract.py`
|
||||
(`BlobRef`, `BlobStore`, the `BlobStoreError` family, `validate_blob_kind`).
|
||||
Operator-facing behaviour and the producer migration plan: `docs/blob-storage.md`.
|
||||
Factory: `storage/manager.py`.
|
||||
|
||||
**Disabled by default**: `blob_storage.enabled` is `false`, and no producer has
|
||||
been migrated, so an untouched deployment behaves exactly as before. Producers
|
||||
call `get_blob_store_if_enabled()` and keep their existing local-path code for the
|
||||
`None` case; only a caller that genuinely requires the store uses
|
||||
`get_blob_store()`, which raises `BlobNotConfiguredError` when it is disabled.
|
||||
Both accessors are exported from this package — prefer
|
||||
`from deerflow.storage import ...` over reaching into submodules.
|
||||
The factory compares the effective backend and backend config before reusing its
|
||||
cached store, so hot-reloaded `blob_storage` edits take effect on new accesses.
|
||||
Replaced stores stay open for callers that already hold them and are closed by
|
||||
`reset_blob_store()`; changing a root requires preserving existing blob refs
|
||||
and coordinating deployment instances.
|
||||
|
||||
**Layout** (local_fs, the default)::
|
||||
|
||||
<root>/<kind>/<sha256[:2]>/<sha256> the bytes
|
||||
<root>/<kind>/<sha256[:2]>/<sha256>.json sidecar metadata
|
||||
|
||||
The sidecar holds what the bytes cannot say (`content_type`,
|
||||
`writer_thread_id`, `created_at`). It is **not** a reference count and reads must
|
||||
not depend on it: losing it cannot make content unreadable.
|
||||
|
||||
**Adding a backend** — the memory backends' drop-in contract, folder name ==
|
||||
backend name == `blob_storage.backend` value:
|
||||
|
||||
1. Create `storage/backends/<name>/` with an `__init__.py` that exposes
|
||||
`STORE_CLASS = <YourStore>`; the factory discovers backends by scanning that
|
||||
folder, so no registry edit is needed.
|
||||
2. Subclass `BlobStore` and implement `from_config`, `put_bytes`, `get_bytes`;
|
||||
`exists` / `delete` / `close` have working defaults (`delete` raises so a
|
||||
backend that cannot delete says so). Verify the digest on read — a store that
|
||||
silently returns wrong bytes is indistinguishable from a corrupt checkpoint.
|
||||
3. Keep the portability rule: a backend talks to the host through the method
|
||||
arguments and its `backend_config` dict only. The single `deerflow` import a
|
||||
backend folder needs is the contract line importing `BlobStore`, so a
|
||||
third-party backend can live outside this tree and be selected with a dotted
|
||||
path (`pkg.mod.Class` or `pkg.mod:Class`).
|
||||
4. Reject an unusable config by raising; a store that silently starts on a
|
||||
different root or backend strands previously written content.
|
||||
5. Add the backend's semantics tests to `backend/tests/test_blob_store.py`; the
|
||||
suite runs every backend-agnostic scenario against a temporary root.
|
||||
|
||||
**Durability rules that reviewers have enforced**, worth keeping intact:
|
||||
|
||||
- Validate the `kind` through `validate_blob_kind` instead of re-implementing the
|
||||
grammar; it is also a path segment in `local_fs`, so traversal and separator
|
||||
surprises are ruled out once, at the contract level.
|
||||
- Publish atomically and tolerate a concurrent writer: two gateway instances may
|
||||
put the same content at the same time (see `LocalFsBlobStore._publish`, which
|
||||
handles the Windows `os.replace`/`WinError 5` case by recognising that the
|
||||
destination already holds this content).
|
||||
- Fail fast on an unresolvable backend: `_resolve_store_class` raises rather than
|
||||
substituting a default, because blobs are persistent state.
|
||||
- Reject oversized puts early: `local_fs` enforces `_MAX_BLOB_BYTES` (64 MiB) per
|
||||
put as a backend-level defense-in-depth limit — not part of the `BlobStore`
|
||||
contract, so a different backend sets its own policy, and the
|
||||
externalized-tool-results follow-up must stream or split payloads before
|
||||
`put_bytes`.
|
||||
|
||||
**Garbage collection is not this module's job.** A blob row may be referenced by
|
||||
several threads once identical content dedupes, so a sweep must establish
|
||||
liveness from the durable references (the checkpoint rows that name the blob) and
|
||||
never from the sidecar's `writer_thread_id`. The exact rule, and how it composes
|
||||
with the checkpoint retention contract (#5255), is stated in
|
||||
`docs/blob-storage.md`.
|
||||
|
||||
**Entry points**: `packages/harness/deerflow/storage/contract.py`,
|
||||
`storage/manager.py`, `storage/backends/local_fs/local_fs_store.py`,
|
||||
`config/blob_storage_config.py`; tests in `backend/tests/test_blob_store.py`.
|
||||
34
backend/packages/harness/deerflow/storage/__init__.py
Normal file
34
backend/packages/harness/deerflow/storage/__init__.py
Normal file
@ -0,0 +1,34 @@
|
||||
"""Content-addressed blob store (issue #4189, item 2).
|
||||
|
||||
See ``contract.py`` for the interface and ``AGENTS.md`` for how to add a
|
||||
backend. Nothing in deer-flow writes here until ``blob_storage.enabled`` is
|
||||
True and a producer has been migrated to the store.
|
||||
"""
|
||||
|
||||
from deerflow.storage.contract import (
|
||||
BlobNotConfiguredError,
|
||||
BlobNotFoundError,
|
||||
BlobReadError,
|
||||
BlobRef,
|
||||
BlobStore,
|
||||
BlobStoreError,
|
||||
BlobWriteError,
|
||||
is_valid_blob_kind,
|
||||
validate_blob_kind,
|
||||
)
|
||||
from deerflow.storage.manager import get_blob_store, get_blob_store_if_enabled, reset_blob_store
|
||||
|
||||
__all__ = [
|
||||
"BlobNotFoundError",
|
||||
"BlobNotConfiguredError",
|
||||
"BlobReadError",
|
||||
"BlobRef",
|
||||
"BlobStore",
|
||||
"BlobStoreError",
|
||||
"BlobWriteError",
|
||||
"get_blob_store",
|
||||
"get_blob_store_if_enabled",
|
||||
"is_valid_blob_kind",
|
||||
"reset_blob_store",
|
||||
"validate_blob_kind",
|
||||
]
|
||||
@ -0,0 +1,13 @@
|
||||
"""Pluggable blob store backends.
|
||||
|
||||
Each subpackage is a self-contained backend that exposes
|
||||
``STORE_CLASS`` (a :class:`~deerflow.storage.contract.BlobStore`
|
||||
subclass) in its ``__init__``. The drop-in contract: folder name ==
|
||||
backend name == ``BlobStorageConfig.backend`` value.
|
||||
|
||||
Add a new backend by dropping a new folder here and setting
|
||||
``blob_storage.backend: <name>`` -- no other deer-flow code changes.
|
||||
The depth (layout, publish rules, portability, dotted-path escape hatch)
|
||||
is owned by ``../AGENTS.md``; the retention/GC interaction lives in
|
||||
``docs/blob-storage.md``.
|
||||
"""
|
||||
@ -0,0 +1,13 @@
|
||||
"""local_fs backend -- the default blob store (content-addressed files).
|
||||
|
||||
Holds the store class (:mod:`local_fs_store`): content-addressed files under
|
||||
``<root>/<kind>/<sha256[:2]>/<sha256>`` plus a JSON sidecar carrying what the
|
||||
bytes cannot say, published atomically and verified on read.
|
||||
"""
|
||||
|
||||
from .local_fs_store import LocalFsBlobStore
|
||||
|
||||
#: The :class:`~deerflow.storage.contract.BlobStore` subclass this
|
||||
#: backend exposes. Discovered by the factory's ``_scan_backends`` drop-in
|
||||
#: mechanism under the folder name ``local_fs``.
|
||||
STORE_CLASS = LocalFsBlobStore
|
||||
@ -0,0 +1,258 @@
|
||||
"""Local filesystem blob store -- the default backend.
|
||||
|
||||
Layout (content-addressed, sharded)::
|
||||
|
||||
<root>/<kind>/<sha256[:2]>/<sha256> the bytes
|
||||
<root>/<kind>/<sha256[:2]>/<sha256>.json sidecar metadata
|
||||
|
||||
The address is the SHA-256 of the content, so:
|
||||
|
||||
* writes are idempotent -- putting the same bytes again finds the file already
|
||||
present and does nothing;
|
||||
* deduplication is free -- two producers that externalize the same output share
|
||||
one file;
|
||||
* a reader can verify what it got back against the address, which is what makes
|
||||
"which content" a safe replacement for "where on this machine".
|
||||
|
||||
The sidecar carries what cannot be derived from the bytes alone
|
||||
(``content_type``, ``writer_thread_id``, creation time). Reads do not depend on
|
||||
it: a lost or truncated sidecar must not make blob content unreadable. It is also
|
||||
**not a reference count** -- ``writer_thread_id`` records whichever writer landed
|
||||
first, and identical bytes collapse to one file, so a sweep must confirm that no
|
||||
surviving reference needs the content before unlinking it (the rule in
|
||||
``docs/blob-storage.md``).
|
||||
|
||||
Concurrency: two gateway instances may put the same blob at the same time.
|
||||
Both write to a unique temp file in the destination directory and
|
||||
``os.replace`` it into place, so the last writer wins with identical content
|
||||
and no reader ever observes a partial file. On Windows ``os.replace`` is
|
||||
atomic within a volume, which is why the temp file is created in the
|
||||
destination directory rather than a global temp dir.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from deerflow.storage.contract import (
|
||||
BlobNotFoundError,
|
||||
BlobReadError,
|
||||
BlobRef,
|
||||
BlobStore,
|
||||
BlobStoreError,
|
||||
BlobWriteError,
|
||||
validate_blob_kind,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_META_SUFFIX = ".json"
|
||||
_MAX_BLOB_BYTES = 64 * 1024 * 1024 # defense-in-depth cap on a single put
|
||||
# os.replace retry budget for the concurrent-writer race (see _publish).
|
||||
_REPLACE_ATTEMPTS = 3
|
||||
_REPLACE_BACKOFF_SECONDS = 0.02
|
||||
|
||||
|
||||
class LocalFsBlobStore(BlobStore):
|
||||
"""Content-addressed store on the local filesystem.
|
||||
|
||||
This is the default and the migration bridge: when ``root`` points at a
|
||||
shared volume (NFS, EFS, a PVC in ``ReadWriteMany``) the store already
|
||||
gives multi-instance resolution with no object storage dependency. The S3
|
||||
backend is a later, optional extra -- it implements the same contract.
|
||||
"""
|
||||
|
||||
def __init__(self, root: str | os.PathLike[str]) -> None:
|
||||
self._root = Path(root).expanduser()
|
||||
try:
|
||||
self._root.mkdir(parents=True, exist_ok=True)
|
||||
except OSError as exc:
|
||||
raise BlobWriteError(f"Cannot create blob store root {self._root}: {exc}") from exc
|
||||
if not os.access(self._root, os.W_OK):
|
||||
raise BlobWriteError(f"Blob store root is not writable: {self._root}")
|
||||
|
||||
@classmethod
|
||||
def from_config(cls, backend_config: dict[str, Any]) -> LocalFsBlobStore:
|
||||
root = backend_config.get("root")
|
||||
if not root:
|
||||
raise ValueError("local_fs blob store requires backend_config.root (the factory injects a default)")
|
||||
return cls(root=str(root))
|
||||
|
||||
# -- layout -----------------------------------------------------------
|
||||
|
||||
def _paths_for(self, ref: BlobRef) -> tuple[Path, Path]:
|
||||
shard_dir = self._root / ref.kind / ref.sha256[:2]
|
||||
return shard_dir / ref.sha256, shard_dir / f"{ref.sha256}{_META_SUFFIX}"
|
||||
|
||||
# -- contract ---------------------------------------------------------
|
||||
|
||||
def put_bytes(
|
||||
self,
|
||||
data: bytes,
|
||||
*,
|
||||
kind: str,
|
||||
content_type: str | None = None,
|
||||
thread_id: str | None = None,
|
||||
) -> BlobRef:
|
||||
validate_blob_kind(kind)
|
||||
if len(data) > _MAX_BLOB_BYTES:
|
||||
raise BlobWriteError(f"Refusing to store {len(data)} bytes (cap is {_MAX_BLOB_BYTES})")
|
||||
|
||||
digest = hashlib.sha256(data).hexdigest()
|
||||
ref = BlobRef(sha256=digest, size=len(data), kind=kind, content_type=content_type)
|
||||
data_path, meta_path = self._paths_for(ref)
|
||||
shard_dir = data_path.parent
|
||||
|
||||
if data_path.is_file():
|
||||
# Idempotent re-put: content already present. Ensure the sidecar
|
||||
# exists (an earlier put may have crashed between the two writes).
|
||||
if not meta_path.is_file():
|
||||
self._write_meta(ref, meta_path, thread_id)
|
||||
return ref
|
||||
|
||||
try:
|
||||
shard_dir.mkdir(parents=True, exist_ok=True)
|
||||
fd, tmp_name = tempfile.mkstemp(prefix=".put-", suffix=".tmp", dir=shard_dir)
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as handle:
|
||||
handle.write(data)
|
||||
handle.flush()
|
||||
os.fsync(handle.fileno())
|
||||
self._publish(tmp_name, data_path, data)
|
||||
except BaseException:
|
||||
# Best effort: a stranded temp file must not accumulate.
|
||||
try:
|
||||
os.unlink(tmp_name)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
self._write_meta(ref, meta_path, thread_id)
|
||||
except BlobWriteError:
|
||||
raise
|
||||
except OSError as exc:
|
||||
raise BlobWriteError(f"Failed to persist blob {ref.sha256[:12]}: {exc}") from exc
|
||||
return ref
|
||||
|
||||
def get_bytes(self, ref: BlobRef) -> bytes:
|
||||
data_path, _ = self._paths_for(ref)
|
||||
try:
|
||||
data = data_path.read_bytes()
|
||||
except FileNotFoundError as exc:
|
||||
raise BlobNotFoundError(f"Blob {ref.sha256[:12]} not found under kind {ref.kind!r}") from exc
|
||||
except OSError as exc:
|
||||
raise BlobReadError(f"Failed to read blob {ref.sha256[:12]}: {exc}") from exc
|
||||
|
||||
# Verify against the address. A content-addressed store that returns
|
||||
# the wrong bytes silently is indistinguishable from a corrupt
|
||||
# checkpoint, so fail loudly instead — naming WHICH check failed:
|
||||
# a size delta points at truncation/partial writes, a digest mismatch
|
||||
# at same-size corruption (bit rot), and conflating the two sends the
|
||||
# debugger in the wrong direction ("got 1024 bytes, expected 1024").
|
||||
if len(data) != ref.size:
|
||||
raise BlobReadError(f"Blob {ref.sha256[:12]} size mismatch on read (got {len(data)} bytes, expected {ref.size}); the store may be truncated or corrupted")
|
||||
digest = hashlib.sha256(data).hexdigest()
|
||||
if digest != ref.sha256:
|
||||
raise BlobReadError(f"Blob {ref.sha256[:12]} content digest mismatch on read (content hashes to {digest[:12]}, ref names {ref.sha256[:12]}); the store may be corrupted")
|
||||
return data
|
||||
|
||||
def exists(self, ref: BlobRef) -> bool:
|
||||
data_path, _ = self._paths_for(ref)
|
||||
return data_path.is_file()
|
||||
|
||||
def delete(self, ref: BlobRef) -> None:
|
||||
data_path, meta_path = self._paths_for(ref)
|
||||
removed = False
|
||||
for path in (data_path, meta_path):
|
||||
try:
|
||||
path.unlink()
|
||||
removed = True
|
||||
except FileNotFoundError:
|
||||
continue
|
||||
except OSError as exc:
|
||||
# The neutral base, not BlobReadError: this error family is
|
||||
# public API from day one, and a caller matching on a read
|
||||
# failure must not be handed a failed unlink it cannot read as
|
||||
# one.
|
||||
raise BlobStoreError(f"Failed to delete blob {ref.sha256[:12]}: {exc}") from exc
|
||||
if not removed:
|
||||
logger.debug("delete on absent blob %s (kind=%s): already gone", ref.sha256[:12], ref.kind)
|
||||
|
||||
def close(self) -> None: # noqa: B027 - nothing to release
|
||||
"""No handles to close; the store is stateless beyond the root path."""
|
||||
|
||||
# -- internals --------------------------------------------------------
|
||||
|
||||
def _publish(self, tmp_path: str, data_path: Path, data: bytes) -> None:
|
||||
"""Move the temp file into place, tolerating a concurrent writer.
|
||||
|
||||
``os.replace`` raises ``PermissionError`` (WinError 5) on Windows when
|
||||
another handle still holds the destination open -- which is exactly
|
||||
what happens when two gateway instances put the same blob while a
|
||||
reader is already serving it. If the destination already holds this
|
||||
exact content, another writer won the race and the put is complete;
|
||||
otherwise retry briefly, then surface the error.
|
||||
"""
|
||||
for attempt in range(_REPLACE_ATTEMPTS):
|
||||
try:
|
||||
os.replace(tmp_path, data_path)
|
||||
return
|
||||
except PermissionError:
|
||||
if self._content_matches(data_path, data):
|
||||
self._discard(tmp_path)
|
||||
return
|
||||
if attempt + 1 == _REPLACE_ATTEMPTS:
|
||||
raise
|
||||
time.sleep(_REPLACE_BACKOFF_SECONDS * (attempt + 1))
|
||||
|
||||
@staticmethod
|
||||
def _content_matches(path: Path, data: bytes) -> bool:
|
||||
try:
|
||||
return path.read_bytes() == data
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _discard(path: str) -> None:
|
||||
try:
|
||||
os.unlink(path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def _write_meta(self, ref: BlobRef, meta_path: Path, thread_id: str | None) -> None:
|
||||
payload = {
|
||||
"sha256": ref.sha256,
|
||||
"size": ref.size,
|
||||
"kind": ref.kind,
|
||||
"content_type": ref.content_type,
|
||||
# Advisory provenance only: the first writer of this content. It is
|
||||
# deliberately not a "threads that reference this" list -- identical
|
||||
# bytes dedupe to one file, and appending to a sidecar from two
|
||||
# gateway instances is a lost-update race, so such a list could not
|
||||
# be trusted as a reference count anyway.
|
||||
"writer_thread_id": thread_id,
|
||||
"created_at": time.time(),
|
||||
}
|
||||
fd, tmp_name = tempfile.mkstemp(prefix=".meta-", suffix=".tmp", dir=meta_path.parent)
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
json.dump(payload, handle)
|
||||
os.replace(tmp_name, meta_path)
|
||||
except OSError:
|
||||
# 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
|
||||
logger.warning("Could not write blob sidecar %s", meta_path, exc_info=True)
|
||||
|
||||
|
||||
__all__ = ["LocalFsBlobStore"]
|
||||
198
backend/packages/harness/deerflow/storage/contract.py
Normal file
198
backend/packages/harness/deerflow/storage/contract.py
Normal file
@ -0,0 +1,198 @@
|
||||
"""Content-addressed blob store contract.
|
||||
|
||||
A :class:`BlobRef` addresses **content**, not a location. That is the whole
|
||||
point of the abstraction: on a multi-gateway deployment a server-local
|
||||
filesystem path is only meaningful on the instance that wrote the file,
|
||||
whereas a content digest is meaningful everywhere -- every instance that can
|
||||
reach the backing store resolves the same bytes.
|
||||
|
||||
Producers (issue #4189, item 2):
|
||||
|
||||
* ``ViewedImageData.actual_path`` -- written by ``view_image_tool``, read by
|
||||
``ViewImageMiddleware``, the gateway artifact routes and the IM channels.
|
||||
* Oversized tool results externalized by ``ToolOutputBudgetMiddleware``.
|
||||
|
||||
Both keep their existing local path when the store is disabled, so the seam is
|
||||
purely additive.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Any, ClassVar
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||||
|
||||
# Blob kinds are also filesystem path segments in the local_fs backend, so the
|
||||
# grammar is deliberately narrow: lowercase, digits and hyphens, no leading or
|
||||
# trailing hyphen, 1-64 chars. That rules out traversal (".."), separators and
|
||||
# case-sensitivity surprises on NTFS/APFS at the contract level instead of
|
||||
# leaving every backend to re-check it.
|
||||
_KIND_PATTERN = re.compile(r"^(?=.{1,64}$)[a-z0-9]([a-z0-9-]*[a-z0-9])?$")
|
||||
# The grammar as a caller sees it. One text, one place: this message is the only
|
||||
# thing a producer gets when a kind is rejected, and a second copy of the
|
||||
# pattern (the local_fs backend had one) drifts from the enforced one.
|
||||
_KIND_GRAMMAR = "^[a-z0-9]([a-z0-9-]*[a-z0-9])?$ (1-64 chars, no leading or trailing hyphen)"
|
||||
|
||||
|
||||
def is_valid_blob_kind(kind: str) -> bool:
|
||||
"""Return whether *kind* is a safe blob-kind identifier."""
|
||||
return isinstance(kind, str) and bool(_KIND_PATTERN.match(kind))
|
||||
|
||||
|
||||
def validate_blob_kind(kind: str) -> str:
|
||||
"""Return *kind* unchanged, or raise ``ValueError`` quoting the real grammar.
|
||||
|
||||
Backends must not re-implement the pattern: a store that rejects a kind the
|
||||
contract accepts -- or accepts one it rejects -- makes content addressable
|
||||
on one backend and not another.
|
||||
"""
|
||||
if not is_valid_blob_kind(kind):
|
||||
raise ValueError(f"kind must match {_KIND_GRAMMAR} (it is used as a path segment)")
|
||||
return kind
|
||||
|
||||
|
||||
class BlobStoreError(RuntimeError):
|
||||
"""Backend-neutral base error exposed at the blob store boundary."""
|
||||
|
||||
|
||||
class BlobNotConfiguredError(BlobStoreError):
|
||||
"""A caller required a blob store but ``blob_storage.enabled`` is False."""
|
||||
|
||||
|
||||
class BlobWriteError(BlobStoreError):
|
||||
"""A write failed after retries; the content was not persisted."""
|
||||
|
||||
|
||||
class BlobReadError(BlobStoreError):
|
||||
"""A read failed for a reason other than the blob being absent."""
|
||||
|
||||
|
||||
class BlobNotFoundError(BlobReadError):
|
||||
"""The referenced content is not present in the backing store."""
|
||||
|
||||
|
||||
class BlobRef(BaseModel):
|
||||
"""A content-addressed reference to blob bytes.
|
||||
|
||||
``sha256`` is the address: writes are idempotent (putting the same bytes
|
||||
twice yields the same ref) and deduplication is free. ``size`` is carried
|
||||
so a reader can sanity-check what it got back without trusting the
|
||||
backend, and ``kind`` scopes garbage collection -- retention sweeps can
|
||||
reason about a kind ("externalized tool outputs for thread X") without
|
||||
parsing paths.
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
sha256: str = Field(description="Lowercase hex SHA-256 of the blob content (64 chars).")
|
||||
size: int = Field(ge=0, description="Content length in bytes.")
|
||||
kind: str = Field(description="Content class, e.g. 'viewed-image' or 'tool-output'.")
|
||||
content_type: str | None = Field(default=None, description="MIME type when the producer knows it.")
|
||||
|
||||
@field_validator("sha256")
|
||||
@classmethod
|
||||
def _sha256_is_hex64(cls, value: str) -> str:
|
||||
value = (value or "").strip().lower()
|
||||
if len(value) != 64 or not re.fullmatch(r"[0-9a-f]{64}", value):
|
||||
raise ValueError("sha256 must be 64 lowercase hex characters")
|
||||
return value
|
||||
|
||||
@field_validator("kind")
|
||||
@classmethod
|
||||
def _kind_is_safe(cls, value: str) -> str:
|
||||
return validate_blob_kind(value)
|
||||
|
||||
def matches(self, data: bytes) -> bool:
|
||||
"""Return whether *data* really is this blob's content."""
|
||||
import hashlib
|
||||
|
||||
return len(data) == self.size and hashlib.sha256(data).hexdigest() == self.sha256
|
||||
|
||||
|
||||
class BlobStore(ABC):
|
||||
"""Backend-neutral blob store contract.
|
||||
|
||||
A plain ABC (like ``MemoryStorage``, not the pydantic ``MemoryManager``):
|
||||
the contract carries no config fields, so there is nothing for pydantic to
|
||||
validate and no ``PrivateAttr`` ceremony. Backends are constructed by the
|
||||
factory through :meth:`from_config`.
|
||||
|
||||
Portability rule (mirrors the memory backends' golden rule): a backend
|
||||
talks to the host through exactly two channels -- the method arguments and
|
||||
the ``backend_config`` dict. The only ``deerflow`` import a backend folder
|
||||
needs is the contract line importing :class:`BlobStore`.
|
||||
"""
|
||||
|
||||
#: Sentinel attribute each backend package's ``__init__`` exposes so the
|
||||
#: folder-scan factory can find it (mirrors memory's ``MANAGER_CLASS``).
|
||||
STORE_CLASS_ATTR: ClassVar[str] = "STORE_CLASS"
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def from_config(cls, backend_config: dict[str, Any]) -> BlobStore:
|
||||
"""Build a store from backend-private config.
|
||||
|
||||
``backend_config`` is the dict from ``blob_storage.backend_config``,
|
||||
with ``root`` injected by the factory when the host did not set one.
|
||||
Must raise on an unusable config -- a blob store that silently starts
|
||||
on the wrong root is worse than one that refuses to start.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def put_bytes(
|
||||
self,
|
||||
data: bytes,
|
||||
*,
|
||||
kind: str,
|
||||
content_type: str | None = None,
|
||||
thread_id: str | None = None,
|
||||
) -> BlobRef:
|
||||
"""Persist *data* and return its :class:`BlobRef`.
|
||||
|
||||
Idempotent: putting identical bytes twice returns the same ref and must
|
||||
not duplicate storage.
|
||||
|
||||
``thread_id`` is **advisory provenance, never a reference count.**
|
||||
Content addressing means identical bytes collapse to one object, so two
|
||||
threads that externalize the same content share it, and a backend can
|
||||
only record whichever writer reached it first -- on a multi-instance
|
||||
deployment it cannot even do that reliably, since two gateway instances
|
||||
appending to one sidecar is a read-modify-write race that loses entries.
|
||||
Garbage collection must therefore establish liveness from the durable
|
||||
references (the checkpoint rows that name the blob), not from this
|
||||
field; see ``docs/blob-storage.md`` for the exact rule.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def get_bytes(self, ref: BlobRef) -> bytes:
|
||||
"""Return the content addressed by *ref*.
|
||||
|
||||
Raises :class:`BlobNotFoundError` when the content is absent and
|
||||
:class:`BlobReadError` when it is present but unreadable. Backends
|
||||
SHOULD verify the digest on read: a content-addressed store that
|
||||
silently returns wrong bytes is indistinguishable from a corrupt
|
||||
checkpoint.
|
||||
"""
|
||||
|
||||
def exists(self, ref: BlobRef) -> bool:
|
||||
"""Return whether *ref* is present. Default probes via :meth:`get_bytes`."""
|
||||
try:
|
||||
self.get_bytes(ref)
|
||||
except BlobNotFoundError:
|
||||
return False
|
||||
return True
|
||||
|
||||
def delete(self, ref: BlobRef) -> None:
|
||||
"""Remove the content addressed by *ref*.
|
||||
|
||||
Default raises so a backend that cannot delete says so instead of
|
||||
leaving a caller to assume a sweep succeeded. Deleting an absent blob
|
||||
is not an error (idempotent), matching how the retention contract
|
||||
treats orphans.
|
||||
"""
|
||||
raise NotImplementedError(f"{type(self).__name__} does not support delete")
|
||||
|
||||
def close(self) -> None: # noqa: B027 - intentional no-op default
|
||||
"""Release backend resources. Default: nothing to release."""
|
||||
223
backend/packages/harness/deerflow/storage/manager.py
Normal file
223
backend/packages/harness/deerflow/storage/manager.py
Normal file
@ -0,0 +1,223 @@
|
||||
"""Blob store factory: folder-scan backend resolution + process singleton.
|
||||
|
||||
Mirrors ``deerflow/agents/memory/manager.py`` (the pattern maintainers pointed
|
||||
at for issue #4189 item 2): backends live in ``storage/backends/<name>/``, each
|
||||
package's ``__init__`` exposes ``STORE_CLASS``, and ``blob_storage.backend``
|
||||
names one of them -- or a dotted import path for a backend that lives elsewhere.
|
||||
|
||||
Two accessors, so call sites cannot drift:
|
||||
|
||||
* :func:`get_blob_store_if_enabled` -- the one producers should use. Returns
|
||||
``None`` when ``blob_storage.enabled`` is False, which is the default, so an
|
||||
unmigrated call site is a no-op rather than an accident.
|
||||
* :func:`get_blob_store` -- for callers that require a store; raises
|
||||
:class:`BlobNotConfiguredError` when the store is disabled.
|
||||
|
||||
Both fail fast on an unresolvable backend. A blob store that silently starts on
|
||||
a different backend than the operator configured would strand previously
|
||||
written content, which is worse than refusing to start.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import logging
|
||||
import threading
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
from types import ModuleType
|
||||
from typing import Any
|
||||
|
||||
from deerflow.config.blob_storage_config import BlobStorageConfig, get_blob_storage_config
|
||||
from deerflow.storage.contract import BlobNotConfiguredError, BlobStore, BlobStoreError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Backend packages live in <this dir>/backends/<name>/.
|
||||
_BACKENDS_DIR = Path(__file__).parent / "backends"
|
||||
# Sentinel attribute each backend's __init__ exposes (a BlobStore subclass).
|
||||
_STORE_CLASS_ATTR = "STORE_CLASS"
|
||||
|
||||
# Singleton instance + the effective settings it was built with. Keep replaced
|
||||
# stores alive until reset so callers already holding one can finish using it.
|
||||
_blob_store: BlobStore | None = None
|
||||
_blob_store_settings: tuple[str, dict[str, Any]] | None = None
|
||||
_retired_blob_stores: list[BlobStore] = []
|
||||
_backends_cache: dict[str, type[BlobStore]] | None = None
|
||||
_store_lock = threading.Lock()
|
||||
|
||||
|
||||
def _import_backend_module(name: str) -> ModuleType:
|
||||
"""Import ``backends/<name>`` and return its package module."""
|
||||
package_dir = _BACKENDS_DIR / name
|
||||
if not package_dir.is_dir() or not (package_dir / "__init__.py").is_file():
|
||||
raise ValueError(f"Unknown blob store backend {name!r}: no storage/backends/{name}/ package")
|
||||
return importlib.import_module(f"{__package__}.backends.{name}")
|
||||
|
||||
|
||||
def _resolve_store_class(backend: str) -> type[BlobStore]:
|
||||
"""Resolve ``blob_storage.backend`` to a :class:`BlobStore` subclass.
|
||||
|
||||
Two forms are accepted, matching the memory backends' contract:
|
||||
|
||||
* a registered backend name -- a folder under ``storage/backends/`` whose
|
||||
``__init__`` exposes ``STORE_CLASS``;
|
||||
* a dotted import path -- ``pkg.mod.Class`` for a backend that lives
|
||||
outside this package (an extension or a deployment-local backend).
|
||||
"""
|
||||
global _backends_cache
|
||||
|
||||
if "." not in backend:
|
||||
if _backends_cache is None:
|
||||
_backends_cache = _scan_backends()
|
||||
cls = _backends_cache.get(backend)
|
||||
if cls is None:
|
||||
available = ", ".join(sorted(_backends_cache)) or "(none)"
|
||||
raise ValueError(f"Unknown blob store backend {backend!r}. Registered backends: {available}. Alternatively pass a dotted import path to a BlobStore subclass.")
|
||||
return cls
|
||||
|
||||
module_path, _, class_name = backend.partition(":") if ":" in backend else backend.rpartition(".")
|
||||
if not module_path or not class_name:
|
||||
raise ValueError(f"Invalid blob storage backend {backend!r}: needs 'pkg.mod.Class' or 'pkg.mod:Class'")
|
||||
try:
|
||||
module = importlib.import_module(module_path)
|
||||
candidate = getattr(module, class_name)
|
||||
except (ImportError, AttributeError) as exc:
|
||||
raise ValueError(f"Cannot import blob storage backend {backend!r}: {exc}") from exc
|
||||
if not (isinstance(candidate, type) and issubclass(candidate, BlobStore)):
|
||||
raise ValueError(f"Blob storage backend {backend!r} is not a BlobStore subclass")
|
||||
return candidate
|
||||
|
||||
|
||||
def _scan_backends() -> dict[str, type[BlobStore]]:
|
||||
"""Return ``{name: BlobStore subclass}`` for every backend folder.
|
||||
|
||||
Imports through :func:`_import_backend_module` so the "is this a backend
|
||||
package?" check lives in exactly one place -- a folder that survives the
|
||||
scan is a folder the factory can resolve by name.
|
||||
"""
|
||||
registry: dict[str, type[BlobStore]] = {}
|
||||
if not _BACKENDS_DIR.is_dir():
|
||||
return registry
|
||||
for entry in sorted(_BACKENDS_DIR.iterdir()):
|
||||
if not entry.is_dir() or entry.name.startswith("_"):
|
||||
continue
|
||||
try:
|
||||
module = _import_backend_module(entry.name)
|
||||
except ValueError as exc:
|
||||
# Not a backend package (no __init__.py); nothing to register.
|
||||
logger.debug("Skipping %s: %s", entry.name, exc)
|
||||
continue
|
||||
except Exception:
|
||||
# A broken optional backend must not take the whole store down:
|
||||
# log it and leave it out of the registry so the factory can still
|
||||
# resolve the default.
|
||||
logger.exception("Skipping blob store backend %r: import failed", entry.name)
|
||||
continue
|
||||
cls = getattr(module, _STORE_CLASS_ATTR, None)
|
||||
if isinstance(cls, type) and issubclass(cls, BlobStore):
|
||||
registry[entry.name] = cls
|
||||
else:
|
||||
logger.warning(
|
||||
"Blob store backend folder %r exposes no %s; skipping",
|
||||
entry.name,
|
||||
_STORE_CLASS_ATTR,
|
||||
)
|
||||
return registry
|
||||
|
||||
|
||||
def _default_backend_config() -> dict[str, Any]:
|
||||
"""Zero-config UX: default the backend root to deer-flow's state dir.
|
||||
|
||||
Absolute and CWD-independent, so every gateway instance in a deployment
|
||||
that has not configured a shared root still lands in the same logical
|
||||
place on its own node -- which is exactly the single-host behaviour we
|
||||
must not change.
|
||||
"""
|
||||
from deerflow.config.runtime_paths import runtime_home
|
||||
|
||||
return {"root": str((Path(runtime_home()) / "blobs").resolve())}
|
||||
|
||||
|
||||
def _resolve_backend_config(cfg: BlobStorageConfig) -> dict[str, Any]:
|
||||
backend_config = dict(cfg.backend_config or {})
|
||||
root = backend_config.get("root")
|
||||
if not root:
|
||||
backend_config.update(_default_backend_config())
|
||||
elif not Path(str(root)).is_absolute():
|
||||
# A relative root is resolved against runtime_home() (base_dir-relative,
|
||||
# CWD-independent) in host code, so backends stay free of any
|
||||
# runtime_home dependency -- same reason the memory factory does it.
|
||||
from deerflow.config.runtime_paths import runtime_home
|
||||
|
||||
backend_config["root"] = str((Path(runtime_home()) / str(root)).resolve())
|
||||
return backend_config
|
||||
|
||||
|
||||
def get_blob_store_if_enabled() -> BlobStore | None:
|
||||
"""Return the singleton store, or ``None`` when the store is disabled.
|
||||
|
||||
This is the accessor producers should call: ``None`` is an explicit,
|
||||
checkable signal that the deployment did not opt in, and the producer then
|
||||
keeps its existing server-local path.
|
||||
"""
|
||||
return _get_blob_store_if_enabled()
|
||||
|
||||
|
||||
def get_blob_store() -> BlobStore:
|
||||
"""Return the singleton :class:`BlobStore` for the active config.
|
||||
|
||||
Raises :class:`BlobNotConfiguredError` when ``blob_storage.enabled`` is
|
||||
False, and ``ValueError`` when the configured backend cannot be resolved.
|
||||
"""
|
||||
store = _get_blob_store_if_enabled()
|
||||
if store is None:
|
||||
raise BlobNotConfiguredError("blob_storage.enabled is False; no blob store is available. Use get_blob_store_if_enabled() at optional call sites.")
|
||||
return store
|
||||
|
||||
|
||||
def _get_blob_store_if_enabled() -> BlobStore | None:
|
||||
"""Select a store against one config snapshot under the singleton lock."""
|
||||
global _blob_store, _blob_store_settings
|
||||
|
||||
with _store_lock:
|
||||
cfg = get_blob_storage_config()
|
||||
if not cfg.enabled:
|
||||
return None
|
||||
|
||||
backend_config = _resolve_backend_config(cfg)
|
||||
settings = (cfg.backend, deepcopy(backend_config))
|
||||
if _blob_store is not None and _blob_store_settings == settings:
|
||||
return _blob_store
|
||||
|
||||
cls = _resolve_store_class(cfg.backend)
|
||||
new_store = cls.from_config(backend_config)
|
||||
if _blob_store is not None:
|
||||
_retired_blob_stores.append(_blob_store)
|
||||
_blob_store = new_store
|
||||
_blob_store_settings = settings
|
||||
logger.info("Blob store resolved: %s (backend=%r)", type(_blob_store).__name__, cfg.backend)
|
||||
return _blob_store
|
||||
|
||||
|
||||
def reset_blob_store() -> None:
|
||||
"""Close current and replaced stores; clear the singleton and registry."""
|
||||
global _blob_store, _blob_store_settings, _backends_cache
|
||||
with _store_lock:
|
||||
for store in [*_retired_blob_stores, *([_blob_store] if _blob_store is not None else [])]:
|
||||
try:
|
||||
store.close()
|
||||
except Exception:
|
||||
logger.warning("Blob store close() raised during reset", exc_info=True)
|
||||
_retired_blob_stores.clear()
|
||||
_blob_store = None
|
||||
_blob_store_settings = None
|
||||
_backends_cache = None
|
||||
|
||||
|
||||
__all__ = [
|
||||
"BlobStoreError",
|
||||
"get_blob_store",
|
||||
"get_blob_store_if_enabled",
|
||||
"reset_blob_store",
|
||||
]
|
||||
@ -32,6 +32,7 @@ EXPECTED_GUIDANCE_PATHS = {
|
||||
"backend/packages/harness/deerflow/persistence/user/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/reflection/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/skills/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/storage/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/subagents/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/tools/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/tracing/AGENTS.md",
|
||||
|
||||
@ -23,6 +23,7 @@ from deerflow.config.title_config import get_title_config, load_title_config_fro
|
||||
from deerflow.config.tool_search_config import get_tool_search_config, load_tool_search_config_from_dict
|
||||
from deerflow.runtime.checkpointer import get_checkpointer, reset_checkpointer
|
||||
from deerflow.runtime.store import get_store, reset_store
|
||||
from deerflow.storage import BlobNotConfiguredError, get_blob_store, get_blob_store_if_enabled, reset_blob_store
|
||||
|
||||
|
||||
def _reset_config_singletons() -> None:
|
||||
@ -804,6 +805,40 @@ def test_get_memory_config_self_syncs_without_prior_get_app_config(tmp_path, mon
|
||||
_reset_config_singletons()
|
||||
|
||||
|
||||
def test_blob_store_follows_config_file_reload(tmp_path, monkeypatch):
|
||||
"""The storage accessor must see config edits without a prior app-config read."""
|
||||
config_path = tmp_path / "config.yaml"
|
||||
extensions_path = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(extensions_path)
|
||||
first_root = tmp_path / "first"
|
||||
second_root = tmp_path / "second"
|
||||
_write_config_with_sections(config_path, {"blob_storage": {"enabled": True, "backend_config": {"root": str(first_root)}}})
|
||||
|
||||
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path))
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
|
||||
reset_app_config()
|
||||
reset_blob_store()
|
||||
|
||||
try:
|
||||
get_app_config()
|
||||
first_store = get_blob_store()
|
||||
|
||||
_write_config_with_sections(config_path, {"blob_storage": {"enabled": True, "backend_config": {"root": str(second_root)}}})
|
||||
second_store = get_blob_store()
|
||||
assert second_store is not first_store
|
||||
ref = second_store.put_bytes(b"after reload", kind="tool-output")
|
||||
assert (second_root / ref.kind / ref.sha256[:2] / ref.sha256).is_file()
|
||||
assert not (first_root / ref.kind / ref.sha256[:2] / ref.sha256).exists()
|
||||
|
||||
_write_config_with_sections(config_path, {"blob_storage": {"enabled": False}})
|
||||
assert get_blob_store_if_enabled() is None
|
||||
with pytest.raises(BlobNotConfiguredError):
|
||||
get_blob_store()
|
||||
finally:
|
||||
reset_blob_store()
|
||||
_reset_config_singletons()
|
||||
|
||||
|
||||
def test_get_memory_config_falls_back_on_broken_config(tmp_path, monkeypatch):
|
||||
"""get_memory_config() does not crash on transiently broken config.yaml.
|
||||
|
||||
|
||||
400
backend/tests/test_blob_store.py
Normal file
400
backend/tests/test_blob_store.py
Normal file
@ -0,0 +1,400 @@
|
||||
"""Tests for the content-addressed blob store contract and the local_fs backend.
|
||||
|
||||
Scope: the seam only. No producer is migrated yet, so these tests never touch
|
||||
``viewed_images`` or ``ToolOutputBudgetMiddleware`` -- that is the follow-up
|
||||
wiring PR for issue #4189 item 2.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from deerflow.config import blob_storage_config as bsc
|
||||
from deerflow.storage import (
|
||||
BlobNotConfiguredError,
|
||||
BlobNotFoundError,
|
||||
BlobReadError,
|
||||
BlobRef,
|
||||
BlobStore,
|
||||
BlobStoreError,
|
||||
BlobWriteError,
|
||||
get_blob_store,
|
||||
get_blob_store_if_enabled,
|
||||
is_valid_blob_kind,
|
||||
reset_blob_store,
|
||||
validate_blob_kind,
|
||||
)
|
||||
from deerflow.storage import contract as contract_module
|
||||
from deerflow.storage.backends.local_fs import LocalFsBlobStore
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolated_config(monkeypatch):
|
||||
"""Give every test a pristine config + a fresh singleton."""
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig())
|
||||
reset_blob_store()
|
||||
yield
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig())
|
||||
reset_blob_store()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store(tmp_path: Path) -> LocalFsBlobStore:
|
||||
return LocalFsBlobStore.from_config({"root": str(tmp_path / "blobs")})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# BlobRef
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_blob_ref_validates_digest_shape():
|
||||
with pytest.raises(ValidationError):
|
||||
BlobRef(sha256="not-a-digest", size=1, kind="tool-output")
|
||||
with pytest.raises(ValidationError):
|
||||
BlobRef(sha256="z" * 64, size=1, kind="tool-output") # right length, not hex
|
||||
with pytest.raises(ValidationError):
|
||||
BlobRef(sha256="a" * 63, size=1, kind="tool-output") # too short
|
||||
ref = BlobRef(sha256=hashlib.sha256(b"x").hexdigest(), size=1, kind="tool-output")
|
||||
assert ref.sha256 == hashlib.sha256(b"x").hexdigest()
|
||||
|
||||
|
||||
def test_blob_ref_kind_is_path_safe():
|
||||
for bad in ("../etc", "a/b", "Aupper", "", "-lead", "trailing-", "has space", "x" * 65):
|
||||
with pytest.raises(ValidationError):
|
||||
BlobRef(sha256=hashlib.sha256(b"x").hexdigest(), size=1, kind=bad)
|
||||
for good in ("viewed-image", "tool-output", "a", "0", "a" * 64):
|
||||
assert is_valid_blob_kind(good)
|
||||
|
||||
|
||||
def test_blob_ref_matches_detects_corruption():
|
||||
data = b"payload"
|
||||
ref = BlobRef(sha256=hashlib.sha256(data).hexdigest(), size=len(data), kind="tool-output")
|
||||
assert ref.matches(data)
|
||||
assert not ref.matches(b"payload!")
|
||||
assert not ref.matches(b"")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# local_fs backend semantics
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_put_then_get_roundtrip(store: LocalFsBlobStore):
|
||||
data = "外部化内容 🚀".encode()
|
||||
ref = store.put_bytes(data, kind="tool-output", content_type="text/plain", thread_id="t-1")
|
||||
assert ref.size == len(data)
|
||||
assert ref.sha256 == hashlib.sha256(data).hexdigest()
|
||||
assert ref.content_type == "text/plain"
|
||||
assert store.get_bytes(ref) == data
|
||||
|
||||
|
||||
def test_put_is_idempotent_and_deduplicates(store: LocalFsBlobStore, tmp_path: Path):
|
||||
data = b"same bytes"
|
||||
first = store.put_bytes(data, kind="tool-output")
|
||||
before = sorted(p.name for p in (tmp_path / "blobs").rglob("*") if p.is_file())
|
||||
second = store.put_bytes(data, kind="tool-output", thread_id="t-9")
|
||||
after = sorted(p.name for p in (tmp_path / "blobs").rglob("*") if p.is_file())
|
||||
|
||||
assert first == second
|
||||
assert before == after, "a re-put must not create new files"
|
||||
|
||||
|
||||
def test_dedup_records_only_the_first_writer_as_advisory_provenance(store: LocalFsBlobStore):
|
||||
"""Identical content is *shared*, so the sidecar cannot be a reference count.
|
||||
|
||||
The GC contract in docs/blob-storage.md says a blob may only be unlinked
|
||||
after liveness is confirmed from the durable references; this test pins the
|
||||
sidecar semantics that make that rule necessary, so nobody later "fixes"
|
||||
dedup by trusting writer_thread_id.
|
||||
"""
|
||||
data = b"the same uploaded image"
|
||||
first = store.put_bytes(data, kind="viewed-image", thread_id="thread-a")
|
||||
second = store.put_bytes(data, kind="viewed-image", thread_id="thread-b")
|
||||
|
||||
assert first == second, "one address for one content"
|
||||
data_path = next((store._root / "viewed-image").rglob(first.sha256))
|
||||
sidecar = next((store._root / "viewed-image").rglob(f"{first.sha256}.json"))
|
||||
assert data_path.is_file()
|
||||
payload = json.loads(sidecar.read_text(encoding="utf-8"))
|
||||
assert payload["writer_thread_id"] == "thread-a"
|
||||
assert "thread-b" not in json.dumps(payload), "dedup keeps the first writer only — never a refcount"
|
||||
|
||||
|
||||
def test_kind_rejection_quotes_the_enforced_grammar():
|
||||
"""One validator, one message: the pattern a producer is shown must be the
|
||||
pattern that is enforced (it was not, before this was centralised)."""
|
||||
assert validate_blob_kind("tool-output") == "tool-output"
|
||||
for bad in ("trailing-", "a--", "-lead", "Aupper", "a/b", "", "x" * 65):
|
||||
with pytest.raises(ValueError) as caught:
|
||||
validate_blob_kind(bad)
|
||||
message = str(caught.value)
|
||||
assert "^[a-z0-9]([a-z0-9-]*[a-z0-9])?$" in message, message
|
||||
assert "1-64 chars" in message, message
|
||||
assert "no leading or trailing hyphen" in message, message
|
||||
assert validate_blob_kind is contract_module.validate_blob_kind
|
||||
|
||||
|
||||
def test_same_content_different_kind_lands_separately(store: LocalFsBlobStore):
|
||||
ref_a = store.put_bytes(b"shared", kind="tool-output")
|
||||
ref_b = store.put_bytes(b"shared", kind="viewed-image")
|
||||
assert ref_a.sha256 == ref_b.sha256
|
||||
assert ref_a != ref_b
|
||||
assert store.exists(ref_a) and store.exists(ref_b)
|
||||
|
||||
|
||||
def test_get_missing_blob_raises_not_found(store: LocalFsBlobStore):
|
||||
ref = BlobRef(sha256=hashlib.sha256(b"absent").hexdigest(), size=6, kind="tool-output")
|
||||
with pytest.raises(BlobNotFoundError):
|
||||
store.get_bytes(ref)
|
||||
assert store.exists(ref) is False
|
||||
|
||||
|
||||
def test_get_detects_size_mismatch(store: LocalFsBlobStore):
|
||||
data = b"intact"
|
||||
ref = store.put_bytes(data, kind="tool-output")
|
||||
# Corrupt the stored bytes behind the store's back with a DIFFERENT length:
|
||||
# this is the truncation / partial-write failure mode.
|
||||
data_path = next((store._root / "tool-output").rglob(ref.sha256))
|
||||
data_path.write_bytes(b"tampered")
|
||||
with pytest.raises(BlobReadError, match="size mismatch"):
|
||||
store.get_bytes(ref)
|
||||
|
||||
|
||||
def test_get_detects_same_size_digest_mismatch(store: LocalFsBlobStore):
|
||||
"""Same-size corruption (bit rot) must hit the digest branch, not the size one.
|
||||
|
||||
``data[::-1]`` keeps the length and flips every byte, so the size check
|
||||
passes and the digest comparison is the only guard that can catch it.
|
||||
"""
|
||||
data = b"intact"
|
||||
ref = store.put_bytes(data, kind="tool-output")
|
||||
data_path = next((store._root / "tool-output").rglob(ref.sha256))
|
||||
data_path.write_bytes(data[::-1])
|
||||
with pytest.raises(BlobReadError, match="digest mismatch"):
|
||||
store.get_bytes(ref)
|
||||
|
||||
|
||||
def test_delete_is_idempotent(store: LocalFsBlobStore):
|
||||
ref = store.put_bytes(b"doomed", kind="tool-output")
|
||||
store.delete(ref)
|
||||
assert store.exists(ref) is False
|
||||
store.delete(ref) # absent blob: not an error
|
||||
with pytest.raises(BlobNotFoundError):
|
||||
store.get_bytes(ref)
|
||||
|
||||
|
||||
def test_delete_failure_raises_the_neutral_base_error(store: LocalFsBlobStore, monkeypatch: pytest.MonkeyPatch):
|
||||
"""A failed unlink is not a read failure.
|
||||
|
||||
The exception family is public API from the first release, so callers that
|
||||
match on BlobReadError must not receive delete failures they cannot
|
||||
interpret.
|
||||
"""
|
||||
ref = store.put_bytes(b"cannot unlink", kind="tool-output")
|
||||
|
||||
def _boom(self, *args, **kwargs):
|
||||
raise OSError("device busy")
|
||||
|
||||
monkeypatch.setattr(Path, "unlink", _boom)
|
||||
with pytest.raises(BlobStoreError) as caught:
|
||||
store.delete(ref)
|
||||
assert not isinstance(caught.value, BlobReadError)
|
||||
assert "Failed to delete blob" in str(caught.value)
|
||||
|
||||
|
||||
def test_delete_unsupported_on_contract_default(tmp_path: Path):
|
||||
class Minimal(BlobStore):
|
||||
def from_config(cls, backend_config):
|
||||
raise NotImplementedError
|
||||
|
||||
def put_bytes(self, data, *, kind, content_type=None, thread_id=None):
|
||||
raise NotImplementedError
|
||||
|
||||
def get_bytes(self, ref):
|
||||
raise BlobNotFoundError("nope")
|
||||
|
||||
assert Minimal().exists(BlobRef(sha256="0" * 64, size=0, kind="tool-output")) is False
|
||||
with pytest.raises(NotImplementedError):
|
||||
Minimal().delete(BlobRef(sha256="0" * 64, size=0, kind="tool-output"))
|
||||
|
||||
|
||||
def test_invalid_kind_is_rejected_before_touching_disk(store: LocalFsBlobStore, tmp_path: Path):
|
||||
with pytest.raises(ValueError):
|
||||
store.put_bytes(b"x", kind="../escape")
|
||||
assert not any(tmp_path.rglob("escape"))
|
||||
|
||||
|
||||
def test_oversized_put_is_refused(store: LocalFsBlobStore, monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr("deerflow.storage.backends.local_fs.local_fs_store._MAX_BLOB_BYTES", 4)
|
||||
with pytest.raises(BlobWriteError):
|
||||
store.put_bytes(b"12345", kind="tool-output")
|
||||
|
||||
|
||||
def test_concurrent_puts_of_same_content(tmp_path: Path):
|
||||
"""Two instances racing the same blob must not corrupt or duplicate it."""
|
||||
store = LocalFsBlobStore.from_config({"root": str(tmp_path / "blobs")})
|
||||
data = b"raced content" * 100
|
||||
results: list[BlobRef] = []
|
||||
errors: list[BaseException] = []
|
||||
barrier = threading.Barrier(8)
|
||||
|
||||
def worker() -> None:
|
||||
try:
|
||||
barrier.wait(timeout=5)
|
||||
results.append(store.put_bytes(data, kind="tool-output"))
|
||||
except BaseException as exc: # pragma: no cover - surfaced below
|
||||
errors.append(exc)
|
||||
|
||||
threads = [threading.Thread(target=worker) for _ in range(8)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join(timeout=30)
|
||||
|
||||
assert not errors, errors
|
||||
assert len({r.sha256 for r in results}) == 1
|
||||
assert store.get_bytes(results[0]) == data
|
||||
assert len(list((tmp_path / "blobs" / "tool-output").rglob(f"{results[0].sha256}"))) == 1
|
||||
|
||||
|
||||
def test_put_survives_a_lost_sidecar(store: LocalFsBlobStore):
|
||||
"""The sidecar is GC metadata, not a read dependency."""
|
||||
data = b"payload-with-sidecar"
|
||||
ref = store.put_bytes(data, kind="tool-output")
|
||||
meta = next((store._root / "tool-output").rglob(f"{ref.sha256}.json"))
|
||||
meta.unlink()
|
||||
assert store.get_bytes(ref) == data
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Factory + config wiring
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_package_exports_the_whole_error_family():
|
||||
"""Callers use `from deerflow.storage import ...`, so the error they must
|
||||
catch has to be reachable there — and be the same class the contract
|
||||
raises, not a re-exported copy."""
|
||||
assert BlobNotConfiguredError is contract_module.BlobNotConfiguredError
|
||||
assert BlobStoreError is contract_module.BlobStoreError
|
||||
for name in ("BlobNotConfiguredError", "BlobStoreError", "BlobNotFoundError", "validate_blob_kind"):
|
||||
assert name in __import__("deerflow.storage", fromlist=["__all__"]).__all__
|
||||
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig())
|
||||
reset_blob_store()
|
||||
with pytest.raises(BlobStoreError):
|
||||
get_blob_store()
|
||||
|
||||
|
||||
def test_disabled_by_default():
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig())
|
||||
reset_blob_store()
|
||||
assert get_blob_store_if_enabled() is None
|
||||
with pytest.raises(BlobNotConfiguredError):
|
||||
get_blob_store()
|
||||
|
||||
|
||||
def test_enabled_factory_resolves_folder_backend(tmp_path: Path):
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend="local_fs", backend_config={"root": str(tmp_path / "b")}))
|
||||
reset_blob_store()
|
||||
store = get_blob_store_if_enabled()
|
||||
assert isinstance(store, LocalFsBlobStore)
|
||||
# Singleton: a second call returns the same instance.
|
||||
assert get_blob_store() is store
|
||||
|
||||
|
||||
def test_factory_replaces_store_when_root_changes(tmp_path: Path):
|
||||
first_root = tmp_path / "first"
|
||||
second_root = tmp_path / "second"
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(first_root)}))
|
||||
|
||||
first_store = get_blob_store()
|
||||
first_store.put_bytes(b"before", kind="tool-output")
|
||||
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(second_root)}))
|
||||
second_store = get_blob_store_if_enabled()
|
||||
assert second_store is not first_store
|
||||
assert get_blob_store() is second_store
|
||||
|
||||
after = second_store.put_bytes(b"after", kind="tool-output")
|
||||
assert (second_root / after.kind / after.sha256[:2] / after.sha256).is_file()
|
||||
assert not (first_root / after.kind / after.sha256[:2] / after.sha256).exists()
|
||||
|
||||
|
||||
def test_factory_rejects_cached_store_after_disable(tmp_path: Path):
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(tmp_path)}))
|
||||
get_blob_store()
|
||||
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=False))
|
||||
assert get_blob_store_if_enabled() is None
|
||||
with pytest.raises(BlobNotConfiguredError):
|
||||
get_blob_store()
|
||||
|
||||
|
||||
def test_factory_does_not_keep_cached_store_for_invalid_backend(tmp_path: Path):
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(tmp_path)}))
|
||||
get_blob_store()
|
||||
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend="does_not_exist"))
|
||||
with pytest.raises(ValueError, match="Unknown blob store backend"):
|
||||
get_blob_store()
|
||||
|
||||
|
||||
def test_replaced_store_stays_open_until_reset(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||||
import deerflow.storage.manager as mgr
|
||||
|
||||
closed: list[Path] = []
|
||||
|
||||
class TrackingStore(LocalFsBlobStore):
|
||||
def close(self) -> None:
|
||||
closed.append(self._root)
|
||||
|
||||
monkeypatch.setattr(mgr, "_resolve_store_class", lambda backend: TrackingStore)
|
||||
first_root = tmp_path / "first"
|
||||
second_root = tmp_path / "second"
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(first_root)}))
|
||||
first_store = get_blob_store()
|
||||
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(second_root)}))
|
||||
get_blob_store()
|
||||
assert closed == []
|
||||
first_store.put_bytes(b"in flight", kind="tool-output")
|
||||
|
||||
reset_blob_store()
|
||||
assert set(closed) == {first_root, second_root}
|
||||
|
||||
|
||||
def test_unknown_backend_fails_fast(tmp_path: Path):
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend="does_not_exist", backend_config={"root": str(tmp_path)}))
|
||||
reset_blob_store()
|
||||
with pytest.raises(ValueError, match="Unknown blob store backend"):
|
||||
get_blob_store()
|
||||
|
||||
|
||||
def test_dotted_import_path_backend(tmp_path: Path):
|
||||
bsc.set_blob_storage_config(
|
||||
bsc.BlobStorageConfig(
|
||||
enabled=True,
|
||||
backend="deerflow.storage.backends.local_fs:LocalFsBlobStore",
|
||||
backend_config={"root": str(tmp_path / "d")},
|
||||
)
|
||||
)
|
||||
reset_blob_store()
|
||||
assert isinstance(get_blob_store(), LocalFsBlobStore)
|
||||
|
||||
|
||||
def test_relative_root_is_resolved_against_runtime_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||||
import deerflow.storage.manager as mgr
|
||||
|
||||
monkeypatch.setattr(mgr, "_default_backend_config", lambda: {"root": str(tmp_path / "state")})
|
||||
bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend="local_fs", backend_config={"root": "relative"}))
|
||||
reset_blob_store()
|
||||
store = get_blob_store()
|
||||
assert Path(str(store._root)).is_absolute()
|
||||
@ -20,7 +20,7 @@
|
||||
# ============================================================================
|
||||
# Bump this number when the config schema changes.
|
||||
# Run `make config-upgrade` to merge new fields into your local config.yaml.
|
||||
config_version: 46
|
||||
config_version: 47
|
||||
|
||||
# Optional per-model request pacing (inside a models[] entry):
|
||||
# request_admission:
|
||||
@ -2262,6 +2262,33 @@ memory:
|
||||
# patterns_dir: "" # empty = bundled defaults
|
||||
# prompts_dir: "" # empty = bundled defaults
|
||||
|
||||
# ============================================================================
|
||||
# Blob Storage Configuration
|
||||
# ============================================================================
|
||||
# Content-addressed blob store: content is addressed by its sha256, so writes
|
||||
# are idempotent, identical bytes dedupe to one object, and reads verify the
|
||||
# digest. Mirrors the memory mechanism: a `backend` name matches a
|
||||
# storage/backends/<name>/ folder that exposes STORE_CLASS (e.g. local_fs), or
|
||||
# takes a dotted import path to a BlobStore subclass. The factory resolves the
|
||||
# backend at get_blob_store() time and fails fast on an unresolvable value --
|
||||
# blobs are persistent state, so a broken backend name must not silently swap
|
||||
# the storage location.
|
||||
#
|
||||
# Off by default and purely opt-in: no producer is migrated yet, so a config
|
||||
# that never sets this key behaves exactly as before. Contract, the two named
|
||||
# producers, and the migration order: backend/docs/blob-storage.md.
|
||||
blob_storage:
|
||||
enabled: false # master switch (inert until the producer follow-ups land)
|
||||
backend: local_fs # folder name under storage/backends/, or a dotted import path
|
||||
# Backend-private config (dict), passed verbatim to the backend; each backend
|
||||
# self-interprets it. local_fs reads `root` -- empty defaults to
|
||||
# {runtime_home}/blobs; point it at a shared volume (NFS / EFS / a
|
||||
# ReadWriteMany PVC) for multi-instance deployments:
|
||||
#
|
||||
# backend_config:
|
||||
# root: /mnt/shared/deerflow-blobs
|
||||
backend_config: {}
|
||||
|
||||
# ============================================================================
|
||||
# Custom Agent Management API
|
||||
# ============================================================================
|
||||
|
||||
@ -135,7 +135,7 @@ they resolve from the `secrets` map):
|
||||
|
||||
```yaml
|
||||
config: |
|
||||
config_version: 46
|
||||
config_version: 47
|
||||
models:
|
||||
- name: gpt-4
|
||||
use: langchain_openai:ChatOpenAI
|
||||
|
||||
@ -249,7 +249,7 @@ ingress:
|
||||
# -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
|
||||
# inline literal secret values here. The default enables provisioner sandbox.
|
||||
config: |
|
||||
config_version: 46
|
||||
config_version: 47
|
||||
log_level: info
|
||||
recursion_limit: 100
|
||||
max_recursion_limit: 1000
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user