From 04f3f3b410eacd0d72694fece1de7c7ad94b203b Mon Sep 17 00:00:00 2001 From: ryan echternacht Date: Wed, 30 Sep 2026 11:06:25 -0400 Subject: [PATCH 1/5] Serve replicator-mode flag checks only once the cache is ready In replicator mode, check_flag and check_flag_with_entitlement read the shared cache with no readiness gate, while check_flags with keys gated on is_connected(). The health poll also called raise_for_status(), so a 503 from /ready never had its body read and its cache_version was dropped. - Health poll reads the JSON body on any status, sets readiness from the ready field, records any non-empty cache_version, and on a failed poll sets not ready and keeps the last cache_version. - Add DataStreamClient.is_cache_ready(). is_connected() is unchanged and documented as reporting replicator readiness in replicator mode. - Single, bulk and the client-mode credit lease check share one gate, AsyncSchematic._get_flag_check_datastream(): while the cache is not ready they take the API path. - README describes the readiness behavior. Depends on SchematicHQ/schematic-replicator#143. Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 8 + src/schematic/client.py | 27 +- src/schematic/datastream/datastream_client.py | 86 +++-- tests/custom/test_client.py | 1 + tests/custom/test_replicator_cache_ready.py | 353 ++++++++++++++++++ tests/datastream/test_datastream_client.py | 11 +- 6 files changed, 453 insertions(+), 33 deletions(-) create mode 100644 tests/custom/test_replicator_cache_ready.py diff --git a/README.md b/README.md index c4b73ac1..8c7c38b2 100644 --- a/README.md +++ b/README.md @@ -931,6 +931,14 @@ When running in Replicator Mode, the client will: - Use cached data populated by the external replicator service - Fall back to direct API calls if the replicator is not available +#### Cache Readiness + +The SDK serves flag checks from the shared cache only once the replicator reports that the cache is ready. The health endpoint answers `503` with `"ready": false` while the replicator is still loading the cache for its `cache_version`, and `200` with `"ready": true` once that load is complete. Until the SDK has seen `"ready": true`, `check_flag`, `check_flag_with_entitlement` and `check_flags` skip the cache and call the Schematic API, falling back to the flag default if the API call fails. Once the cache is ready, single and bulk checks both evaluate locally from the cache, and a flag that is not in the cache still falls back to the API. + +If a health check fails (connection error, timeout, or an unreadable response), the SDK treats the cache as not ready until the next successful check reports `"ready": true`, and keeps using the last `cache_version` it saw. + +`DataStreamClient.is_cache_ready()` reports this readiness. `is_connected()` returns the same value in Replicator Mode and is kept for backward compatibility. + ## Contributing While we value open-source contributions to this SDK, this library is generated programmatically. diff --git a/src/schematic/client.py b/src/schematic/client.py index 5024b820..ae0a20dc 100644 --- a/src/schematic/client.py +++ b/src/schematic/client.py @@ -1365,6 +1365,21 @@ async def initialize(self) -> None: def _get_datastream(self) -> Optional[DataStreamClient]: return self._datastream_client + def _get_flag_check_datastream(self) -> Optional[DataStreamClient]: + """The DataStream client flag checks may evaluate against, or None to + use the API. + + Single and bulk flag checks both go through this gate so they cannot + drift. In replicator mode the shared cache is read only once the + replicator reports it ready (``is_cache_ready()``); before that the + check takes the API path, which falls back to the flag default if the + API fails. + """ + ds = self._get_datastream() + if ds is None or not ds.is_cache_ready(): + return None + return ds + async def check_flag( self, flag_key: str, @@ -1386,7 +1401,7 @@ async def check_flag_with_entitlement( return self._default_response(flag_key, options, REASON_OFFLINE) # Try DataStream first if available - ds = self._get_datastream() + ds = self._get_flag_check_datastream() if ds is not None: try: resp = await ds.check_flag( @@ -1414,8 +1429,10 @@ async def check_flags( # 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. - ds = self._get_datastream() + # semantic only exists via the bulk API. The readiness gate is shared + # with check_flag; in replicator mode is_connected() is the same cache + # readiness, so only websocket mode additionally needs a live socket. + ds = self._get_flag_check_datastream() if ds is not None and ds.is_connected() and flag_keys: try: results: List[CheckFlagResponseData] = [] @@ -1672,7 +1689,9 @@ async def enqueue_flag_check(body: EventBodyFlagCheck) -> None: return await check_with_lease( CreditCheckDeps( - datastream=self._datastream_client, + # Same readiness gate as the plain checks: with no DataStream + # the lease flow defers to the plain check, which uses the API. + datastream=self._get_flag_check_datastream(), lease_store=lease_store, reservations=reservations, manager=manager, diff --git a/src/schematic/datastream/datastream_client.py b/src/schematic/datastream/datastream_client.py index 3c128168..dcc72acd 100644 --- a/src/schematic/datastream/datastream_client.py +++ b/src/schematic/datastream/datastream_client.py @@ -314,10 +314,33 @@ async def start(self) -> None: self._ws_client.start() def is_connected(self) -> bool: + """Whether the DataStream can serve data. + + In websocket mode this reports whether the websocket is connected. In + replicator mode it reports replicator readiness, the same value as + ``is_cache_ready()``, which is the clearer name for that question. + """ if self._replicator_mode: return self._replicator_ready return self._ws_client.is_connected() if self._ws_client else False + def is_cache_ready(self) -> bool: + """Whether flag checks may be evaluated from the cache. + + In replicator mode this is the ``ready`` field of the replicator's + last health response: true once the replicator has finished loading + the cache for its current ``cache_version``, false before that or when + the health poll fails. While it is false, ``AsyncSchematic`` does not + read the cache and answers flag checks from the API. + + Outside replicator mode the SDK fills its own cache over the WebSocket + and fetches what it lacks on demand, so there is nothing to wait for + and this returns true. + """ + if self._replicator_mode: + return self._replicator_ready + return True + def is_replicator_ready(self) -> bool: return self._replicator_ready @@ -1177,42 +1200,49 @@ async def _replicator_health_loop(self) -> None: break async def _check_replicator_health(self) -> None: + """Poll the replicator's health URL and record readiness and cache version. + + The body is read whatever the HTTP status: before the replicator has + loaded the cache it answers 503 with ``ready: false`` and the + ``cache_version`` it is loading. A failed poll (connection error, + timeout, unparseable body) marks the cache not ready and keeps the + last known cache version. + """ if not self._replicator_health_url: return try: if not self._health_check_client: self._health_check_client = httpx.AsyncClient(timeout=REPLICATOR_HEALTH_TIMEOUT_S) resp = await self._health_check_client.get(self._replicator_health_url) - resp.raise_for_status() health_data = resp.json() + if not isinstance(health_data, dict): + raise ValueError(f"unexpected health response body: {health_data!r}") + except Exception as exc: + self._logger.debug("Replicator health check failed: %s", exc) + self._set_replicator_ready(False) + return - self._logger.debug("Replicator health response: %s", health_data) - - was_ready = self._replicator_ready - self._replicator_ready = health_data.get("ready", False) + self._logger.debug("Replicator health response (%s): %s", resp.status_code, health_data) - new_version = health_data.get("cache_version") or health_data.get("cacheVersion") - if new_version and new_version != self._replicator_cache_version: - old = self._replicator_cache_version - self._replicator_cache_version = new_version - self._logger.info("Cache version changed from %s to %s", old, new_version) + new_version = health_data.get("cache_version") or health_data.get("cacheVersion") + if new_version and new_version != self._replicator_cache_version: + old = self._replicator_cache_version + self._replicator_cache_version = new_version + self._logger.info("Cache version changed from %s to %s", old, new_version) - if self._replicator_ready and not was_ready: - self._logger.info("External replicator is now ready") - if self._on_replicator_health_changed: - self._on_replicator_health_changed(True) - elif not self._replicator_ready and was_ready: - self._logger.info("External replicator is no longer ready") - if self._on_replicator_health_changed: - self._on_replicator_health_changed(False) + self._set_replicator_ready(health_data.get("ready") is True) - except Exception as exc: - if self._replicator_ready: - self._replicator_ready = False - self._logger.info("External replicator is no longer ready") - if self._on_replicator_health_changed: - self._on_replicator_health_changed(False) - self._logger.debug("Replicator health check failed: %s", exc) + def _set_replicator_ready(self, ready: bool) -> None: + was_ready = self._replicator_ready + self._replicator_ready = ready + if ready and not was_ready: + self._logger.info("External replicator is now ready") + if self._on_replicator_health_changed: + self._on_replicator_health_changed(True) + elif not ready and was_ready: + self._logger.info("External replicator is no longer ready") + if self._on_replicator_health_changed: + self._on_replicator_health_changed(False) async def get_replicator_cache_version_async(self, timeout_s: float = REPLICATOR_CACHE_VERSION_TIMEOUT_S) -> Optional[str]: """Attempt to fetch cache version immediately if not already available.""" @@ -1224,8 +1254,10 @@ async def get_replicator_cache_version_async(self, timeout_s: float = REPLICATOR if not self._health_check_client: self._health_check_client = httpx.AsyncClient(timeout=timeout_s) resp = await self._health_check_client.get(self._replicator_health_url) - if resp.status_code == 200: - data = resp.json() + # A 503 from a replicator still loading the cache carries the + # cache_version it is loading, so read the body on any status. + data = resp.json() + if isinstance(data, dict): version = data.get("cache_version") or data.get("cacheVersion") if version: self._replicator_cache_version = version diff --git a/tests/custom/test_client.py b/tests/custom/test_client.py index 4e21b734..a6327e37 100644 --- a/tests/custom/test_client.py +++ b/tests/custom/test_client.py @@ -2589,6 +2589,7 @@ def _lease_datastream(results: list, **overrides) -> ScriptedDataStream: **overrides, ) datastream.is_connected = MagicMock(return_value=True) # type: ignore[attr-defined] + datastream.is_cache_ready = MagicMock(return_value=True) # type: ignore[attr-defined] datastream.close = AsyncMock() # type: ignore[attr-defined] datastream.update_company_metrics = AsyncMock() # type: ignore[attr-defined] return datastream diff --git a/tests/custom/test_replicator_cache_ready.py b/tests/custom/test_replicator_cache_ready.py new file mode 100644 index 00000000..91f53f56 --- /dev/null +++ b/tests/custom/test_replicator_cache_ready.py @@ -0,0 +1,353 @@ +"""Replicator mode serves flag checks from the cache only once it is ready. + +The replicator's /ready endpoint answers 200 with ``ready: true`` once the +cache is complete for its ``cache_version`` and 503 with ``ready: false`` +before that. Until the SDK has seen ``ready: true``, check_flag and check_flags +both skip the cache and ask the API; once it has, both evaluate from the cache. +The cache here is seeded the way the replicator writes it: JSON values in +Redis under the versioned keys pinned by tests/datastream/redis_key_layout.json. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any, AsyncIterator, Callable, Dict, List, Optional +from unittest.mock import AsyncMock, MagicMock + +import fakeredis.aioredis +import httpx +import pytest + +from schematic.cache.redis import RedisCache +from schematic.client import REASON_ERROR, AsyncSchematic, AsyncSchematicConfig, DataStreamConfig +from schematic.datastream.datastream_client import DataStreamClient +from schematic.types import CheckFlagResponseData + +HEALTH_URL = "http://replicator.test/ready" +CACHE_VERSION = "v-ready" +FLAG_KEY = "premium" +COMPANY_KEYS = {"id": "acme"} + +# Flag defaults to off; a standard rule turns it on for comp_1 only, so a True +# verdict can only come from evaluating the cached flag against the cached +# company. +FLAG = { + "id": "flag_1", + "key": FLAG_KEY, + "account_id": "acc_1", + "environment_id": "env_1", + "default_value": False, + "rules": [ + { + "id": "rule_1", + "account_id": "acc_1", + "environment_id": "env_1", + "flag_id": "flag_1", + "name": "acme gets premium", + "priority": 1, + "rule_type": "standard", + "value": True, + "condition_groups": [], + "conditions": [ + { + "id": "cond_1", + "account_id": "acc_1", + "environment_id": "env_1", + "condition_type": "company", + "operator": "eq", + "resource_ids": ["comp_1"], + "trait_value": "", + } + ], + } + ], +} + +COMPANY = { + "id": "comp_1", + "account_id": "acc_1", + "environment_id": "env_1", + "keys": COMPANY_KEYS, + "billing_product_ids": [], + "credit_balances": {}, + "metrics": [], + "plan_ids": [], + "plan_version_ids": [], + "rules": [], + "traits": [], +} + + +class HealthEndpoint: + """A scripted replicator /ready endpoint behind an httpx MockTransport.""" + + def __init__(self) -> None: + self.status = 503 + self.body: Any = {"ready": False, "cache_version": CACHE_VERSION} + self.error: Optional[Exception] = None + + def handler(self, request: httpx.Request) -> httpx.Response: + if self.error is not None: + raise self.error + if isinstance(self.body, str): + return httpx.Response(self.status, text=self.body) + return httpx.Response(self.status, json=self.body) + + def client(self) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.MockTransport(self.handler)) + + +async def _seed_like_the_replicator(redis: Any) -> None: + # RedisCache prefixes "schematic:" and JSON-encodes every value, the + # lookup key's company id included. + await redis.set(f"schematic:flags:{CACHE_VERSION}:{FLAG_KEY}", json.dumps(FLAG)) + await redis.set(f"schematic:company:{CACHE_VERSION}:comp_1", json.dumps(COMPANY)) + await redis.set(f"schematic:company:{CACHE_VERSION}:id:acme", json.dumps("comp_1")) + + +def _bulk_response(flags: List[CheckFlagResponseData]) -> MagicMock: + resp = MagicMock() + resp.data = MagicMock(flags=flags) + return resp + + +API_ANSWER = CheckFlagResponseData(flag=FLAG_KEY, value=False, reason="api says no") + + +class ReplicatorHarness: + def __init__(self, client: AsyncSchematic, health: HealthEndpoint) -> None: + self.client = client + self.health = health + + @property + def ds(self) -> DataStreamClient: + ds = self.client._datastream_client + assert ds is not None + return ds + + async def poll(self) -> None: + await self.ds._check_replicator_health() + + def api_answers(self) -> None: + self.client.features.check_flag = AsyncMock(return_value=MagicMock(data=API_ANSWER)) + self.client.features.check_flags = AsyncMock(return_value=_bulk_response([API_ANSWER])) + + def api_fails(self) -> None: + self.client.features.check_flag = AsyncMock(side_effect=RuntimeError("api down")) + self.client.features.check_flags = AsyncMock(side_effect=RuntimeError("api down")) + + +@pytest.fixture +async def harness() -> AsyncIterator[ReplicatorHarness]: + redis = fakeredis.aioredis.FakeRedis() + await _seed_like_the_replicator(redis) + cache: RedisCache[Any] = RedisCache(redis) + client = AsyncSchematic( + "test_key", + AsyncSchematicConfig( + logger=logging.getLogger("test_replicator_cache_ready"), + httpx_client=MagicMock(spec=httpx.AsyncClient), + event_buffer_period=1, + use_datastream=True, + # The flag check cache would answer repeat checks without the API. + cache_providers=[], + flag_defaults={FLAG_KEY: True}, + datastream=DataStreamConfig( + replicator_mode=True, + replicator_health_url=HEALTH_URL, + company_cache=cache, + company_lookup_cache=cache, + user_cache=cache, + user_lookup_cache=cache, + flag_cache=cache, + ), + ), + ) + # Cache-evaluated checks enqueue flag_check events; flushing them to the + # mocked HTTP client at teardown would only retry and time out. + client.event_buffer.push = AsyncMock() # type: ignore[method-assign] + health = HealthEndpoint() + h = ReplicatorHarness(client, health) + h.ds._health_check_client = health.client() + try: + yield h + finally: + await client.event_buffer.stop() + assert h.ds._health_check_client is not None + await h.ds._health_check_client.aclose() + await redis.aclose() + + +async def _single(h: ReplicatorHarness) -> CheckFlagResponseData: + return await h.client.check_flag_with_entitlement(FLAG_KEY, company=COMPANY_KEYS) + + +async def _bulk(h: ReplicatorHarness) -> CheckFlagResponseData: + results = await h.client.check_flags([FLAG_KEY], company=COMPANY_KEYS) + assert len(results) == 1 + return results[0] + + +CHECKS: Dict[str, Callable[[ReplicatorHarness], Any]] = {"single": _single, "bulk": _bulk} + + +@pytest.mark.asyncio +class TestReplicatorNotReady: + @pytest.mark.parametrize("check", CHECKS.values(), ids=CHECKS.keys()) + async def test_skips_the_cache_and_uses_the_api(self, harness: ReplicatorHarness, check: Any) -> None: + await harness.poll() # 503, ready: false + assert not harness.ds.is_cache_ready() + harness.ds.check_flag = AsyncMock(side_effect=AssertionError("cache must not be read")) # type: ignore[method-assign] + harness.api_answers() + + result = await check(harness) + + assert result.value is False + assert result.reason == "api says no" + harness.ds.check_flag.assert_not_called() + + @pytest.mark.parametrize("check", CHECKS.values(), ids=CHECKS.keys()) + async def test_returns_the_flag_default_when_the_api_fails( + self, harness: ReplicatorHarness, check: Any + ) -> None: + await harness.poll() + harness.api_fails() + + result = await check(harness) + + # flag_defaults sets premium to True, so the default is distinguishable + # from the API's False. + assert result.value is True + assert result.reason is not None and result.reason.startswith(REASON_ERROR) + + async def test_before_any_health_response(self, harness: ReplicatorHarness) -> None: + # No poll yet: the SDK has never seen ready: true. + harness.api_answers() + assert (await _single(harness)).reason == "api says no" + assert (await _bulk(harness)).reason == "api says no" + + async def test_readiness_lost_goes_back_to_the_api(self, harness: ReplicatorHarness) -> None: + harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": CACHE_VERSION} + await harness.poll() + harness.health.error = httpx.ConnectError("connection refused") + await harness.poll() + harness.api_answers() + + assert (await _single(harness)).reason == "api says no" + assert (await _bulk(harness)).reason == "api says no" + + +@pytest.mark.asyncio +class TestReplicatorReady: + @pytest.fixture(autouse=True) + async def ready(self, harness: ReplicatorHarness) -> None: + await harness.ds._rules_engine.initialize() + harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": CACHE_VERSION} + await harness.poll() + assert harness.ds.is_cache_ready() + harness.api_fails() + + @pytest.mark.parametrize("check", CHECKS.values(), ids=CHECKS.keys()) + async def test_evaluates_from_the_cache_without_the_api(self, harness: ReplicatorHarness, check: Any) -> None: + result = await check(harness) + + assert result.value is True + assert result.company_id == "comp_1" + assert result.rule_id == "rule_1" + harness.client.features.check_flag.assert_not_called() + harness.client.features.check_flags.assert_not_called() + + async def test_single_and_bulk_agree(self, harness: ReplicatorHarness) -> None: + single = await _single(harness) + bulk = await _bulk(harness) + assert single.model_dump() == bulk.model_dump() + + async def test_a_flag_missing_from_the_cache_still_falls_back_to_the_api(self, harness: ReplicatorHarness) -> None: + harness.api_answers() + missing = CheckFlagResponseData(flag="not-cached", value=True, reason="api match") + harness.client.features.check_flag = AsyncMock(return_value=MagicMock(data=missing)) + harness.client.features.check_flags = AsyncMock(return_value=_bulk_response([missing])) + + single = await harness.client.check_flag_with_entitlement("not-cached", company=COMPANY_KEYS) + bulk = await harness.client.check_flags(["not-cached"], company=COMPANY_KEYS) + + assert single.reason == "api match" + assert bulk[0].reason == "api match" + + +@pytest.mark.asyncio +class TestReplicatorHealthPoll: + async def test_503_sets_not_ready_and_records_the_cache_version(self, harness: ReplicatorHarness) -> None: + harness.health.status, harness.health.body = 503, {"ready": False, "cache_version": "vX"} + await harness.poll() + + assert not harness.ds.is_cache_ready() + assert not harness.ds.is_connected() + assert harness.ds.replicator_cache_version == "vX" + + async def test_200_ready_sets_ready(self, harness: ReplicatorHarness) -> None: + harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": "v2"} + await harness.poll() + + assert harness.ds.is_cache_ready() + # is_connected keeps reporting replicator readiness for compatibility. + assert harness.ds.is_connected() + assert harness.ds.replicator_cache_version == "v2" + + async def test_unreachable_sets_not_ready_and_keeps_the_cache_version(self, harness: ReplicatorHarness) -> None: + harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": "v1"} + await harness.poll() + assert harness.ds.is_cache_ready() + + harness.health.error = httpx.ConnectError("connection refused") + await harness.poll() + + assert not harness.ds.is_cache_ready() + assert harness.ds.replicator_cache_version == "v1" + + async def test_unparseable_body_sets_not_ready_and_keeps_the_cache_version( + self, harness: ReplicatorHarness + ) -> None: + harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": "v1"} + await harness.poll() + + harness.health.status, harness.health.body = 502, "bad gateway" + await harness.poll() + + assert not harness.ds.is_cache_ready() + assert harness.ds.replicator_cache_version == "v1" + + async def test_empty_cache_version_keeps_the_last_one(self, harness: ReplicatorHarness) -> None: + harness.health.status, harness.health.body = 503, {"ready": False, "cache_version": "v1"} + await harness.poll() + harness.health.status, harness.health.body = 503, {"ready": False, "cache_version": ""} + await harness.poll() + + assert harness.ds.replicator_cache_version == "v1" + + async def test_health_changed_callback_fires_on_transitions(self, harness: ReplicatorHarness) -> None: + changes: List[bool] = [] + harness.ds._on_replicator_health_changed = changes.append + + harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": "v1"} + await harness.poll() + await harness.poll() + harness.health.status, harness.health.body = 503, {"ready": False, "cache_version": "v2"} + await harness.poll() + + assert changes == [True, False] + + async def test_cache_version_lookup_reads_a_503_body(self, harness: ReplicatorHarness) -> None: + harness.health.status, harness.health.body = 503, {"ready": False, "cache_version": "v9"} + assert await harness.ds.get_replicator_cache_version_async() == "v9" + + +def test_is_cache_ready_is_true_outside_replicator_mode() -> None: + """Websocket mode fills and fetches its own cache, so it never waits on + readiness; its flag check behavior is unchanged.""" + from schematic.datastream.datastream_client import DataStreamClientOptions + + ds = DataStreamClient(DataStreamClientOptions(api_key="k", logger=logging.getLogger("t"), base_url="ws://x")) + assert ds.is_cache_ready() + assert not ds.is_connected() diff --git a/tests/datastream/test_datastream_client.py b/tests/datastream/test_datastream_client.py index 31ee2683..7baeb00d 100644 --- a/tests/datastream/test_datastream_client.py +++ b/tests/datastream/test_datastream_client.py @@ -1167,7 +1167,11 @@ async def test_health_check_failure_marks_replicator_not_ready( async def test_health_check_http_error_marks_not_ready( self, logger: logging.Logger ) -> None: - """A non-2xx response (raise_for_status) should also mark unhealthy.""" + """A 503 from a replicator still loading the cache marks it not ready. + + Its body is still read: it carries ``ready: false`` and the + cache_version being loaded. + """ cache = MockCacheProvider() client = DataStreamClient(DataStreamClientOptions( api_key="test-key", @@ -1179,7 +1183,8 @@ async def test_health_check_http_error_marks_not_ready( client._replicator_ready = True mock_response = MagicMock() - mock_response.raise_for_status = MagicMock(side_effect=Exception("503 Service Unavailable")) + mock_response.status_code = 503 + mock_response.json = MagicMock(return_value={"ready": False, "cache_version": "v2"}) mock_http = MagicMock() mock_http.get = AsyncMock(return_value=mock_response) @@ -1187,6 +1192,8 @@ async def test_health_check_http_error_marks_not_ready( await client._check_replicator_health() assert not client.is_replicator_ready() + assert not client.is_cache_ready() + assert client.replicator_cache_version == "v2" class TestDataStreamClientDefaultReplicatorHealthUrl: From 1a5411888c696cd9c9eb120f37799727a7899e71 Mon Sep 17 00:00:00 2001 From: ryan echternacht Date: Wed, 30 Sep 2026 11:12:29 -0400 Subject: [PATCH 2/5] Document and test the lease-check readiness gate; prewarm stays ungated Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 2 ++ tests/custom/test_client.py | 17 +++++++++++++++++ 2 files changed, 19 insertions(+) diff --git a/README.md b/README.md index 8c7c38b2..27a464d5 100644 --- a/README.md +++ b/README.md @@ -937,6 +937,8 @@ The SDK serves flag checks from the shared cache only once the replicator report If a health check fails (connection error, timeout, or an unreadable response), the SDK treats the cache as not ready until the next successful check reports `"ready": true`, and keeps using the last `cache_version` it saw. +The same gate applies to `check()` with `usage` in client credit-lease mode: while the cache is not ready, it runs as a plain check through the API instead of gating on a local lease. `prewarm` is not a flag check and still looks up companies through the DataStream without waiting for readiness. + `DataStreamClient.is_cache_ready()` reports this readiness. `is_connected()` returns the same value in Replicator Mode and is kept for backward compatibility. ## Contributing diff --git a/tests/custom/test_client.py b/tests/custom/test_client.py index a6327e37..0c4d8ed9 100644 --- a/tests/custom/test_client.py +++ b/tests/custom/test_client.py @@ -2658,6 +2658,23 @@ async def test_auto_with_datastream_gates_on_a_local_lease(self): finally: await self._drain(client) + async def test_replicator_cache_not_ready_checks_plainly_over_the_api(self): + """The lease check reads the cache, so it shares the flag checks' + readiness gate: while the replicator cache is not ready it runs as a + plain check against the API and takes no lease.""" + client = _async_lease_client() + datastream = _lease_datastream([LEASE_PROBE, LEASE_GATE]) + datastream.is_cache_ready = MagicMock(return_value=False) # type: ignore[attr-defined] + client._datastream_client = datastream + try: + result = await self._check(client) + assert result.allowed is True + assert result.reservation is None + client.features.check_flag.assert_awaited_once() + client.credits.acquire_credit_lease.assert_not_awaited() + finally: + await self._drain(client) + async def test_auto_falls_back_to_server_mode_when_datastream_fails_to_start(self): client = _async_lease_client() client._datastream_client.start = AsyncMock(side_effect=RuntimeError("no socket")) # type: ignore[union-attr] From 9cee479d26abf75f0f494707d8a3591907c4cfc6 Mon Sep 17 00:00:00 2001 From: ryan echternacht Date: Wed, 30 Sep 2026 13:56:03 -0400 Subject: [PATCH 3/5] Update the track test for cache-ready gating The test from #134 expected check_flag to read the cache while the replicator is not ready. Flag checks now skip the cache until it is ready, so the test checks the API path while not ready and the tracked usage from the cache once the replicator reports ready. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/custom/test_replicator_track.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/tests/custom/test_replicator_track.py b/tests/custom/test_replicator_track.py index 3a69cc68..30894c67 100644 --- a/tests/custom/test_replicator_track.py +++ b/tests/custom/test_replicator_track.py @@ -155,7 +155,10 @@ async def test_track_counts_usage_locally_when_replicator_not_ready() -> None: assert ds is not None try: with patch.object(client.event_buffer, "push", new=AsyncMock()) as push: - assert await client.check_flag(FLAG_KEY, company={"id": COMPANY_ID}) is True + # Not ready: the check skips the cache and asks the API, which is + # down, so it returns the flag default. + assert await client.check_flag(FLAG_KEY, company={"id": COMPANY_ID}) is False + client.features.check_flag.assert_awaited() await client.track(EVENT, company={"id": COMPANY_ID}, quantity=10) @@ -168,8 +171,12 @@ async def test_track_counts_usage_locally_when_replicator_not_ready() -> None: assert cached.metrics is not None assert cached.metrics[0].value == 105 - # The cached figure now crosses the limit, so the check denies without - # waiting for the replicator to come back. + # Once the replicator reports ready, the check reads the tracked usage + # from the cache: it now crosses the limit, so it denies without the API. + ds._health_check_client = _health_client({"ready": True, "cache_version": "v1"}) + await ds._check_replicator_health() + assert ds.is_cache_ready() + client.features.check_flag.reset_mock() assert await client.check_flag(FLAG_KEY, company={"id": COMPANY_ID}) is False client.features.check_flag.assert_not_called() finally: From 3b04493e9243d6ea8412e87b410161af20a405ec Mon Sep 17 00:00:00 2001 From: Ben Papillon Date: Wed, 11 Nov 2026 20:24:03 -0800 Subject: [PATCH 4/5] chore(client): correct the track gate comment for cache readiness --- src/schematic/client.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/schematic/client.py b/src/schematic/client.py index ae0a20dc..48123473 100644 --- a/src/schematic/client.py +++ b/src/schematic/client.py @@ -1973,10 +1973,11 @@ async def _update_company_metrics( self, company: Optional[Dict[str, str]], event: str, quantity: Optional[int], ) -> None: # Not gated on is_connected(). Flag checks keep evaluating from the - # cache while a replicator reports not ready or the WebSocket is down, - # so the cached metric has to keep counting the usage tracked in the - # meantime or those checks would enforce limits against a frozen - # figure. The bump cannot double count: the server's next push for the + # cache while the WebSocket is down, and resume from the shared cache + # once a replicator reports ready again, so the cached metric has to + # keep counting the usage tracked in the meantime or those checks + # would enforce limits against a frozen figure. The bump cannot double + # count: the server's next push for the # company replaces the metric outright, and a company missing from the # cache is left alone. ds = self._get_datastream() From ec43059da33f24e47adef20c25b1cc933f3d7785 Mon Sep 17 00:00:00 2001 From: Ben Papillon Date: Thu, 12 Nov 2026 13:40:30 -0800 Subject: [PATCH 5/5] keep the last replicator readiness on a failed health poll A timeout or replicator restart says nothing about whether the shared cache is complete. Flipping every check to the API on one missed poll turns a replicator blip into a burst of API checks from every instance. Only an explicit ready field now changes readiness. --- README.md | 2 +- src/schematic/datastream/datastream_client.py | 30 ++++---- tests/custom/test_replicator_cache_ready.py | 73 +++++++++++++++++-- tests/custom/test_replicator_track.py | 7 +- tests/datastream/test_datastream_client.py | 11 ++- 5 files changed, 93 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index 27a464d5..331c782e 100644 --- a/README.md +++ b/README.md @@ -935,7 +935,7 @@ When running in Replicator Mode, the client will: The SDK serves flag checks from the shared cache only once the replicator reports that the cache is ready. The health endpoint answers `503` with `"ready": false` while the replicator is still loading the cache for its `cache_version`, and `200` with `"ready": true` once that load is complete. Until the SDK has seen `"ready": true`, `check_flag`, `check_flag_with_entitlement` and `check_flags` skip the cache and call the Schematic API, falling back to the flag default if the API call fails. Once the cache is ready, single and bulk checks both evaluate locally from the cache, and a flag that is not in the cache still falls back to the API. -If a health check fails (connection error, timeout, or an unreadable response), the SDK treats the cache as not ready until the next successful check reports `"ready": true`, and keeps using the last `cache_version` it saw. +Readiness changes only when the replicator answers with a `"ready"` field. If a health check fails (connection error, timeout, or an unreadable response), the SDK keeps the readiness and `cache_version` from the last answer it read, so a replicator restart or a missed poll does not send every check to the API. Before the first answer, the cache is not ready. The same gate applies to `check()` with `usage` in client credit-lease mode: while the cache is not ready, it runs as a plain check through the API instead of gating on a local lease. `prewarm` is not a flag check and still looks up companies through the DataStream without waiting for readiness. diff --git a/src/schematic/datastream/datastream_client.py b/src/schematic/datastream/datastream_client.py index dcc72acd..72e0a1d8 100644 --- a/src/schematic/datastream/datastream_client.py +++ b/src/schematic/datastream/datastream_client.py @@ -327,10 +327,10 @@ def is_connected(self) -> bool: def is_cache_ready(self) -> bool: """Whether flag checks may be evaluated from the cache. - In replicator mode this is the ``ready`` field of the replicator's - last health response: true once the replicator has finished loading - the cache for its current ``cache_version``, false before that or when - the health poll fails. While it is false, ``AsyncSchematic`` does not + In replicator mode this is the ``ready`` field of the last health + response the SDK could read: true once the replicator has finished + loading the cache for its current ``cache_version``, false before + that. A failed poll leaves it unchanged. While it is false, ``AsyncSchematic`` does not read the cache and answers flag checks from the API. Outside replicator mode the SDK fills its own cache over the WebSocket @@ -1202,11 +1202,16 @@ async def _replicator_health_loop(self) -> None: async def _check_replicator_health(self) -> None: """Poll the replicator's health URL and record readiness and cache version. - The body is read whatever the HTTP status: before the replicator has - loaded the cache it answers 503 with ``ready: false`` and the - ``cache_version`` it is loading. A failed poll (connection error, - timeout, unparseable body) marks the cache not ready and keeps the - last known cache version. + Readiness changes only on an explicit answer from the replicator: a + parsed body whose ``ready`` field is a boolean, whatever the HTTP + status. Before the replicator has loaded the cache it answers 503 with + ``ready: false`` and the ``cache_version`` it is loading. + + A failed poll (connection error, timeout, unparseable body, or a body + without a boolean ``ready``) changes nothing: it says nothing about + whether the shared cache is complete, and flipping every check to the + API on one missed poll would turn a replicator blip into a burst of + API checks across every SDK instance. """ if not self._replicator_health_url: return @@ -1215,11 +1220,10 @@ async def _check_replicator_health(self) -> None: self._health_check_client = httpx.AsyncClient(timeout=REPLICATOR_HEALTH_TIMEOUT_S) resp = await self._health_check_client.get(self._replicator_health_url) health_data = resp.json() - if not isinstance(health_data, dict): + if not isinstance(health_data, dict) or not isinstance(health_data.get("ready"), bool): raise ValueError(f"unexpected health response body: {health_data!r}") except Exception as exc: - self._logger.debug("Replicator health check failed: %s", exc) - self._set_replicator_ready(False) + self._logger.debug("Replicator health check failed, keeping last readiness: %s", exc) return self._logger.debug("Replicator health response (%s): %s", resp.status_code, health_data) @@ -1230,7 +1234,7 @@ async def _check_replicator_health(self) -> None: self._replicator_cache_version = new_version self._logger.info("Cache version changed from %s to %s", old, new_version) - self._set_replicator_ready(health_data.get("ready") is True) + self._set_replicator_ready(health_data["ready"]) def _set_replicator_ready(self, ready: bool) -> None: was_ready = self._replicator_ready diff --git a/tests/custom/test_replicator_cache_ready.py b/tests/custom/test_replicator_cache_ready.py index 91f53f56..3d4fe1e5 100644 --- a/tests/custom/test_replicator_cache_ready.py +++ b/tests/custom/test_replicator_cache_ready.py @@ -230,7 +230,7 @@ async def test_before_any_health_response(self, harness: ReplicatorHarness) -> N async def test_readiness_lost_goes_back_to_the_api(self, harness: ReplicatorHarness) -> None: harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": CACHE_VERSION} await harness.poll() - harness.health.error = httpx.ConnectError("connection refused") + harness.health.status, harness.health.body = 503, {"ready": False, "cache_version": CACHE_VERSION} await harness.poll() harness.api_answers() @@ -276,6 +276,30 @@ async def test_a_flag_missing_from_the_cache_still_falls_back_to_the_api(self, h assert bulk[0].reason == "api match" +def _unreachable(health: HealthEndpoint) -> None: + health.error = httpx.ConnectError("connection refused") + + +def _timeout(health: HealthEndpoint) -> None: + health.error = httpx.ReadTimeout("timed out") + + +def _html_502(health: HealthEndpoint) -> None: + health.status, health.body = 502, "bad gateway" + + +def _no_ready_field(health: HealthEndpoint) -> None: + health.status, health.body = 200, {"cache_version": "v-garbled"} + + +FAILED_POLLS: Dict[str, Callable[[HealthEndpoint], None]] = { + "unreachable": _unreachable, + "timeout": _timeout, + "unparseable": _html_502, + "no_ready_field": _no_ready_field, +} + + @pytest.mark.asyncio class TestReplicatorHealthPoll: async def test_503_sets_not_ready_and_records_the_cache_version(self, harness: ReplicatorHarness) -> None: @@ -295,29 +319,64 @@ async def test_200_ready_sets_ready(self, harness: ReplicatorHarness) -> None: assert harness.ds.is_connected() assert harness.ds.replicator_cache_version == "v2" - async def test_unreachable_sets_not_ready_and_keeps_the_cache_version(self, harness: ReplicatorHarness) -> None: + async def test_ready_then_explicit_not_ready_sets_not_ready(self, harness: ReplicatorHarness) -> None: harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": "v1"} await harness.poll() assert harness.ds.is_cache_ready() - harness.health.error = httpx.ConnectError("connection refused") + harness.health.status, harness.health.body = 503, {"ready": False, "cache_version": "v2"} await harness.poll() assert not harness.ds.is_cache_ready() + assert harness.ds.replicator_cache_version == "v2" + + @pytest.mark.parametrize("failure", FAILED_POLLS.values(), ids=FAILED_POLLS.keys()) + async def test_ready_then_a_failed_poll_stays_ready(self, harness: ReplicatorHarness, failure: Any) -> None: + harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": "v1"} + await harness.poll() + + failure(harness.health) + await harness.poll() + + assert harness.ds.is_cache_ready() assert harness.ds.replicator_cache_version == "v1" - async def test_unparseable_body_sets_not_ready_and_keeps_the_cache_version( - self, harness: ReplicatorHarness + @pytest.mark.parametrize("failure", FAILED_POLLS.values(), ids=FAILED_POLLS.keys()) + async def test_not_ready_then_a_failed_poll_stays_not_ready( + self, harness: ReplicatorHarness, failure: Any ) -> None: - harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": "v1"} + harness.health.status, harness.health.body = 503, {"ready": False, "cache_version": "v1"} await harness.poll() - harness.health.status, harness.health.body = 502, "bad gateway" + failure(harness.health) await harness.poll() assert not harness.ds.is_cache_ready() assert harness.ds.replicator_cache_version == "v1" + @pytest.mark.parametrize("failure", FAILED_POLLS.values(), ids=FAILED_POLLS.keys()) + async def test_never_answered_then_a_failed_poll_stays_not_ready( + self, harness: ReplicatorHarness, failure: Any + ) -> None: + failure(harness.health) + await harness.poll() + + assert not harness.ds.is_cache_ready() + assert harness.ds.replicator_cache_version is None + + async def test_serves_from_the_cache_through_a_failed_poll(self, harness: ReplicatorHarness) -> None: + await harness.ds._rules_engine.initialize() + harness.health.status, harness.health.body = 200, {"ready": True, "cache_version": CACHE_VERSION} + await harness.poll() + harness.health.error = httpx.ConnectError("connection refused") + await harness.poll() + harness.api_fails() + + assert (await _single(harness)).rule_id == "rule_1" + assert (await _bulk(harness)).rule_id == "rule_1" + harness.client.features.check_flag.assert_not_called() + harness.client.features.check_flags.assert_not_called() + async def test_empty_cache_version_keeps_the_last_one(self, harness: ReplicatorHarness) -> None: harness.health.status, harness.health.body = 503, {"ready": False, "cache_version": "v1"} await harness.poll() diff --git a/tests/custom/test_replicator_track.py b/tests/custom/test_replicator_track.py index 30894c67..8d215d6a 100644 --- a/tests/custom/test_replicator_track.py +++ b/tests/custom/test_replicator_track.py @@ -2,9 +2,10 @@ 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 keep -evaluating from that cache, so track has to keep bumping the cached company -metric, or a numeric limit would never trip while the replicator is down. +stays up and keeps its cache but reports ``ready: false``. Flag checks go to +the API until it reports ready again and then read that cache, so track has to +keep bumping the cached company metric, or those checks would enforce limits +against a figure frozen while the replicator was down. """ from typing import Any diff --git a/tests/datastream/test_datastream_client.py b/tests/datastream/test_datastream_client.py index 7baeb00d..6ce2701b 100644 --- a/tests/datastream/test_datastream_client.py +++ b/tests/datastream/test_datastream_client.py @@ -1135,11 +1135,11 @@ async def test_health_check_invokes_callback_on_state_change( class TestDataStreamClientReplicatorUnhealthyFallback: - """Spec §Replicator Mode: failed health check sets replicator_ready=False - and fires the on_replicator_health_changed(False) callback. + """A failed health check keeps the last readiness; only an explicit + ``ready: false`` marks the replicator not ready. """ - async def test_health_check_failure_marks_replicator_not_ready( + async def test_health_check_failure_keeps_replicator_ready( self, logger: logging.Logger ) -> None: cache = MockCacheProvider() @@ -1152,7 +1152,6 @@ async def test_health_check_failure_marks_replicator_not_ready( user_cache=cache, user_lookup_cache=cache, flag_cache=cache, on_replicator_health_changed=callbacks.append, )) - # Force ready=True so the failure produces a state transition client._replicator_ready = True mock_http = MagicMock() @@ -1161,8 +1160,8 @@ async def test_health_check_failure_marks_replicator_not_ready( await client._check_replicator_health() - assert not client.is_replicator_ready() - assert callbacks == [False] + assert client.is_replicator_ready() + assert callbacks == [] async def test_health_check_http_error_marks_not_ready( self, logger: logging.Logger