Skip to content

feat: Add async FDv2 data sources#485

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jsonbailey merged 6 commits into from
Aug 14, 2026
Merged

feat: Add async FDv2 data sources #485

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
1d9f3df
feat: Add async FDv2 data sources
jsonbailey Aug 12, 2026
565b437
refactor: Share FDv2 polling and streaming error handling between syn…
jsonbailey Aug 13, 2026
6a237d8
fix: Close the async polling transport on stop and make FDv2 test-dat…
jsonbailey Aug 13, 2026
69f2f1a
test: Add async FDv2 polling and streaming coverage from review
jsonbailey Aug 14, 2026
9616eba
fix: Close the async polling transport after the sync loop exits
jsonbailey Aug 14, 2026
161bb0f
fix: Send the FDv2 polling selector as the "basis" query parameter
jsonbailey Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions ldclient/impl/aio/concurrency.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -242,8 +242,6 @@ async def _run(self):
result = self.__action()
if inspect.isawaitable(result):
await result
except asyncio.CancelledError:

@jsonbaileyjsonbaileyAug 13, 2026

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

raise
except Exception as e:
log.exception("Unexpected exception on worker task: %s" % e)
delay = next_time - time.time()
Expand Down
28 changes: 12 additions & 16 deletions ldclient/impl/datasource/async_status.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -3,7 +3,6 @@

from ldclient.impl.dependency_tracker import DependencyTracker, KindAndKey
from ldclient.impl.listeners import Listeners
from ldclient.impl.rwlock import ReadWriteLock

@jsonbaileyjsonbaileyAug 13, 2026

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

from ldclient.interfaces import (
AsyncDataSourceUpdateSink,
AsyncFeatureStore,
Expand All@@ -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
Expand DownExpand Up@@ -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)
Expand Down
Loading
Loading