feat: Add async FDv2 data sources (#485)

## 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

<!-- CURSOR_SUMMARY -->
---

> [!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.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
161bb0f. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: jsonbailey

ldclient/impl/aio/concurrency.py

@@ -242,8 +242,6 @@ async def _run(self):
                     result = self.__action()
                     if inspect.isawaitable(result):
                         await result
-                except asyncio.CancelledError:
-                    raise
                 except Exception as e:
                     log.exception("Unexpected exception on worker task: %s" % e)
                 delay = next_time - time.time()

ldclient/impl/datasource/async_status.py

@@ -3,7 +3,6 @@
 
 from ldclient.impl.dependency_tracker import DependencyTracker, KindAndKey
 from ldclient.impl.listeners import Listeners
-from ldclient.impl.rwlock import ReadWriteLock
 from ldclient.interfaces import (
     AsyncDataSourceUpdateSink,
     AsyncFeatureStore,
@@ -23,13 +22,11 @@ def __init__(self, store: AsyncFeatureStore, status_listeners: Listeners, flag_c
         self.__flag_change_listeners = flag_change_listeners
         self.__tracker = DependencyTracker()
 
-        self.__lock = ReadWriteLock()
         self.__status = DataSourceStatus(DataSourceState.INITIALIZING, time.time(), None)
 
     @property
     def status(self) -> DataSourceStatus:
-        with self.__lock.read():
-            return self.__status
+        return self.__status
 
     async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None:
         old_data: Optional[Dict[VersionedDataKind, Mapping[str, dict]]] = None
@@ -73,22 +70,21 @@ async def delete(self, kind: VersionedDataKind, key: str, version: int) -> None:
     def update_status(self, new_state: DataSourceState, new_error: Optional[DataSourceErrorInfo]) -> None:
         status_to_broadcast = None
 
-        with self.__lock.write():
-            old_status = self.__status
+        old_status = self.__status
 
-            if new_state == DataSourceState.INTERRUPTED and old_status.state == DataSourceState.INITIALIZING:
-                new_state = DataSourceState.INITIALIZING
+        if new_state == DataSourceState.INTERRUPTED and old_status.state == DataSourceState.INITIALIZING:
+            new_state = DataSourceState.INITIALIZING
 
-            if new_state == old_status.state and new_error is None:
-                return
+        if new_state == old_status.state and new_error is None:
+            return
 
-            self.__status = DataSourceStatus(
-                new_state,
-                self.__status.since if new_state == self.__status.state else time.time(),
-                self.__status.error if new_error is None else new_error,
-            )
+        self.__status = DataSourceStatus(
+            new_state,
+            self.__status.since if new_state == self.__status.state else time.time(),
+            self.__status.error if new_error is None else new_error,
+        )
 
-            status_to_broadcast = self.__status
+        status_to_broadcast = self.__status
 
         if status_to_broadcast is not None:
             self.__status_listeners.notify(status_to_broadcast)