From 545595b0773e7232c96afe6090762535cbb8aa5f Mon Sep 17 00:00:00 2001 From: ryan echternacht Date: Tue, 29 Sep 2026 17:00:17 -0400 Subject: [PATCH] Evaluate check_flags from the cache when the replicator is not ready check_flags gated its DataStream path on is_connected(), which in replicator mode is the replicator's ready flag. When a replicator reports ready: false (account closed, Schematic unreachable) it keeps its Redis cache, but check_flags skipped that cache and went to the bulk API, which cannot answer either, so callers got flag defaults. check_flag had no such gate and kept evaluating from the cache. Drop the gate so check_flags matches check_flag and the Go SDK: evaluate requested keys through DataStream and fall back to the bulk API only when local evaluation raises. Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 3 +- src/schematic/client.py | 11 +- tests/custom/test_client.py | 20 +-- tests/custom/test_replicator_flag_checks.py | 149 ++++++++++++++++++++ 4 files changed, 170 insertions(+), 13 deletions(-) create mode 100644 tests/custom/test_replicator_flag_checks.py diff --git a/README.md b/README.md index 1596b8a..a357984 100644 --- a/README.md +++ b/README.md @@ -895,7 +895,8 @@ When running in Replicator Mode, the client will: - Skip establishing WebSocket connections - Periodically check if the replicator service is ready - Use cached data populated by the external replicator service -- Fall back to direct API calls if the replicator is not available +- Keep evaluating `check_flag` and `check_flags` from that cache when the replicator reports not ready (for example, when it cannot reach Schematic), since the replicator keeps its cache in place +- Fall back to direct API calls only when a flag cannot be evaluated locally, such as when the flag is missing from the cache ## Contributing diff --git a/src/schematic/client.py b/src/schematic/client.py index 72b4be7..36c3c56 100644 --- a/src/schematic/client.py +++ b/src/schematic/client.py @@ -1376,10 +1376,15 @@ async def check_flags( return [self._default_response(k, options, REASON_OFFLINE) for k in keys] # DataStream evaluation only makes sense when specific keys are - # requested AND the client is connected — the "give me everything" - # semantic only exists via the bulk API. + # requested, since the "give me everything" semantic only exists via + # the bulk API. It does not gate on is_connected(), matching + # check_flag: in replicator mode a replicator that reports not ready + # still leaves its cache in place, so keys evaluate from what the + # cache holds. In WebSocket mode a disconnected client evaluates from + # the cache when the entities are there and raises when they are not, + # which falls back to the bulk API below. ds = self._get_datastream() - if ds is not None and ds.is_connected() and flag_keys: + if ds is not None and flag_keys: try: results: List[CheckFlagResponseData] = [] for flag_key in flag_keys: diff --git a/tests/custom/test_client.py b/tests/custom/test_client.py index 981c221..3b66ca5 100644 --- a/tests/custom/test_client.py +++ b/tests/custom/test_client.py @@ -1446,8 +1446,11 @@ async def test_check_flags_offline_with_no_keys_returns_all_defaults(self): ("flag_a", True), ("flag_b", False), ] - async def test_check_flags_skips_datastream_when_not_connected(self): - """If DataStream is configured but not connected, skip it and use the bulk API.""" + async def test_check_flags_uses_datastream_when_not_connected(self): + """check_flags does not gate on is_connected(), matching check_flag: + a DataStream client that reports not connected (or a replicator that + reports not ready) still answers from its cache, and the bulk API is + only the fallback for when that evaluation raises.""" config = AsyncSchematicConfig( logger=MagicMock(), httpx_client=MagicMock(spec=AsyncClient), @@ -1458,18 +1461,17 @@ async def test_check_flags_skips_datastream_when_not_connected(self): try: mock_ds = MagicMock() mock_ds.is_connected = MagicMock(return_value=False) - mock_ds.check_flag = AsyncMock() + mock_ds.check_flag = AsyncMock(return_value=RulesengineCheckFlagResult( + value=True, flag_key="flag_a", reason="match", + )) client._datastream_client = mock_ds client.flag_check_cache_providers = [] - - client.features.check_flags = AsyncMock(return_value=self._bulk_response([ - CheckFlagResponseData(value=True, flag="flag_a", reason="match"), - ])) + client.features.check_flags = AsyncMock() results = await client.check_flags(["flag_a"]) assert results[0].value is True - mock_ds.check_flag.assert_not_called() - client.features.check_flags.assert_called_once() + mock_ds.check_flag.assert_awaited_once() + client.features.check_flags.assert_not_called() finally: await client.event_buffer.stop() diff --git a/tests/custom/test_replicator_flag_checks.py b/tests/custom/test_replicator_flag_checks.py new file mode 100644 index 0000000..bfd3e3a --- /dev/null +++ b/tests/custom/test_replicator_flag_checks.py @@ -0,0 +1,149 @@ +"""Flag checks in replicator mode while the replicator reports not ready. + +When a Schematic account is closed or Schematic is unreachable, the replicator +stays up and keeps its cache but reports ``ready: false``. Flag checks should +keep evaluating from that cache, as the Go SDK does, rather than falling back +to the API (which cannot answer either) and handing back defaults. +""" + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from httpx import AsyncClient +from lease_support import make_fake_redis + +from schematic.cache import RedisCache +from schematic.client import AsyncSchematic, AsyncSchematicConfig, DataStreamConfig +from schematic.types import ( + RulesengineCompany, + RulesengineCondition, + RulesengineFlag, + RulesengineRule, +) + +COMPANY_ID = "co_cached" +FLAG_KEY = "cached-flag" + + +def _company() -> RulesengineCompany: + # A company override grants the flag, so a True verdict can only come from + # evaluating the cached company. The flag's own default is False. + condition = RulesengineCondition( + id="cond_company", + account_id="acc_1", + environment_id="env_1", + condition_type="company", + operator="eq", + resource_ids=[COMPANY_ID], + trait_value="", + ) + rule = RulesengineRule( + id="rule_override", + flag_id="flag_1", + account_id="acc_1", + environment_id="env_1", + name="Company Override", + rule_type="company_override", + value=True, + priority=0, + conditions=[condition], + condition_groups=[], + ) + return RulesengineCompany( + id=COMPANY_ID, + account_id="acc_1", + environment_id="env_1", + keys={"id": COMPANY_ID}, + traits=[], + metrics=[], + rules=[rule], + entitlements=[], + billing_product_ids=[], + credit_balances={}, + plan_ids=[], + plan_version_ids=[], + ) + + +def _flag() -> RulesengineFlag: + return RulesengineFlag( + id="flag_1", + key=FLAG_KEY, + account_id="acc_1", + environment_id="env_1", + default_value=False, + rules=[], + ) + + +def _health_client(body: dict) -> MagicMock: + resp = MagicMock() + resp.raise_for_status = MagicMock() + resp.json = MagicMock(return_value=body) + client = MagicMock() + client.get = AsyncMock(return_value=resp) + return client + + +async def _replicator_client_not_ready() -> AsyncSchematic: + """An AsyncSchematic in replicator mode whose replicator has just reported + ``ready: false`` with the company and flag still in its Redis cache.""" + redis: RedisCache[Any] = RedisCache(make_fake_redis()) + client = AsyncSchematic("test_key", AsyncSchematicConfig( + logger=MagicMock(), + httpx_client=MagicMock(spec=AsyncClient), + event_buffer_period=1, + flag_defaults={FLAG_KEY: False}, + use_datastream=True, + datastream=DataStreamConfig( + replicator_mode=True, + replicator_health_url="http://replicator.test/ready", + company_cache=redis, + company_lookup_cache=redis, + user_cache=redis, + user_lookup_cache=redis, + flag_cache=redis, + ), + )) + ds = client._datastream_client + assert ds is not None + await ds._rules_engine.initialize() + + # The replicator was ready, then reports not ready. The poll is what the + # background health loop runs, driven once here. + ds._replicator_ready = True + ds._health_check_client = _health_client({"ready": False, "cache_version": "v1"}) + await ds._check_replicator_health() + assert not ds.is_connected() + + await ds._cache_company(_company()) + await ds._flag_cache.set(ds._flag_cache_key(FLAG_KEY), _flag()) + + # The API is unreachable too, so any fallback to it would surface as the + # configured default (False) rather than the cached verdict. + client.features.check_flag = AsyncMock(side_effect=RuntimeError("api unreachable")) + client.features.check_flags = AsyncMock(side_effect=RuntimeError("api unreachable")) + client.flag_check_cache_providers = [] + return client + + +async def test_check_flag_evaluates_from_cache_when_replicator_not_ready() -> None: + client = await _replicator_client_not_ready() + try: + resp = await client.check_flag_with_entitlement(FLAG_KEY, company={"id": COMPANY_ID}) + assert resp.value is True + assert resp.company_id == COMPANY_ID + client.features.check_flag.assert_not_called() + finally: + await client.shutdown() + + +async def test_check_flags_evaluates_from_cache_when_replicator_not_ready() -> None: + client = await _replicator_client_not_ready() + try: + results = await client.check_flags([FLAG_KEY], company={"id": COMPANY_ID}) + assert [r.value for r in results] == [True] + assert results[0].company_id == COMPANY_ID + client.features.check_flags.assert_not_called() + finally: + await client.shutdown()