mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(config): sign the bytes the config cache parsed, not a second read (#5848)
* fix(config): sign the bytes the config cache parsed, not a second read _load_and_cache_app_config parsed config.yaml through AppConfig.from_file and then called get_config_signature, which opens and hashes the file again. A write landing between those two reads left the cache holding the older content under the newer content's signature. get_app_config() compares the on-disk signature against the recorded one, so from then on it saw "unchanged" and the edit only surfaced when yet another edit changed the signature — silently stale until then, with no self-healing. Read the file once through a new shared helper, file_signature.read_config_with_signature, which returns the bytes together with the (mtime, size, sha256) of exactly those bytes, and parse them through AppConfig._from_yaml_text (from_file's body, split out; from_file itself is unchanged). The recorded signature now always describes the parsed content: a write that races the load can only produce a signature mismatch on the next call, i.e. one extra reload. Signing before parsing but re-reading for the parse would not be enough — an edit that lands between the two reads and is then reverted would be served forever — so the tests pin that the parsed bytes are the signed bytes. The stale-metadata reload test now routes the loader's read through the same stale-stat simulation it already applies to the probe, since the loader no longer records the probe's result. * docs(changelog): reference #5848 in the config cache entries --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
82671596f3
commit
38b82057e1
@ -957,6 +957,14 @@ This release closes that milestone with **765 merged pull requests**.
|
||||
|
||||
### Fixed
|
||||
|
||||
- **config:** A `config.yaml` edit that lands while the previous edit is still
|
||||
being loaded is no longer lost until the next edit. `get_app_config()`'s
|
||||
loader parsed the file and then hashed it again to record the cache
|
||||
signature, so a write between those two reads left the cache holding the
|
||||
older content under the newer content's signature — a state the signature
|
||||
comparison can never detect. The loader now reads the file once and signs
|
||||
the bytes it parsed; a write that races the load just triggers one more
|
||||
reload on the next call. ([#5848])
|
||||
- **config:** `request_admission.requests_per_minute` and `max_queue_size` now
|
||||
accept `$VAR` environment references like every other field. Both are strict
|
||||
integers so a bool or float is still rejected, but `$VAR` substitution always
|
||||
@ -4428,4 +4436,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#5776]: https://github.com/bytedance/deer-flow/pull/5776
|
||||
[#5777]: https://github.com/bytedance/deer-flow/pull/5777
|
||||
[#5838]: https://github.com/bytedance/deer-flow/pull/5838
|
||||
[#5848]: https://github.com/bytedance/deer-flow/pull/5848
|
||||
|
||||
|
||||
@ -727,6 +727,11 @@
|
||||
|
||||
### 修复
|
||||
|
||||
- **配置:** 在上一次编辑仍在加载时落盘的 `config.yaml` 编辑,不再要等到下一次编辑才生效。
|
||||
`get_app_config()` 的加载器先解析文件,再重新读取一遍来计算缓存签名,因此夹在两次读取
|
||||
之间的写入会让缓存以较新内容的签名保存较旧的内容,而签名比较永远无法发现这种状态。
|
||||
现在加载器只读取文件一次,并对解析的那份字节计算签名;与加载竞争的写入只会在下一次
|
||||
调用时多触发一次重载。([#5848])
|
||||
- **配置:** `request_admission.requests_per_minute` 与 `max_queue_size` 现在与其他字段一样
|
||||
接受 `$VAR` 环境变量引用。这两个字段是严格整数,布尔值与浮点数仍会被拒绝;但 `$VAR`
|
||||
替换得到的永远是字符串,因此即使 `RPM=60`,`requests_per_minute: $RPM` 也会让整个配置
|
||||
@ -3578,3 +3583,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子
|
||||
[#5776]: https://github.com/bytedance/deer-flow/pull/5776
|
||||
[#5777]: https://github.com/bytedance/deer-flow/pull/5777
|
||||
[#5838]: https://github.com/bytedance/deer-flow/pull/5838
|
||||
[#5848]: https://github.com/bytedance/deer-flow/pull/5848
|
||||
|
||||
@ -39,7 +39,7 @@ their tool-local settings unchanged.
|
||||
|
||||
Top-level `recursion_limit` and `max_recursion_limit` are hot-reloaded per Gateway run. The former supplies the default when a request omits or provides an invalid value; the latter caps both configured and client-provided budgets.
|
||||
|
||||
**Config Caching**: `get_app_config()` caches the parsed config, but automatically reloads it when the resolved config path or file content signature changes. The signature includes file metadata and a content digest, so Gateway and LangGraph reads stay aligned with `config.yaml` edits even on object-store or network mounts where mtime can remain stale.
|
||||
**Config Caching**: `get_app_config()` caches the parsed config, but automatically reloads it when the resolved config path or file content signature changes. The signature includes file metadata and a content digest, so Gateway and LangGraph reads stay aligned with `config.yaml` edits even on object-store or network mounts where mtime can remain stale. The loader reads the file once through `file_signature.read_config_with_signature` and parses those same bytes, so the recorded signature always describes the parsed content: a write that races the load can only cause one extra reload, never a cache that holds one revision under another revision's signature (which the comparison could never detect).
|
||||
|
||||
**Config Hot-Reload Boundary**: Gateway dependencies route through `get_app_config()` on every request, so per-run fields like `models[*].max_tokens`, `summarization.*`, `title.*`, `memory.*`, `subagents.*`, `verification.*`, `tools[*]`, and the agent system prompt pick up `config.yaml` edits on the next message. `AppConfig` is intentionally **not** cached on `app.state` — `lifespan()` keeps a local `startup_config` variable for one-shot bootstrap work and passes it to `langgraph_runtime(app, startup_config)`.
|
||||
|
||||
|
||||
@ -22,6 +22,7 @@ from deerflow.config.dedupe_storage_config import DedupeStorageConfig
|
||||
from deerflow.config.extensions_config import ExtensionsConfig
|
||||
from deerflow.config.file_signature import ConfigSignature as _ConfigSignature
|
||||
from deerflow.config.file_signature import get_config_signature as _get_config_signature
|
||||
from deerflow.config.file_signature import read_config_with_signature as _read_config_with_signature
|
||||
from deerflow.config.guardrails_config import GuardrailsConfig, load_guardrails_config_from_dict
|
||||
from deerflow.config.input_polish_config import InputPolishConfig
|
||||
from deerflow.config.knowledge_base_config import KnowledgeBaseConfig
|
||||
@ -437,7 +438,17 @@ class AppConfig(BaseModel):
|
||||
"""
|
||||
resolved_path = cls.resolve_config_path(config_path)
|
||||
with open(resolved_path, encoding="utf-8") as f:
|
||||
config_data = yaml.safe_load(f) or {}
|
||||
return cls._from_yaml_text(f.read(), resolved_path)
|
||||
|
||||
@classmethod
|
||||
def _from_yaml_text(cls, text: str, resolved_path: Path) -> Self:
|
||||
"""Build the config from already-read YAML *text* of *resolved_path*.
|
||||
|
||||
Split out of :meth:`from_file` so the process-wide cache can parse the
|
||||
exact bytes it signed (see ``_load_and_cache_app_config``) instead of
|
||||
reading the file a second time.
|
||||
"""
|
||||
config_data = yaml.safe_load(text) or {}
|
||||
|
||||
# Check config version before processing
|
||||
cls._check_config_version(config_data, resolved_path)
|
||||
@ -680,14 +691,25 @@ def _get_config_mtime(config_path: Path) -> float | None:
|
||||
|
||||
|
||||
def _load_and_cache_app_config(config_path: str | None = None) -> AppConfig:
|
||||
"""Load config from disk and refresh cache metadata."""
|
||||
"""Load config from disk and refresh cache metadata.
|
||||
|
||||
The file is read exactly once and the recorded signature is computed from
|
||||
those bytes. Parsing the file and then hashing it again would open a window
|
||||
in which a concurrent write leaves the cache holding the older content under
|
||||
the newer content's signature — a state ``get_app_config`` can never detect,
|
||||
because the on-disk signature already matches, so the edit would only be
|
||||
picked up by the *next* edit. Signing the parsed bytes makes any write that
|
||||
races the load show up as a signature mismatch on the next call instead.
|
||||
"""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_signature, _app_config_is_custom
|
||||
|
||||
resolved_path = AppConfig.resolve_config_path(config_path)
|
||||
_app_config = AppConfig.from_file(str(resolved_path))
|
||||
raw, signature = _read_config_with_signature(resolved_path)
|
||||
config = AppConfig._from_yaml_text(raw.decode("utf-8"), resolved_path)
|
||||
_app_config = config
|
||||
_app_config_path = resolved_path
|
||||
_app_config_mtime = _get_config_mtime(resolved_path)
|
||||
_app_config_signature = _get_config_signature(resolved_path)
|
||||
_app_config_mtime = signature[0]
|
||||
_app_config_signature = signature
|
||||
_app_config_is_custom = False
|
||||
return _app_config
|
||||
|
||||
|
||||
@ -53,3 +53,27 @@ def get_config_signature(config_path: Path) -> ConfigSignature | None:
|
||||
return (stat_result.st_mtime, stat_result.st_size, None)
|
||||
|
||||
return (stat_result.st_mtime, stat_result.st_size, digest.hexdigest())
|
||||
|
||||
|
||||
def read_config_with_signature(config_path: Path) -> tuple[bytes, ConfigSignature]:
|
||||
"""Read *config_path* once and return its bytes with the signature of those bytes.
|
||||
|
||||
``get_config_signature`` is a probe: it re-reads the file, so a caller that
|
||||
parses the file and *then* probes it can record one revision's signature
|
||||
against another revision's content when a write lands in between. That
|
||||
leaves the cache serving stale content that the signature comparison can
|
||||
never detect, because the on-disk signature already matches. A cache that
|
||||
parses the bytes returned here and records the signature returned here
|
||||
cannot get into that state: the size and digest describe exactly the bytes
|
||||
that were parsed, and a write that lands at any point around the read only
|
||||
ever produces a signature mismatch on the next comparison, i.e. one more
|
||||
reload.
|
||||
|
||||
The stat is taken before the read, matching ``get_config_signature``, so a
|
||||
stable file yields an identical signature from both helpers. Unlike the
|
||||
probe, a missing or unreadable file raises (``OSError``): the caller wants
|
||||
the content, so there is no "no file" value to return.
|
||||
"""
|
||||
stat_result = config_path.stat()
|
||||
data = config_path.read_bytes()
|
||||
return data, (stat_result.st_mtime, len(data), hashlib.sha256(data).hexdigest())
|
||||
|
||||
@ -520,8 +520,17 @@ def test_get_app_config_reloads_when_content_digest_changes_without_metadata(tmp
|
||||
assert current_signature is not None
|
||||
return (initial_signature[0], initial_signature[1], current_signature[2])
|
||||
|
||||
real_read_config_with_signature = app_config_module._read_config_with_signature
|
||||
|
||||
def stale_metadata_reader(path: Path):
|
||||
# The loader signs the bytes it parses; on a mount with stale
|
||||
# metadata its stat sees the same stale mtime/size as the probe.
|
||||
data, current_signature = real_read_config_with_signature(path)
|
||||
return data, (initial_signature[0], initial_signature[1], current_signature[2])
|
||||
|
||||
monkeypatch.setattr(app_config_module, "_get_config_mtime", lambda _path: initial_mtime)
|
||||
monkeypatch.setattr(app_config_module, "_get_config_signature", stale_metadata_signature)
|
||||
monkeypatch.setattr(app_config_module, "_read_config_with_signature", stale_metadata_reader)
|
||||
|
||||
reloaded = get_app_config()
|
||||
assert reloaded.models[0].name == "model-b"
|
||||
|
||||
@ -13,10 +13,11 @@ can silently drift apart.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from deerflow.config.file_signature import ConfigSignature, get_config_signature
|
||||
from deerflow.config.file_signature import ConfigSignature, get_config_signature, read_config_with_signature
|
||||
|
||||
|
||||
def test_missing_file_returns_none(tmp_path: Path):
|
||||
@ -75,3 +76,80 @@ def test_app_config_and_mcp_cache_share_the_same_implementation():
|
||||
assert cache_module._get_config_signature is get_config_signature
|
||||
assert app_config_module._ConfigSignature is ConfigSignature
|
||||
assert cache_module._ConfigSignature is ConfigSignature
|
||||
|
||||
|
||||
def test_read_config_with_signature_describes_exactly_the_bytes_it_returns(tmp_path: Path):
|
||||
"""The returned signature must be derived from the returned bytes, not re-read from disk."""
|
||||
cfg = tmp_path / "config.yaml"
|
||||
cfg.write_text("log_level: info\n", encoding="utf-8")
|
||||
|
||||
data, signature = read_config_with_signature(cfg)
|
||||
|
||||
assert data == b"log_level: info\n"
|
||||
assert signature == get_config_signature(cfg) # a stable file yields the same signature either way
|
||||
assert signature[1] == len(data)
|
||||
assert signature[2] == hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def test_read_config_with_signature_raises_for_a_missing_file(tmp_path: Path):
|
||||
"""Unlike ``get_config_signature`` (a probe), reading for parsing must fail loudly like ``open()`` does."""
|
||||
import pytest
|
||||
|
||||
with pytest.raises(FileNotFoundError):
|
||||
read_config_with_signature(tmp_path / "missing.yaml")
|
||||
|
||||
|
||||
def test_app_config_cache_loader_reads_through_the_shared_reader():
|
||||
"""``_load_and_cache_app_config`` must parse the same bytes the shared reader signed."""
|
||||
import deerflow.config.app_config as app_config_module
|
||||
import deerflow.config.file_signature as file_signature_module
|
||||
|
||||
assert app_config_module._read_config_with_signature is file_signature_module.read_config_with_signature
|
||||
|
||||
|
||||
def test_read_config_with_signature_signs_the_bytes_even_when_stat_is_stale(tmp_path: Path, monkeypatch):
|
||||
"""Size and digest come from the bytes actually read, not from the stat taken before the read."""
|
||||
cfg = tmp_path / "config.yaml"
|
||||
cfg.write_text("log_level: info\n", encoding="utf-8")
|
||||
original_stat = Path.stat
|
||||
fired = False
|
||||
|
||||
def stat_then_grow(self: Path, *args, **kwargs):
|
||||
nonlocal fired
|
||||
result = original_stat(self, *args, **kwargs)
|
||||
if self == cfg and not fired: # the file grows between the stat and the read
|
||||
fired = True
|
||||
with cfg.open("a", encoding="utf-8") as handle:
|
||||
handle.write("# appended after stat\n")
|
||||
return result
|
||||
|
||||
monkeypatch.setattr(Path, "stat", stat_then_grow)
|
||||
|
||||
data, signature = read_config_with_signature(cfg)
|
||||
|
||||
assert data.endswith(b"# appended after stat\n")
|
||||
assert signature[1] == len(data)
|
||||
assert signature[2] == hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def test_read_config_with_signature_hashes_the_bytes_it_returns_not_a_second_read(tmp_path: Path, monkeypatch):
|
||||
"""The digest must be computed from the returned bytes; a second read could see a newer revision."""
|
||||
cfg = tmp_path / "config.yaml"
|
||||
cfg.write_text("log_level: info\n", encoding="utf-8")
|
||||
original_read_bytes = Path.read_bytes
|
||||
fired = False
|
||||
|
||||
def read_then_rewrite(self: Path):
|
||||
nonlocal fired
|
||||
data = original_read_bytes(self)
|
||||
if self == cfg and not fired: # the file is rewritten right after the first read
|
||||
fired = True
|
||||
cfg.write_text("log_level: debug\n", encoding="utf-8")
|
||||
return data
|
||||
|
||||
monkeypatch.setattr(Path, "read_bytes", read_then_rewrite)
|
||||
|
||||
data, signature = read_config_with_signature(cfg)
|
||||
|
||||
assert data == b"log_level: info\n"
|
||||
assert signature[2] == hashlib.sha256(data).hexdigest()
|
||||
|
||||
@ -13,6 +13,7 @@ runtime ``ContextVar`` override must keep working for per-request injection.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
@ -20,10 +21,12 @@ import pytest
|
||||
from fastapi import Depends, FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
import deerflow.config.app_config as app_config_module
|
||||
from app.gateway import deps as gateway_deps
|
||||
from app.gateway.deps import get_config
|
||||
from deerflow.config.app_config import (
|
||||
AppConfig,
|
||||
get_app_config,
|
||||
pop_current_app_config,
|
||||
push_current_app_config,
|
||||
reset_app_config,
|
||||
@ -98,6 +101,100 @@ def test_get_config_reflects_file_mtime_reload(tmp_path, monkeypatch):
|
||||
assert client.get("/probe").json() == {"log_level": "debug"}
|
||||
|
||||
|
||||
def _install_edit_during_load(monkeypatch, config_file: Path, *, log_level: str) -> None:
|
||||
"""Rewrite ``config_file`` once, from inside the next config load.
|
||||
|
||||
``AppConfig._check_config_version`` runs after the YAML has been parsed and
|
||||
before the loader records its cache metadata, so an edit written there
|
||||
lands in exactly the window between "content parsed" and "signature
|
||||
recorded".
|
||||
"""
|
||||
original = AppConfig._check_config_version.__func__
|
||||
fired = False
|
||||
|
||||
def racy(cls, config_data, resolved_path):
|
||||
nonlocal fired
|
||||
if not fired:
|
||||
fired = True
|
||||
_write_config_yaml(config_file, log_level=log_level)
|
||||
return original(cls, config_data, resolved_path)
|
||||
|
||||
monkeypatch.setattr(AppConfig, "_check_config_version", classmethod(racy))
|
||||
|
||||
|
||||
def test_edit_landing_during_load_is_applied_on_the_next_call(tmp_path, monkeypatch):
|
||||
"""An edit that lands while the previous edit is being loaded must not be lost.
|
||||
|
||||
The loader used to parse the file and then hash the file again to record
|
||||
the cache signature. A write between those two reads left the cache holding
|
||||
the older content under the newer content's signature, so ``get_app_config``
|
||||
saw "unchanged" forever after and the edit only surfaced on the *next* edit.
|
||||
"""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
_write_config_yaml(config_file, log_level="info")
|
||||
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_file))
|
||||
assert get_app_config().log_level == "info"
|
||||
|
||||
_write_config_yaml(config_file, log_level="warning") # edit #1: triggers a reload
|
||||
_install_edit_during_load(monkeypatch, config_file, log_level="debug") # edit #2: lands mid-load
|
||||
|
||||
assert get_app_config().log_level == "warning" # the load that was in flight parsed edit #1
|
||||
assert get_app_config().log_level == "debug" # edit #2 must be visible on the very next call
|
||||
|
||||
|
||||
def test_recorded_signature_is_the_signature_of_the_parsed_content(tmp_path, monkeypatch):
|
||||
"""The cache metadata must describe the bytes that were parsed, not whatever is on disk afterwards."""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
_write_config_yaml(config_file, log_level="info")
|
||||
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_file))
|
||||
assert get_app_config().log_level == "info"
|
||||
|
||||
_write_config_yaml(config_file, log_level="warning")
|
||||
parsed_bytes = config_file.read_bytes()
|
||||
_install_edit_during_load(monkeypatch, config_file, log_level="debug")
|
||||
|
||||
assert get_app_config().log_level == "warning"
|
||||
|
||||
recorded = app_config_module._app_config_signature
|
||||
assert recorded is not None
|
||||
assert recorded[1] == len(parsed_bytes)
|
||||
assert recorded[2] == hashlib.sha256(parsed_bytes).hexdigest()
|
||||
assert recorded != app_config_module._get_config_signature(config_file) # disk already holds edit #2
|
||||
|
||||
|
||||
def test_parsed_content_is_the_signed_content_even_across_an_edit_and_revert(tmp_path, monkeypatch):
|
||||
"""Signing the file before parsing it is not enough: the parsed bytes must *be* the signed bytes.
|
||||
|
||||
A loader that hashes the file, then re-opens it to parse, can parse a
|
||||
revision it never signed. If that revision is then reverted, the disk
|
||||
matches the recorded signature again and the stale parse is served
|
||||
forever. Reading once and parsing those bytes makes the scenario moot.
|
||||
"""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
_write_config_yaml(config_file, log_level="info")
|
||||
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_file))
|
||||
assert get_app_config().log_level == "info"
|
||||
|
||||
_write_config_yaml(config_file, log_level="warning") # edit #1: triggers a reload
|
||||
real_read = app_config_module._read_config_with_signature
|
||||
fired = False
|
||||
|
||||
def read_then_edit(path):
|
||||
nonlocal fired
|
||||
result = real_read(path)
|
||||
if not fired: # edit #2 lands right after the signed read, before parsing
|
||||
fired = True
|
||||
_write_config_yaml(config_file, log_level="debug")
|
||||
return result
|
||||
|
||||
monkeypatch.setattr(app_config_module, "_read_config_with_signature", read_then_edit)
|
||||
|
||||
assert get_app_config().log_level == "warning" # what was signed is what was parsed
|
||||
|
||||
_write_config_yaml(config_file, log_level="warning") # edit #3 reverts edit #2; disk matches the recorded signature again
|
||||
assert get_app_config().log_level == "warning" # correct for the disk; a re-reading loader would serve "debug" here forever
|
||||
|
||||
|
||||
def test_get_config_respects_runtime_context_override(tmp_path, monkeypatch):
|
||||
"""Per-request ``push_current_app_config`` injection must still win."""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user