feat: Add async FDv2 data sources

[jsonbailey]

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 AsyncFDv2 data 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.py plus a new
  TestDataV2.async_builder property 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. Sync polling.py now imports them (and re-exports them).
• streaming_common.py — process_message, moved out of StreamingDataSource._process_message.
  Sync streaming.py now 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 from
  AsyncDataSourceUpdateSinkImpl; the single-threaded event loop does not need it.
• impl/aio/concurrency.py — stop swallowing/re-raising CancelledError separately in
  AsyncRepeatingTask; 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 FDv2 data sources for the experimental async Python SDK: AsyncPollingDataSource (with aiohttp requesters and FDv1 fallback builders) and AsyncStreamingDataSource (SSE via AsyncSSEClient), plus an async TestDataV2 source exposed through TestDataV2.async_builder.

Shared logic is pulled into polling_common.py and streaming_common.py so sync polling.py / streaming.py and the new async modules share payload parsing, map_polling_result, FDv1 fallback signaling, and stream error classification. Sync polling now sends the selector as query param basis (was selector).

Small supporting changes: AsyncDataSourceUpdateSinkImpl drops its read/write lock; AsyncRepeatingTask no longer special-cases CancelledError inside the action loop.

Coverage includes new async polling/streaming tests and updates to polling payload parsing tests after the move to common modules.

^{Reviewed by Cursor Bugbot for commit 161bb0f. Bugbot is set up for automated code reviews on this repo. Configure here.}

[jsonbailey]

There is no need to catch and raise here, it isn't caught by Exception so this was just extra code.

[jsonbailey]

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.

[cursor]

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.}

[kinyoklion]

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:

1. async_polling.py — change the recoverable-error branch (WAIT_CONTINUE) to break. One HTTP 500 then stops the poll loop permanently. 15/15 async polling tests pass.
2. async_polling.py — remove await self._requester.close() from stop(). The owned HTTP session leaks. 15/15 tests pass.
3. async_streaming.py — remove the fallback_requested or condition on the Fault path. A latched FDv1 fallback directive no longer stops the stream on a recoverable fault. 21/21 async streaming tests pass.
4. async_streaming.py — remove the envid default from the Start handler. A reconnect Start without 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.