deer-flow/backend/tests/test_persistence_bootstrap_sqlite_lock.py
AnoobFeng debb0fd161
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 (#3706)
* 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.
2026-06-24 13:57:12 +08:00

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