"""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)