mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(config): pin explicit app config path across get_app_config (#7092)
* fix(config): pin explicit app config path across get_app_config Fixes #7086. reload_app_config(config_path) now records an _app_config_source pin (mirroring _extensions_config_source) so get_app_config and DeerFlowClient(config_path=...) keep following the explicit file instead of ambient default resolution. * fix(config): synchronize app config cache and absolute pins Serialize singleton reads, reloads, resets and injections so a concurrent call cannot publish a config under another file's pin. Keep explicit paths absolute across working-directory changes while preserving ambient discovery and failed-reload behavior. Add deterministic concurrency and relative-path regressions. Validation: 33,875 offline tests and 262 blocking-I/O tests passed; frontend check, backend formatting and agent guidance passed. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
DDT
GitHub
parent
8984093552
commit
afe244e7f9
@ -3293,6 +3293,11 @@ editing the saved agent configuration to refresh the selection.
|
||||
|
||||
DeerFlow can be used as an embedded Python library without running the full HTTP services. The `DeerFlowClient` provides direct in-process access to all agent and Gateway capabilities, returning the same response schemas as the HTTP Gateway API. The HTTP Gateway also exposes `DELETE /api/threads/{thread_id}` to remove DeerFlow-managed local thread data after the LangGraph thread itself has been deleted:
|
||||
|
||||
An embedded `DeerFlowClient(config_path="custom/config.yaml")` keeps the process
|
||||
config singleton on that file, even after a working-directory change. Later edits
|
||||
to the file are still detected. Call `reload_app_config()` without a path to resume
|
||||
ambient config discovery, or `reset_app_config()` to clear the cached selection.
|
||||
|
||||
For database-backed run events, deleting a run preserves its thread's sequence watermark. Thread deletion removes that watermark once no events remain, allowing a recreated thread to restart at sequence 1. Owner-scoped deletion preserves the watermark when another owner's events remain.
|
||||
|
||||
Single-process JSONL event storage also retains the thread sequence watermark
|
||||
|
||||
@ -61,6 +61,14 @@ YAML loading and both middleware tool-call paths in `test_tool_output_config_lim
|
||||
|
||||
**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). The cached `extensions` snapshot follows the process extensions singleton: `get_app_config()` also reloads when `get_extensions_config()` returns a new instance (see **Extensions Config Caching** below), and the loader takes that instance before reading `config.yaml`, so it can only be older than the YAML it merges with, never newer.
|
||||
|
||||
AppConfig singleton reads, reloads, resets and injections share `_app_config_lock`;
|
||||
publish the parsed revision, signature, extensions snapshot and explicit source pin
|
||||
within that critical section. An explicit reload pins an absolute path, so later
|
||||
working-directory changes cannot redirect it. Failed explicit loads preserve the
|
||||
prior pin; reset, injection and argument-less reload clear it. Ambient reads never
|
||||
pin a path. Runtime-scoped overrides bypass the singleton lock;
|
||||
`peek_loaded_app_config()` also returns the existing snapshot without locking.
|
||||
|
||||
**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)`.
|
||||
|
||||
Infrastructure fields are **restart-required**. The authoritative list lives in `packages/harness/deerflow/config/reload_boundary.py::STARTUP_ONLY_FIELDS` and is mirrored by the standardised `"startup-only:"` prefix on the corresponding `Field(description=...)` in `AppConfig` or an explicitly registered nested config model, so IDE hover on those fields surfaces the reason inline (no need to context-switch into this table). Currently registered: `plugins`, `database`, `checkpointer`, `run_events`, `agent_storage`, `stream_bridge`, `sandbox`, `skills.container_path`, `log_level`, `logging`, `channels`, `channels.leader_lease` (validated by `channel_leader_lease_config.py` from the non-schema `channels` mapping; the multi-process gate refuses `enabled: false` while a non-GitHub channel is enabled), `channel_connections`, `scheduler`, `mcp_tasks`, `subagent_runtime`, `subagent_batches`, `run_ownership`, `dedupe_storage`, `auth.local.throttle_storage` (the login throttle store is installed once by `langgraph_runtime`; `max_login_attempts` / `lockout_seconds` stay live-read). Adding a new restart-required field requires updating the registry; drift is pinned by `tests/test_reload_boundary.py`. `scheduler.recursion_limit` is the exception inside that section: it is read from `get_app_config()` at each scheduled dispatch, so a YAML edit applies to the next run without restarting the poller.
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
from collections.abc import Mapping
|
||||
from contextvars import ContextVar
|
||||
from pathlib import Path
|
||||
@ -726,11 +727,20 @@ class AppConfig(BaseModel):
|
||||
# Compatibility singleton layer for code paths that have not yet been
|
||||
# migrated to explicit ``AppConfig`` threading. New composition roots should
|
||||
# prefer constructing ``AppConfig`` once and passing it down directly.
|
||||
# Serialize cache revisions and their source pin across readers and writers.
|
||||
_app_config_lock = threading.RLock()
|
||||
_app_config: AppConfig | None = None
|
||||
_app_config_path: Path | None = None
|
||||
_app_config_mtime: float | None = None
|
||||
_app_config_signature: _ConfigSignature | None = None
|
||||
_app_config_is_custom = False
|
||||
# The absolute explicit path the cached revision was loaded from, when
|
||||
# ``reload_app_config(config_path=...)`` chose a file that default resolution
|
||||
# would not. ``get_app_config()`` follows that file until ``reset_app_config()``,
|
||||
# ``set_app_config()``, or an argument-less reload; probing default resolution
|
||||
# instead would either flip the cache back to a sibling default or raise when
|
||||
# no ambient config exists (discarding the explicit load).
|
||||
_app_config_source: str | None = None
|
||||
# The ``get_extensions_config()`` instance the cached AppConfig's ``extensions``
|
||||
# snapshot was built from. That singleton is replaced only when the shared
|
||||
# ``extensions_config.json`` changes (or a writer reloads it), so identity is
|
||||
@ -749,7 +759,7 @@ def _get_config_mtime(config_path: Path) -> float | None:
|
||||
|
||||
|
||||
def _load_and_cache_app_config(config_path: str | None = None, *, extensions_source: ExtensionsConfig | None = None) -> AppConfig:
|
||||
"""Load config from disk and refresh cache metadata.
|
||||
"""Load config from disk and refresh cache metadata while the caller holds the cache lock.
|
||||
|
||||
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
|
||||
@ -771,7 +781,7 @@ def _load_and_cache_app_config(config_path: str | None = None, *, extensions_sou
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_signature, _app_config_is_custom
|
||||
global _app_config_extensions_source
|
||||
|
||||
resolved_path = AppConfig.resolve_config_path(config_path)
|
||||
resolved_path = AppConfig.resolve_config_path(config_path).absolute()
|
||||
if extensions_source is None:
|
||||
extensions_source = get_extensions_config()
|
||||
raw, signature = _read_config_with_signature(resolved_path)
|
||||
@ -792,8 +802,10 @@ def get_app_config() -> AppConfig:
|
||||
underlying config file path or content signature changes, or when the
|
||||
process extensions config (`get_extensions_config()`) was reloaded because
|
||||
the shared ``extensions_config.json`` changed, so the ``extensions``
|
||||
snapshot follows that file too. Use `reload_app_config()` to force a
|
||||
reload, or `reset_app_config()` to clear the cache.
|
||||
snapshot follows that file too. When the cache was loaded via an explicit
|
||||
``reload_app_config(config_path)``, that path is preferred over ambient
|
||||
resolution. Use `reload_app_config()` to force a reload, or
|
||||
`reset_app_config()` to clear the cache.
|
||||
"""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_signature
|
||||
|
||||
@ -801,30 +813,32 @@ def get_app_config() -> AppConfig:
|
||||
if runtime_override is not None:
|
||||
return runtime_override
|
||||
|
||||
if _app_config is not None and _app_config_is_custom:
|
||||
return _app_config
|
||||
with _app_config_lock:
|
||||
if _app_config is not None and _app_config_is_custom:
|
||||
return _app_config
|
||||
|
||||
resolved_path = AppConfig.resolve_config_path()
|
||||
current_mtime = _get_config_mtime(resolved_path)
|
||||
current_signature = _get_config_signature(resolved_path)
|
||||
current_extensions_source = get_extensions_config()
|
||||
resolved_path = AppConfig.resolve_config_path(_app_config_source).absolute()
|
||||
current_mtime = _get_config_mtime(resolved_path)
|
||||
current_signature = _get_config_signature(resolved_path)
|
||||
current_extensions_source = get_extensions_config()
|
||||
|
||||
should_reload = _app_config is None or _app_config_path != resolved_path or _app_config_signature != current_signature or current_extensions_source is not _app_config_extensions_source
|
||||
if should_reload:
|
||||
if _app_config_path == resolved_path and _app_config_mtime is not None and current_mtime is not None and _app_config_mtime != current_mtime:
|
||||
logger.info(
|
||||
"Config file has been modified (mtime: %s -> %s), reloading AppConfig",
|
||||
_app_config_mtime,
|
||||
current_mtime,
|
||||
)
|
||||
elif _app_config_path == resolved_path and _app_config_signature != current_signature:
|
||||
logger.info("Config file content signature changed, reloading AppConfig")
|
||||
elif _app_config_path == resolved_path:
|
||||
logger.info("Extensions config changed, reloading AppConfig")
|
||||
_load_and_cache_app_config(str(resolved_path), extensions_source=current_extensions_source)
|
||||
should_reload = _app_config is None or _app_config_path != resolved_path or _app_config_signature != current_signature or current_extensions_source is not _app_config_extensions_source
|
||||
if should_reload:
|
||||
if _app_config_path == resolved_path and _app_config_mtime is not None and current_mtime is not None and _app_config_mtime != current_mtime:
|
||||
logger.info(
|
||||
"Config file has been modified (mtime: %s -> %s), reloading AppConfig",
|
||||
_app_config_mtime,
|
||||
current_mtime,
|
||||
)
|
||||
elif _app_config_path == resolved_path and _app_config_signature != current_signature:
|
||||
logger.info("Config file content signature changed, reloading AppConfig")
|
||||
elif _app_config_path == resolved_path:
|
||||
logger.info("Extensions config changed, reloading AppConfig")
|
||||
_load_and_cache_app_config(str(resolved_path), extensions_source=current_extensions_source)
|
||||
config = _app_config
|
||||
from deerflow.config.managed_models import merge_managed_models
|
||||
|
||||
return merge_managed_models(_app_config)
|
||||
return merge_managed_models(config)
|
||||
|
||||
|
||||
def reload_app_config(config_path: str | None = None) -> AppConfig:
|
||||
@ -837,12 +851,32 @@ def reload_app_config(config_path: str | None = None) -> AppConfig:
|
||||
config_path: Optional path to config file. If not provided,
|
||||
uses the default resolution strategy.
|
||||
|
||||
When *config_path* is given, its absolute path is pinned so working-directory
|
||||
changes cannot redirect later reads. Later edits are picked up without
|
||||
consulting default resolution until
|
||||
``reset_app_config()``, ``set_app_config()``, or an argument-less reload.
|
||||
An argument-less call clears any explicit pin before resolving, so ambient
|
||||
search is used again even if the previous load was from an explicit path.
|
||||
|
||||
Returns:
|
||||
The newly loaded AppConfig instance.
|
||||
"""
|
||||
global _app_config_source
|
||||
from deerflow.config.managed_models import merge_managed_models
|
||||
|
||||
return merge_managed_models(_load_and_cache_app_config(config_path))
|
||||
with _app_config_lock:
|
||||
if config_path is None:
|
||||
# Drop the pin before ambient resolve so a subsequent ``get_app_config``
|
||||
# does not keep following a discarded explicit file. Failed ambient
|
||||
# loads leave the cache content as-is but with no pin.
|
||||
_app_config_source = None
|
||||
loaded = _load_and_cache_app_config(config_path)
|
||||
# Record only after a successful load so a failed explicit reload keeps the
|
||||
# prior pin (mirrors ``_extensions_config_source``). Ambient refreshes go
|
||||
# through ``get_app_config`` -> ``_load_and_cache_app_config`` and must not
|
||||
# pin the resolved path.
|
||||
_app_config_source = str(_app_config_path) if config_path else None
|
||||
return merge_managed_models(loaded)
|
||||
|
||||
|
||||
def reset_app_config() -> None:
|
||||
@ -853,13 +887,15 @@ def reset_app_config() -> None:
|
||||
or when switching between different configurations.
|
||||
"""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_signature, _app_config_is_custom
|
||||
global _app_config_extensions_source
|
||||
_app_config = None
|
||||
_app_config_path = None
|
||||
_app_config_mtime = None
|
||||
_app_config_signature = None
|
||||
_app_config_is_custom = False
|
||||
_app_config_extensions_source = None
|
||||
global _app_config_source, _app_config_extensions_source
|
||||
with _app_config_lock:
|
||||
_app_config = None
|
||||
_app_config_path = None
|
||||
_app_config_mtime = None
|
||||
_app_config_signature = None
|
||||
_app_config_is_custom = False
|
||||
_app_config_source = None
|
||||
_app_config_extensions_source = None
|
||||
|
||||
|
||||
def set_app_config(config: AppConfig) -> None:
|
||||
@ -871,13 +907,15 @@ def set_app_config(config: AppConfig) -> None:
|
||||
config: The AppConfig instance to use.
|
||||
"""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_signature, _app_config_is_custom
|
||||
global _app_config_extensions_source
|
||||
_app_config = config
|
||||
_app_config_path = None
|
||||
_app_config_mtime = None
|
||||
_app_config_signature = None
|
||||
_app_config_is_custom = True
|
||||
_app_config_extensions_source = None
|
||||
global _app_config_source, _app_config_extensions_source
|
||||
with _app_config_lock:
|
||||
_app_config = config
|
||||
_app_config_path = None
|
||||
_app_config_mtime = None
|
||||
_app_config_signature = None
|
||||
_app_config_is_custom = True
|
||||
_app_config_source = None
|
||||
_app_config_extensions_source = None
|
||||
|
||||
|
||||
def peek_current_app_config() -> AppConfig | None:
|
||||
|
||||
279
backend/tests/test_app_config_explicit_path.py
Normal file
279
backend/tests/test_app_config_explicit_path.py
Normal file
@ -0,0 +1,279 @@
|
||||
"""Explicit ``reload_app_config(config_path)`` must pin that file for later reads.
|
||||
|
||||
Mirrors the ``_extensions_config_source`` contract: once an explicit path is
|
||||
reloaded, ``get_app_config()`` (and therefore ``DeerFlowClient(config_path=...)``)
|
||||
must keep following that file instead of ambient default resolution.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
import deerflow.config.app_config as app_config_module
|
||||
from deerflow.client import DeerFlowClient
|
||||
from deerflow.config.acp_config import load_acp_config_from_dict
|
||||
from deerflow.config.agents_api_config import load_agents_api_config_from_dict
|
||||
from deerflow.config.app_config import (
|
||||
AppConfig,
|
||||
get_app_config,
|
||||
reload_app_config,
|
||||
reset_app_config,
|
||||
set_app_config,
|
||||
)
|
||||
from deerflow.config.checkpointer_config import load_checkpointer_config_from_dict
|
||||
from deerflow.config.extensions_config import reset_extensions_config
|
||||
from deerflow.config.guardrails_config import load_guardrails_config_from_dict
|
||||
from deerflow.config.memory_config import load_memory_config_from_dict
|
||||
from deerflow.config.stream_bridge_config import load_stream_bridge_config_from_dict
|
||||
from deerflow.config.subagents_config import load_subagents_config_from_dict
|
||||
from deerflow.config.summarization_config import load_summarization_config_from_dict
|
||||
from deerflow.config.title_config import load_title_config_from_dict
|
||||
from deerflow.config.tool_search_config import load_tool_search_config_from_dict
|
||||
from deerflow.runtime.checkpointer import reset_checkpointer
|
||||
from deerflow.runtime.store import reset_store
|
||||
from deerflow.storage import reset_blob_store
|
||||
|
||||
|
||||
def _reset_singletons() -> None:
|
||||
reset_extensions_config()
|
||||
load_title_config_from_dict({})
|
||||
load_summarization_config_from_dict({})
|
||||
load_memory_config_from_dict({})
|
||||
load_agents_api_config_from_dict({})
|
||||
load_subagents_config_from_dict({})
|
||||
load_tool_search_config_from_dict({})
|
||||
load_guardrails_config_from_dict({})
|
||||
load_checkpointer_config_from_dict(None)
|
||||
load_stream_bridge_config_from_dict(None)
|
||||
load_acp_config_from_dict({})
|
||||
reset_checkpointer()
|
||||
reset_store()
|
||||
reset_blob_store()
|
||||
reset_app_config()
|
||||
|
||||
|
||||
def _write_config(path: Path, *, model_name: str) -> None:
|
||||
path.write_text(
|
||||
yaml.safe_dump(
|
||||
{
|
||||
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
|
||||
"models": [
|
||||
{
|
||||
"name": model_name,
|
||||
"use": "langchain_openai:ChatOpenAI",
|
||||
"model": "gpt-test",
|
||||
}
|
||||
],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
|
||||
def _model_names(config: AppConfig) -> list[str]:
|
||||
return [model.name for model in config.models]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def isolated_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||||
"""Point ambient resolution at an empty tmp project root (no config.yaml)."""
|
||||
monkeypatch.delenv("DEER_FLOW_CONFIG_PATH", raising=False)
|
||||
monkeypatch.setenv("DEER_FLOW_PROJECT_ROOT", str(tmp_path))
|
||||
# Avoid legacy monorepo candidates finding the checkout's config.yaml.
|
||||
monkeypatch.setattr(
|
||||
app_config_module,
|
||||
"_legacy_config_candidates",
|
||||
lambda: (),
|
||||
)
|
||||
_reset_singletons()
|
||||
yield tmp_path
|
||||
_reset_singletons()
|
||||
|
||||
|
||||
def test_get_app_config_follows_explicit_reload_without_ambient(isolated_project: Path) -> None:
|
||||
custom = isolated_project / "custom" / "config.yaml"
|
||||
custom.parent.mkdir()
|
||||
_write_config(custom, model_name="custom-model")
|
||||
|
||||
reload_app_config(str(custom))
|
||||
cfg = get_app_config()
|
||||
assert _model_names(cfg) == ["custom-model"]
|
||||
|
||||
|
||||
def test_client_config_path_overrides_sibling_default(isolated_project: Path) -> None:
|
||||
_write_config(isolated_project / "config.yaml", model_name="default-model")
|
||||
custom = isolated_project / "custom.yaml"
|
||||
_write_config(custom, model_name="custom-model")
|
||||
|
||||
client = DeerFlowClient(config_path=str(custom))
|
||||
assert _model_names(client._app_config) == ["custom-model"]
|
||||
assert _model_names(get_app_config()) == ["custom-model"]
|
||||
|
||||
|
||||
def test_client_config_path_works_with_no_default_config(isolated_project: Path) -> None:
|
||||
custom = isolated_project / "only-custom.yaml"
|
||||
_write_config(custom, model_name="custom-model")
|
||||
|
||||
client = DeerFlowClient(config_path=str(custom))
|
||||
assert _model_names(client._app_config) == ["custom-model"]
|
||||
assert _model_names(get_app_config()) == ["custom-model"]
|
||||
|
||||
|
||||
def test_explicit_pin_revalidates_edits_and_clears_on_reset(isolated_project: Path) -> None:
|
||||
custom = isolated_project / "pinned.yaml"
|
||||
_write_config(custom, model_name="first-model")
|
||||
reload_app_config(str(custom))
|
||||
assert _model_names(get_app_config()) == ["first-model"]
|
||||
|
||||
_write_config(custom, model_name="edited-model")
|
||||
assert _model_names(get_app_config()) == ["edited-model"]
|
||||
|
||||
other = isolated_project / "other.yaml"
|
||||
_write_config(other, model_name="other-model")
|
||||
reload_app_config(str(other))
|
||||
assert _model_names(get_app_config()) == ["other-model"]
|
||||
assert app_config_module._app_config_source == str(other)
|
||||
|
||||
# Failed reload keeps the prior pin.
|
||||
missing = isolated_project / "missing.yaml"
|
||||
with pytest.raises(FileNotFoundError):
|
||||
reload_app_config(str(missing))
|
||||
assert app_config_module._app_config_source == str(other)
|
||||
assert _model_names(get_app_config()) == ["other-model"]
|
||||
|
||||
# Argument-less reload clears the pin (and fails here: no ambient file).
|
||||
with pytest.raises(FileNotFoundError):
|
||||
reload_app_config()
|
||||
assert app_config_module._app_config_source is None
|
||||
|
||||
# Re-pin, then reset clears.
|
||||
reload_app_config(str(custom))
|
||||
assert app_config_module._app_config_source == str(custom)
|
||||
reset_app_config()
|
||||
assert app_config_module._app_config_source is None
|
||||
|
||||
# set_app_config clears the pin.
|
||||
reload_app_config(str(custom))
|
||||
injected = AppConfig.from_file(str(custom))
|
||||
set_app_config(injected)
|
||||
assert app_config_module._app_config_source is None
|
||||
assert get_app_config() is injected
|
||||
|
||||
|
||||
@pytest.mark.parametrize("entrypoint", ["reload", "client"])
|
||||
@pytest.mark.parametrize("has_decoy", [False, True])
|
||||
def test_relative_explicit_pin_survives_working_directory_changes(isolated_project: Path, monkeypatch: pytest.MonkeyPatch, entrypoint: str, has_decoy: bool) -> None:
|
||||
custom = isolated_project / "custom" / "config.yaml"
|
||||
custom.parent.mkdir()
|
||||
_write_config(custom, model_name="original-model")
|
||||
elsewhere = isolated_project / "elsewhere"
|
||||
elsewhere.mkdir()
|
||||
if has_decoy:
|
||||
decoy = elsewhere / "custom" / "config.yaml"
|
||||
decoy.parent.mkdir()
|
||||
_write_config(decoy, model_name="decoy-model")
|
||||
monkeypatch.chdir(isolated_project)
|
||||
if entrypoint == "client":
|
||||
DeerFlowClient(config_path="custom/config.yaml")
|
||||
else:
|
||||
reload_app_config("custom/config.yaml")
|
||||
|
||||
monkeypatch.chdir(elsewhere)
|
||||
|
||||
assert _model_names(get_app_config()) == ["original-model"]
|
||||
assert app_config_module._app_config_source == str(custom)
|
||||
_write_config(custom, model_name="edited-model")
|
||||
assert _model_names(get_app_config()) == ["edited-model"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("operation", ["get", "explicit-reload", "ambient-reload", "reset", "set"])
|
||||
def test_explicit_reload_publishes_cache_and_pin_atomically(isolated_project: Path, monkeypatch: pytest.MonkeyPatch, operation: str) -> None:
|
||||
ambient = isolated_project / "config.yaml"
|
||||
first = isolated_project / "first.yaml"
|
||||
second = isolated_project / "second.yaml"
|
||||
third = isolated_project / "third.yaml"
|
||||
for path, name in [(ambient, "ambient-model"), (first, "first-model"), (second, "second-model"), (third, "third-model")]:
|
||||
_write_config(path, model_name=name)
|
||||
injected = AppConfig.from_file(str(third))
|
||||
reload_app_config(str(first))
|
||||
|
||||
published = threading.Event()
|
||||
release = threading.Event()
|
||||
contender_reached = threading.Event()
|
||||
roles = threading.local()
|
||||
original_load = app_config_module._load_and_cache_app_config
|
||||
original_lock = getattr(app_config_module, "_app_config_lock", threading.RLock())
|
||||
|
||||
class ObservedLock:
|
||||
def __enter__(self):
|
||||
if getattr(roles, "value", None) == "contender":
|
||||
contender_reached.set()
|
||||
return original_lock.__enter__()
|
||||
|
||||
def __exit__(self, *args):
|
||||
return original_lock.__exit__(*args)
|
||||
|
||||
def pause_after_cache_load(*args, **kwargs):
|
||||
loaded = original_load(*args, **kwargs)
|
||||
if getattr(roles, "value", None) == "writer":
|
||||
published.set()
|
||||
assert release.wait(10), "writer was not released"
|
||||
return loaded
|
||||
|
||||
def writer():
|
||||
roles.value = "writer"
|
||||
return reload_app_config(str(second))
|
||||
|
||||
def contender():
|
||||
roles.value = "contender"
|
||||
try:
|
||||
if operation == "get":
|
||||
return get_app_config()
|
||||
if operation == "explicit-reload":
|
||||
return reload_app_config(str(third))
|
||||
if operation == "ambient-reload":
|
||||
return reload_app_config()
|
||||
if operation == "reset":
|
||||
return reset_app_config()
|
||||
return set_app_config(injected)
|
||||
finally:
|
||||
# Unfixed operations complete without a critical section. Observe that
|
||||
# boundary too, so the test reproduces the race without timing sleeps.
|
||||
contender_reached.set()
|
||||
|
||||
monkeypatch.setattr(app_config_module, "_load_and_cache_app_config", pause_after_cache_load)
|
||||
monkeypatch.setattr(app_config_module, "_app_config_lock", ObservedLock(), raising=False)
|
||||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||||
pending_writer = pool.submit(writer)
|
||||
try:
|
||||
assert published.wait(10), "writer did not publish its cache"
|
||||
pending_contender = pool.submit(contender)
|
||||
assert contender_reached.wait(10), "contender did not reach the cache operation"
|
||||
finally:
|
||||
release.set()
|
||||
assert _model_names(pending_writer.result(timeout=10)) == ["second-model"]
|
||||
contender_result = pending_contender.result(timeout=10)
|
||||
|
||||
expected_source = {"get": str(second), "explicit-reload": str(third)}.get(operation)
|
||||
assert app_config_module._app_config_source == expected_source
|
||||
if operation == "get":
|
||||
assert _model_names(contender_result) == ["second-model"]
|
||||
if operation == "set":
|
||||
assert get_app_config() is injected
|
||||
if operation == "reset":
|
||||
assert app_config_module._app_config is None
|
||||
expected_model = {
|
||||
"get": "second-model",
|
||||
"explicit-reload": "third-model",
|
||||
"ambient-reload": "ambient-model",
|
||||
"reset": "ambient-model",
|
||||
"set": "third-model",
|
||||
}[operation]
|
||||
if operation != "reset":
|
||||
assert _model_names(app_config_module._app_config) == [expected_model]
|
||||
assert _model_names(get_app_config()) == [expected_model]
|
||||
Loading…
x
Reference in New Issue
Block a user