feat: Add async FDv2 data sources - #485
Conversation
jsonbailey
changed the title
jsonbailey
changed the title
| result = self.__action() | ||
| if inspect.isawaitable(result): | ||
| await result | ||
| except asyncio.CancelledError: |
There was a problem hiding this comment.
There is no need to catch and raise here, it isn't caught by Exception so this was just extra code.
|
|
||
| from ldclient.impl.dependency_tracker import DependencyTracker, KindAndKey | ||
| from ldclient.impl.listeners import Listeners | ||
| from ldclient.impl.rwlock import ReadWriteLock |
There was a problem hiding this comment.
The locks are not always needed in async python unless you are doing multiple awaits between the values you want to keep consistent which we are not doing here. It was over eager and trying to match sync too much.
a07dac5 to
c33ab27
Compare
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit aecca5e. Configure here.
| log.info("Stopping PollingDataSourceV2 synchronizer") | ||
| self._interrupt_event.set() | ||
| self._stop.set() | ||
| await self._requester.close() |
There was a problem hiding this comment.
Stop closes transport mid-request
Medium Severity
stop now closes the requester immediately, which can tear down the aiohttp session while sync is still awaiting fetch. That request then raises, and sync does not catch it, so shutdown can surface as an unexpected exception instead of a clean halt. The existing FDv1 async poller waits for the current poll to finish before closing the transport for this reason.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit aecca5e. Configure here.
aecca5e to
6a237d8
Compare
kinyoklion
left a comment
kinyoklion
left a comment
There was a problem hiding this comment.
Note
This comment is from Claude (a multi-agent review with general, security, and adversarial agents). @rlamb ran the review and posted the results.
Problem: the async tests do not examine the async loop logic
Short description (ASD-STE100):
- The sync sources and the async sources contain the same loop logic in two copies.
- The shared modules (
polling_common.py,streaming_common.py) have good test coverage. The two async loop copies do not. - We did a mutation test. We changed the async loop code in four different ways. All the async tests stayed green each time.
- Thus, a defect in the async loop logic can ship, and no test will show it.
The four mutations that shipped green:
async_polling.py— change the recoverable-error branch (WAIT_CONTINUE) tobreak. One HTTP 500 then stops the poll loop permanently. 15/15 async polling tests pass.async_polling.py— removeawait self._requester.close()fromstop(). The owned HTTP session leaks. 15/15 tests pass.async_streaming.py— remove thefallback_requested orcondition on the Fault path. A latched FDv1 fallback directive no longer stops the stream on a recoverable fault. 21/21 async streaming tests pass.async_streaming.py— remove theenviddefault from theStarthandler. A reconnectStartwithout the header clears the environment ID. 21/21 tests pass.
Solution
The two suggestions below add four tests, one for each mutation. We made sure of this behavior:
- Each test passes on this branch (head
aecca5e; full suites: 40/40 with the additions). - Each test fails fast (< 1 s, no hang) when we apply its mutation.
The polling tests use asyncio.wait_for, so a regression cannot hang CI. Related: several existing tests in these files wait with no time limit (for example, bare await gen.__anext__() with poll_interval=60). We recommend asyncio.wait_for or pytest-timeout for those as well — a hung test stops CI and does not show a failure.
| await asyncio.wait_for(task, timeout=2) | ||
|
|
||
| assert len(updates) == 1 | ||
| assert updates[0].state == DataSourceState.VALID |
There was a problem hiding this comment.
[Claude — suggested tests]
These two tests pin the async-only loop behavior:
test_sync_recoverable_error_continues_polling— makes sure one recoverable error (HTTP 500) does not stop the poll loop. The sync suite has this test; the async suite does not. Kills mutation 1.test_requester_is_closed_after_stop— makes sure the requester closes afterstop(). The test asserts after the generator completes. Thus the test stays correct ifstop()later becomes signal-only with afinallyinsync(). Kills mutation 2.
Both tests use asyncio.wait_for. A regression fails in less than 2 seconds. It does not hang CI.
| assert updates[0].state == DataSourceState.VALID | |
| assert updates[0].state == DataSourceState.VALID | |
| @pytest.mark.asyncio | |
| async def test_sync_recoverable_error_continues_polling(): | |
| change_set = _valid_change_set() | |
| src = _make_source( | |
| [ | |
| _Fail(error="500", exception=UnsuccessfulResponseException(500)), | |
| _Success(value=(change_set, {})), | |
| ], | |
| poll_interval=0.01, | |
| ) | |
| gen = src.sync(_ss()) | |
| first = await asyncio.wait_for(gen.__anext__(), timeout=2) | |
| second = await asyncio.wait_for(gen.__anext__(), timeout=2) | |
| await gen.aclose() | |
| assert first.state == DataSourceState.INTERRUPTED | |
| assert first.error is not None | |
| assert first.error.status_code == 500 | |
| assert second.state == DataSourceState.VALID | |
| assert second.change_set is change_set | |
| @pytest.mark.asyncio | |
| async def test_requester_is_closed_after_stop(): | |
| class CloseTrackingRequester(MockPollingRequester): | |
| def __init__(self, results): | |
| super().__init__(results) | |
| self.closed = False | |
| async def close(self) -> None: | |
| self.closed = True | |
| requester = CloseTrackingRequester([_Success(value=(_valid_change_set(), {}))]) | |
| src = AsyncPollingDataSource(poll_interval=60, requester=requester) | |
| first_update_received = asyncio.Event() | |
| async def consume(): | |
| async for _ in src.sync(_ss()): | |
| first_update_received.set() | |
| task = asyncio.create_task(consume()) | |
| await asyncio.wait_for(first_update_received.wait(), timeout=2) | |
| await src.stop() | |
| await asyncio.wait_for(task, timeout=2) | |
| assert requester.closed is True |
| make_session.assert_not_called() | ||
| assert factory_cls.call_args.kwargs["session"] is supplied | ||
| assert supplied.closed is False | ||
| assert src._owned_session is None |
There was a problem hiding this comment.
[Claude — suggested tests]
These two tests pin the async-only stream-loop behavior:
test_fallback_latched_on_start_carries_through_recoverable_fault— makes sure a latched fallback directive stops the stream when a recoverable fault occurs. The sync suite has three tests for the latch; the async suite has none. Kills mutation 3.test_envid_preserved_across_headerless_reconnect_start— makes sure a reconnectStartwithout theX-LD-EnvIDheader does not clear the environment ID. Kills mutation 4.
The mock action lists are finite, so these tests cannot hang.
| assert src._owned_session is None | |
| assert src._owned_session is None | |
| @pytest.mark.asyncio | |
| async def test_fallback_latched_on_start_carries_through_recoverable_fault(): | |
| """A directive latched on Start must halt the stream even on a recoverable fault.""" | |
| src = make_streaming_data_source() | |
| actions = [ | |
| Start(headers={_LD_FD_FALLBACK_HEADER: 'true'}), | |
| Fault(error=HTTPStatusError(503)), | |
| # Must never be reached — the latched directive halts the stream. | |
| server_intent_event(IntentCode.TRANSFER_FULL), | |
| put_object_event("flag-1"), | |
| payload_transferred_event(), | |
| ] | |
| updates = await collect_updates(src, actions) | |
| assert len(updates) == 1 | |
| assert updates[0].state == DataSourceState.INTERRUPTED | |
| assert updates[0].error.status_code == 503 | |
| assert updates[0].fallback_to_fdv1 is True | |
| @pytest.mark.asyncio | |
| async def test_envid_preserved_across_headerless_reconnect_start(): | |
| """A reconnect Start without X-LD-EnvID must not clear the latched value.""" | |
| src = make_streaming_data_source() | |
| actions = [ | |
| Start(headers={_LD_ENVID_HEADER: 'my-env'}), | |
| server_intent_event(IntentCode.TRANSFER_FULL), | |
| put_object_event("flag-1"), | |
| payload_transferred_event(1), | |
| Start(headers={}), | |
| server_intent_event(IntentCode.TRANSFER_FULL), | |
| put_object_event("flag-2"), | |
| payload_transferred_event(2), | |
| ] | |
| updates = await collect_updates(src, actions) | |
| assert len(updates) == 2 | |
| assert updates[0].environment_id == 'my-env' | |
| assert updates[1].environment_id == 'my-env' |
2 participants
Overview
Part of the async Python SDK work (epic SDK-60). This is the first of two stacked PRs
that add async FDv2 support. It adds the async FDv2 data sources and the shared
support they need. The follow-up PR adds the async FDv2 data system (coordinator)
and wires it into the async client.
This is experimental and not yet wired into a public code path on its own; the async
client only builds an
AsyncFDv2data system in the stacked follow-up PR.What this PR adds
impl/datasourcev2/async_polling.py— async polling data source and its builders.impl/datasourcev2/async_streaming.py— async streaming data source and its builder.impl/integrations/test_datav2/async_test_data_sourcev2.pyplus a newTestDataV2.async_builderproperty for use with the async FDv2 data system.Shared refactors
To avoid duplicating parsing logic between the sync and async sources, the payload
parsing and message handling are extracted into new shared modules that both consume:
polling_common.py—polling_payload_to_changeset/fdv1_polling_payload_to_changeset,moved out of
polling.py. Syncpolling.pynow imports them (and re-exports them).streaming_common.py—process_message, moved out ofStreamingDataSource._process_message.Sync
streaming.pynow calls the shared function.Two small fixes to existing async infrastructure that the async sources depend on:
impl/datasource/async_status.py— drop the read/write lock fromAsyncDataSourceUpdateSinkImpl; the single-threaded event loop does not need it.impl/aio/concurrency.py— stop swallowing/re-raisingCancelledErrorseparately inAsyncRepeatingTask; let cancellation propagate normally.Testing
LD_SKIP_DATABASE_TESTS=1 uv run pytest ldclient/testing/impl/datasourcev2/— 115 passed.ldclient/testing/integrations/test_test_data_sourcev2.py— 34 passed.make lint(mypy, isort, pycodestyle) — clean.Tracked internally: SDK-2869
Note
Overview
Adds async Flag Delivery v2 polling and streaming data sources (
AsyncPollingDataSource,AsyncStreamingDataSource) with aiohttp-based requesters, builders, and FDv1 fallback polling—intended for the stacked async FDv2 data system (not yet on a public client path alone).Shared logic is moved into
polling_common.pyandstreaming_common.py(payload parsing, poll-loop decisions, stream message handling, error classification, FDv1 fallback signaling); syncpolling.pyandstreaming.pynow call those helpers instead of duplicating code.Also adds an async TestDataV2 source (
TestDataV2.async_builder) with thread-safe queue delivery, drops the read/write lock fromAsyncDataSourceUpdateSinkImpl, and adjustsAsyncRepeatingTasksoCancelledErroris not re-raised from the action callback loop.Reviewed by Cursor Bugbot for commit 6a237d8. Bugbot is set up for automated code reviews on this repo. Configure here.