202 lines
10 KiB
Python
202 lines
10 KiB
Python
"""Concurrent changes must not lose controls or serve invalidated catalogue data."""
|
|
|
|
import asyncio
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
from test_api_read_cache import FakeRedis, _config
|
|
|
|
from app.gateway import api_read_cache as cache
|
|
from deerflow.persistence.api_read_cache_settings.model import ApiReadCacheSettingRow
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def settings_database(monkeypatch, tmp_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'edge-controls.db'}")
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(ApiReadCacheSettingRow.__table__.create)
|
|
|
|
class SlowReadSession(AsyncSession):
|
|
async def get(self, *args, **kwargs):
|
|
result = await super().get(*args, **kwargs)
|
|
# Simulate real database latency between reading settings and writing
|
|
# an independently edited interface toggle from another admin tab.
|
|
await asyncio.sleep(0.025)
|
|
return result
|
|
|
|
factory = async_sessionmaker(engine, class_=SlowReadSession, expire_on_commit=False)
|
|
monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: factory)
|
|
monkeypatch.setattr(cache, "_api_read_cache_settings_cache", None)
|
|
yield factory
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_first_saves_keep_both_individual_choices(settings_database):
|
|
config = _config()
|
|
results = await asyncio.gather(
|
|
cache.save_api_read_cache_settings(None, config=config, collection_overrides={"agents": False}),
|
|
cache.save_api_read_cache_settings(None, config=config, collection_overrides={"skills": False}),
|
|
return_exceptions=True,
|
|
)
|
|
assert not [result for result in results if isinstance(result, Exception)]
|
|
async with settings_database() as session:
|
|
row = await session.get(ApiReadCacheSettingRow, "default")
|
|
assert row.collection_overrides == {"agents": False, "skills": False}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_updates_do_not_replace_other_interface_choices(settings_database):
|
|
config = _config()
|
|
await cache.save_api_read_cache_settings(True, 777, config=config)
|
|
await asyncio.gather(
|
|
cache.save_api_read_cache_settings(None, config=config, collection_overrides={"agents": False}),
|
|
cache.save_api_read_cache_settings(None, config=config, collection_overrides={"skills": False}),
|
|
)
|
|
async with settings_database() as session:
|
|
row = await session.get(ApiReadCacheSettingRow, "default")
|
|
assert row.collection_overrides == {"agents": False, "skills": False}
|
|
assert row.enabled is True
|
|
assert row.ttl_seconds == 777
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("change", ["disable", "invalidate"])
|
|
async def test_waiting_requests_reject_fill_changed_before_wakeup(monkeypatch, change):
|
|
config = _config()
|
|
redis = FakeRedis()
|
|
monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: None)
|
|
monkeypatch.setattr(cache, "_api_read_cache_settings_cache", None)
|
|
monkeypatch.setattr(cache, "_fill_events", {})
|
|
monkeypatch.setattr(cache, "_fill_local_payloads", {})
|
|
monkeypatch.setattr(cache, "_client", lambda *_args, **_kwargs: (redis, "test:edge"))
|
|
version = await cache.get_cache_version(config, "agents")
|
|
_, owner = await cache.wait_for_cached_json_or_claim_fill(config, "agents", "u1", version=version)
|
|
assert owner
|
|
waiter = asyncio.create_task(cache.wait_for_cached_json_or_claim_fill(config, "agents", "u1", version=version, wait_ms=1000))
|
|
await asyncio.sleep(0.01)
|
|
await cache.set_cached_json(config, "agents", "u1", {"old": True}, expected_version=version)
|
|
if change == "disable":
|
|
cache.set_api_read_cache_settings_local(True, 300, collection_overrides={"agents": False})
|
|
else:
|
|
await cache.invalidate_cached_collection(config, "agents", force=True)
|
|
await cache.release_cache_fill(config, "agents", "u1", owner)
|
|
value, new_owner = await waiter
|
|
if new_owner:
|
|
await cache.release_cache_fill(config, "agents", "u1", new_owner)
|
|
assert value is None, "A waiting request must not use the old local payload after controls/data changed"
|
|
|
|
|
|
def test_controls_migration_preserves_existing_ttl_and_defaults_status_to_enabled(tmp_path):
|
|
migration_path = Path(__file__).parents[1] / "packages/harness/deerflow/persistence/migrations/versions/20261029_02_api_performance_controls.py"
|
|
spec = importlib.util.spec_from_file_location("performance_controls_migration", migration_path)
|
|
migration = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(migration)
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'migration.db'}")
|
|
with engine.begin() as connection:
|
|
connection.execute(text("CREATE TABLE api_read_cache_settings (key VARCHAR(32) PRIMARY KEY, enabled BOOLEAN NOT NULL, ttl_seconds INTEGER NOT NULL)"))
|
|
connection.execute(text("INSERT INTO api_read_cache_settings VALUES ('default', 1, 777)"))
|
|
with Operations.context(MigrationContext.configure(connection)):
|
|
migration.upgrade()
|
|
migration.upgrade() # Existing/create_all schemas are safe to revisit.
|
|
row = connection.execute(text("SELECT * FROM api_read_cache_settings")).mappings().one()
|
|
assert row["ttl_seconds"] == 777
|
|
assert bool(row["enabled"]) is True
|
|
assert bool(row["browser_status_enabled"]) is True
|
|
assert row["collection_overrides"] is None
|
|
assert row["cache_epoch"] == "0"
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_saved_epoch_survives_redis_outage_and_process_restart(settings_database, monkeypatch):
|
|
config = _config()
|
|
|
|
class RecoverableRedis(FakeRedis):
|
|
online = True
|
|
|
|
async def get(self, key):
|
|
if not self.online:
|
|
raise ConnectionError("cache offline")
|
|
return await super().get(key)
|
|
|
|
async def incr(self, key):
|
|
if not self.online:
|
|
raise ConnectionError("cache offline")
|
|
return await super().incr(key)
|
|
|
|
redis = RecoverableRedis()
|
|
# Keep the real _client prefix construction, while using an in-memory
|
|
# Redis connection shared by the simulated old/new Gateway processes.
|
|
monkeypatch.setattr(cache, "_clients", {})
|
|
monkeypatch.setattr("redis.asyncio.Redis.from_url", lambda *args, **kwargs: redis)
|
|
await cache.save_api_read_cache_settings(True, config=config)
|
|
await cache.set_cached_json(config, "agents", "u1", {"old": True})
|
|
assert await cache.get_cached_json(config, "agents", "u1") == {"old": True}
|
|
old_keys = set(redis.values)
|
|
old_version = await cache.get_cache_version(config, "agents")
|
|
redis.online = False
|
|
await cache.save_api_read_cache_settings(False, config=config)
|
|
await cache.invalidate_cached_collection(config, "agents", force=True)
|
|
await cache.save_api_read_cache_settings(True, config=config)
|
|
await cache.invalidate_cached_collection(config, "agents", force=True)
|
|
|
|
# Restart clears every process-local control/client; DB and Redis survive.
|
|
monkeypatch.setattr(cache, "_api_read_cache_settings_cache", None)
|
|
monkeypatch.setattr(cache, "_clients", {})
|
|
redis.online = True
|
|
assert await cache.get_cached_json(config, "agents", "u1") is None
|
|
assert old_keys.issubset(redis.values), "Old keys still exist but must be unreachable"
|
|
assert await cache.get_cache_version(config, "agents") != old_version
|
|
await cache.set_cached_json(config, "agents", "u1", {"fresh": True})
|
|
assert await cache.get_cached_json(config, "agents", "u1") == {"fresh": True}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_fences_fills_started_in_the_previous_epoch(settings_database, monkeypatch):
|
|
config = _config()
|
|
redis = FakeRedis()
|
|
monkeypatch.setattr(cache, "_clients", {})
|
|
monkeypatch.setattr("redis.asyncio.Redis.from_url", lambda *args, **kwargs: redis)
|
|
await cache.save_api_read_cache_settings(True, config=config)
|
|
version = await cache.get_cache_version(config, "agents")
|
|
_, owner = await cache.wait_for_cached_json_or_claim_fill(config, "agents", "u1", version=version)
|
|
waiter = asyncio.create_task(cache.wait_for_cached_json_or_claim_fill(config, "agents", "u1", version=version, wait_ms=1000))
|
|
await asyncio.sleep(0.01)
|
|
await cache.save_api_read_cache_settings(True, config=config, collection_overrides={"skills": False})
|
|
await cache.set_cached_json(config, "agents", "u1", {"stale": True}, expected_version=version)
|
|
await cache.release_cache_fill(config, "agents", "u1", owner)
|
|
value, token = await waiter
|
|
assert (value, token) == (None, None)
|
|
assert await cache.get_cached_json(config, "agents", "u1") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_old_epoch_owner_release_does_not_wake_new_epoch_waiters(settings_database, monkeypatch):
|
|
config = _config()
|
|
redis = FakeRedis()
|
|
monkeypatch.setattr(cache, "_clients", {})
|
|
monkeypatch.setattr("redis.asyncio.Redis.from_url", lambda *args, **kwargs: redis)
|
|
await cache.save_api_read_cache_settings(True, config=config)
|
|
old_version = await cache.get_cache_version(config, "agents")
|
|
_, old_owner = await cache.wait_for_cached_json_or_claim_fill(config, "agents", "u1", version=old_version)
|
|
await cache.save_api_read_cache_settings(True, config=config)
|
|
version = await cache.get_cache_version(config, "agents")
|
|
_, owner = await cache.wait_for_cached_json_or_claim_fill(config, "agents", "u1", version=version)
|
|
waiter = asyncio.create_task(
|
|
cache.wait_for_cached_json_or_claim_fill(config, "agents", "u1", version=version, wait_ms=1000)
|
|
)
|
|
await asyncio.sleep(0.01)
|
|
await cache.release_cache_fill(config, "agents", "u1", old_owner)
|
|
await asyncio.sleep(0.01)
|
|
assert not waiter.done()
|
|
await cache.set_cached_json(config, "agents", "u1", {"fresh": True}, expected_version=version)
|
|
await cache.release_cache_fill(config, "agents", "u1", owner)
|
|
assert await waiter == ({"fresh": True}, None)
|