mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-09 16:08:31 +00:00
Some checks are pending
Backend Blocking IO / backend-blocking-io (push) Waiting to run
Unit Tests / backend-unit-tests (push) Waiting to run
E2E Tests / e2e-tests (push) Waiting to run
Frontend Unit Tests / frontend-unit-tests (push) Waiting to run
Lint Check / lint-backend (push) Waiting to run
Lint Check / lint-frontend (push) Waiting to run
Replay E2E (front-back contract) / Layer 1 — backend golden (no API key) (push) Waiting to run
Replay E2E (front-back contract) / Layer 2 — full-stack render (no API key) (push) Waiting to run
* feat(persistence): wire alembic migrations + bootstrap schema on startup Closes #3682. Pre-#3658 DBs lack the `runs.token_usage_by_model` column because alembic was never wired up — startup only ran `create_all`, which never ALTERs existing tables. Adds a hybrid bootstrap in FastAPI lifespan (replaces bare `create_all`): - empty DB → create_all + stamp head - legacy DB → stamp 0001_baseline + upgrade head - versioned DB → upgrade head Concurrency: Postgres `pg_advisory_lock` (cross-process); SQLite per-engine `asyncio.Lock` + 30s `PRAGMA busy_timeout` on both prod and alembic engines. Column revisions use `safe_add_column` / `safe_drop_column` idempotent helpers as fallback. Other bits: - 0001 baseline (chain root) + 0002 add `runs.token_usage_by_model` - `include_object` filter so alembic ignores LangGraph checkpointer tables - `make migrate-rev MSG="..."` for authoring new revisions (no migrate/stamp targets — startup is the only execution path) - Tests: three-branch decision, concurrency, #3682 regression, env filter, blocking-IO gate anchor - CLAUDE.md: new "Schema migrations" section * fix(style): fix lint error * perf(persistence): address review feedback on alembic bootstrap Behavioural fixes - _SQLITE_LOCKS now keyed via WeakKeyDictionary so id-reuse after GC cannot return a stale, loop-bound lock and the cache cannot leak one entry per disposed engine. - safe_add_column compares nullable / server_default against the desired column when the name already exists and emits a warning on drift, surfacing manual-ALTER workarounds instead of silently no-op'ing. - _postgres_lock issues SET LOCAL idle_in_transaction_session_timeout=0 before pg_advisory_lock, so managed Postgres cannot kill the idle lock-holding session mid-upgrade and silently release the advisory lock. - legacy branch now backfills missing baseline tables via a restricted create_all (Base.metadata.create_all scoped to _BASELINE_TABLE_NAMES). Restores pre-#1930 upgraders whose channel_* tables were never provisioned, without pre-empting future create_table revisions for newly-added models. Schema parity - runs.token_usage_by_model gains server_default=text("'{}'") in both the ORM model and the 0001_baseline create_table, matching what 0002 adds via ALTER. create_all and alembic-upgrade paths now produce identical column definitions. - New parity test compares Base.metadata.create_all output against a pure alembic upgrade base->head, asserting column-set, nullable, and server_default agree across all tables (normalized through the same helper safe_add_column's drift check uses). Guards - test_baseline_table_names_constant_matches_0001 pins _BASELINE_TABLE_NAMES to 0001_baseline.upgrade()'s actual output -- the constant cannot drift silently when someone edits 0001. - test_legacy_backfill_skips_non_baseline_tables verifies the restricted backfill does not create a phantom table on Base.metadata, modelling a future revision that would otherwise collide on op.create_table. Doc residuals - Three-branch decision table is now consistent across bootstrap.py top docstring, engine.py comment, test module docstring, and CLAUDE.md. - Stale test anchor in blocking_io/test_persistence_engine_sqlite.py docstring now points at the real file. * fix(style): fix lint error * fix(persistence): close drift detection holes - _check_column_drift compares column type via a family equivalence allowlist ({JSON, JSONB}). Catches the wrong-type workaround `TEXT NOT NULL DEFAULT '{}'` that previously slipped through silently, while keeping Postgres JSON/JSONB dialect reflection quiet. Reflected and desired type are also echoed in every drift warning's payload for operator triage. - Extract _escape_url_for_alembic so bootstrap._alembic_safe_url and scripts/_autogen_revision share the ConfigParser % escape rule instead of duplicating it. - backend/README.md: add `make migrate-rev MSG=...` to Commands and a Schema Migrations section per the repo's README/CLAUDE.md sync policy. - test_base_to_dict.py: scope the test ORM class to an isolated MetaData so the create_all-vs-alembic parity test (added in the previous commit) is not polluted by the phantom table on the full pytest session.
116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
"""Regression tests for the per-engine SQLite bootstrap lock cache.
|
|
|
|
The cache (``deerflow.persistence.bootstrap._SQLITE_LOCKS``) maps an engine
|
|
to the ``asyncio.Lock`` that serialises its in-process bootstrap. It is keyed
|
|
by the engine object itself via ``WeakKeyDictionary`` -- not ``id(engine)`` --
|
|
to avoid two failure modes that are silent in production (one long-lived
|
|
engine) but real in pytest (one fresh engine per test):
|
|
|
|
1. **CPython id reuse.** After an engine is garbage-collected its memory
|
|
address can be reused by a new engine. An ``id``-keyed cache would hand
|
|
the new engine the dead engine's ``Lock``. That lock was bound to the
|
|
dead engine's event loop at first ``async with``; pytest gives each async
|
|
test its own loop, so reusing it raises ``RuntimeError: ... bound to a
|
|
different event loop``.
|
|
2. **Unbounded growth.** An ``id``-keyed cache never drops entries because
|
|
nothing notifies it when the engine dies. With ``WeakKeyDictionary`` the
|
|
entry disappears as soon as the engine is collected.
|
|
|
|
These tests do not open any DB connection -- they exercise the cache helper
|
|
directly so they can run without an event loop and without aiosqlite warnings
|
|
about unclosed engines.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gc
|
|
import weakref
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from deerflow.persistence import bootstrap as bootstrap_mod
|
|
from deerflow.persistence.bootstrap import _get_sqlite_local_lock
|
|
|
|
|
|
def _make_engine():
|
|
return create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
|
|
|
|
def test_cache_is_weak_key_dictionary() -> None:
|
|
"""Pin the cache type so a refactor cannot silently revert to a plain
|
|
dict (which would reintroduce the id-reuse bug)."""
|
|
assert isinstance(bootstrap_mod._SQLITE_LOCKS, weakref.WeakKeyDictionary)
|
|
|
|
|
|
def test_same_engine_returns_same_lock() -> None:
|
|
engine = _make_engine()
|
|
assert _get_sqlite_local_lock(engine) is _get_sqlite_local_lock(engine)
|
|
|
|
|
|
def test_distinct_engines_get_distinct_locks() -> None:
|
|
"""Two live engines must not share a lock -- otherwise unrelated
|
|
bootstraps would serialise against each other."""
|
|
engine_a = _make_engine()
|
|
engine_b = _make_engine()
|
|
assert _get_sqlite_local_lock(engine_a) is not _get_sqlite_local_lock(engine_b)
|
|
|
|
|
|
def test_entry_drops_when_engine_is_garbage_collected() -> None:
|
|
"""The cache must not pin the engine alive.
|
|
|
|
This is the structural guarantee behind the id-reuse fix: when the engine
|
|
is collected, its lock entry goes with it, so a future engine landing on
|
|
the same address cannot inherit a stale, loop-bound lock.
|
|
"""
|
|
engine = _make_engine()
|
|
_get_sqlite_local_lock(engine)
|
|
assert engine in bootstrap_mod._SQLITE_LOCKS
|
|
|
|
engine_ref = weakref.ref(engine)
|
|
del engine
|
|
gc.collect()
|
|
|
|
assert engine_ref() is None, "engine should be collectible -- cache must not hold a strong ref"
|
|
# WeakKeyDictionary may defer removal until the next access; touch it.
|
|
assert all(ref() is not None for ref in bootstrap_mod._SQLITE_LOCKS.keyrefs())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fresh_engine_gets_lock_usable_on_current_loop() -> None:
|
|
"""End-to-end guard for the pytest pattern: a brand-new engine in a
|
|
brand-new event loop must receive a lock that ``async with`` accepts.
|
|
|
|
This is the behaviour an ``id``-keyed cache could break if the new engine
|
|
landed on a previously-used address -- it would return a lock bound to a
|
|
dead loop and raise ``RuntimeError: ... bound to a different event loop``.
|
|
"""
|
|
engine = _make_engine()
|
|
try:
|
|
lock = _get_sqlite_local_lock(engine)
|
|
async with lock:
|
|
pass
|
|
# Re-entrant acquire on the same loop must also succeed.
|
|
async with lock:
|
|
pass
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_does_not_grow_across_disposed_engines() -> None:
|
|
"""Create + dispose + drop many engines and assert the cache stays bounded.
|
|
|
|
Without ``WeakKeyDictionary`` this loop would leak one entry per engine.
|
|
"""
|
|
initial = len(bootstrap_mod._SQLITE_LOCKS)
|
|
for _ in range(20):
|
|
engine = _make_engine()
|
|
_get_sqlite_local_lock(engine)
|
|
await engine.dispose()
|
|
del engine
|
|
gc.collect()
|
|
# Touch the dict so WeakKeyDictionary clears any deferred removals.
|
|
_ = list(bootstrap_mod._SQLITE_LOCKS.items())
|
|
# Allow a small slack for any engine that is still pinned by a frame.
|
|
assert len(bootstrap_mod._SQLITE_LOCKS) - initial <= 1
|