Skip to content

feat: Add async FDv1 streaming and data source status tracking - #464

Open
jsonbailey wants to merge 9 commits into
mainfrom
jb/sdk-2743/async-fdv1-streaming
Open

feat: Add async FDv1 streaming and data source status tracking#464
jsonbailey wants to merge 9 commits into
mainfrom
jb/sdk-2743/async-fdv1-streaming

Conversation

@jsonbailey

@jsonbailey jsonbailey commented Jul 21, 2026

Copy link
Copy Markdown
Contributor

Overview

Adds the async FDv1 streaming data source and its companion async data source status tracking, plus a new shared sans-I/O datasource_common module. This is one slice of the async Python SDK work.

What's included

  • Async FDv1 streaming data source (ldclient/impl/datasource/async_streaming.py) — driven by ld_eventsource's AsyncSSEClient and owning its own aiohttp session via make_client_session. It consumes SSE put/patch/delete events and pushes the resulting data into the async data source update sink.

  • Async data source status tracking (ldclient/impl/datasource/async_status.pyAsyncDataSourceUpdateSinkImpl and friends). The streaming source pushes data and status transitions into this update sink, so the two are runtime-coupled and land together in this slice.

  • New shared sans-I/O datasource_common module (ldclient/impl/datasource/datasource_common.py) — STREAM_ALL_PATH, sink_or_store, and parse_path. Both the sync data sources and the new async streaming source use it.

  • Behavior-preserving extraction of the sync data sources. streaming.py and polling.py are routed through datasource_common: the inline _sink_or_store/_parse_path/STREAM_ALL_PATH/ParsedPath definitions are removed in favor of importing the shared helpers, with call sites switched accordingly. No behavior change. The shared module is also needed by the upcoming async polling slice.

  • Unified status provider (sync/async). The async status provider was unified with the sync one: the redundant AsyncDataSourceStatusProviderImpl (which had no async-specific behavior — byte-identical to the sync provider apart from its sink type) is removed, and the async data system now reuses the sync DataSourceStatusProviderImpl. To let both sinks satisfy it structurally, the sync provider's update_sink parameter was widened to a small status-only Protocol (_DataSourceStatusSource). This is the same interface-segregation pattern as the PR 1b config read-protocols and mirrors the already-unified big-segment status provider. This adds a small behavior-preserving change to sync status.py.

Notes

  • The async public class carries the experimental .. caution:: block.
  • No CHANGELOG entries or version bumps (handled at release time).

Tracked internally: SDK-2743


Note

Medium Risk
Touches core flag-ingestion paths (streaming, store versioning, listener semantics) for the experimental async client; sync changes are mostly refactors but still affect live data-source code.

Overview
Adds async FDv1 streaming (AsyncStreamingUpdateProcessor) that consumes SSE put/patch/delete events, pushes updates through the async data source sink, and lazily creates an SDK-configured aiohttp session when no SSE factory is injected.

Introduces AsyncDataSourceUpdateSinkImpl for async status and flag-change notifications, including dependency tracking and store-error handling. Upserts and deletes only fire flag-change listeners when the async store actually applied the write (new bool return from AsyncFeatureStore.delete, matching upsert).

Extracts shared datasource_common (sink_or_store, parse_path, stream/poll paths) and wires sync polling/streaming (and polling endpoint in feature_requester) through it without intended behavior changes.

Unifies data source status reporting by widening DataSourceStatusProviderImpl to a small _DataSourceStatusSource protocol so the same provider works with sync and async sinks.

Reviewed by Cursor Bugbot for commit c7f3f28. Bugbot is set up for automated code reviews on this repo. Configure here.

@jsonbailey
jsonbailey marked this pull request as ready for review July 27, 2026 20:38
@jsonbailey
jsonbailey requested a review from a team as a code owner July 27, 2026 20:38
Comment thread ldclient/impl/datasource/async_streaming.py
Comment thread ldclient/impl/datasource/async_status.py Outdated
@jsonbailey
jsonbailey force-pushed the jb/sdk-2743/async-fdv1-streaming branch from 924244b to 485382a Compare July 27, 2026 21:32
Comment thread ldclient/impl/datasource/async_status.py Outdated
@jsonbailey
jsonbailey force-pushed the jb/sdk-2743/async-fdv1-streaming branch from 485382a to 73977fd Compare July 27, 2026 22:00
if self._data_source_update_sink is not None:
error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e))

self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Store error kind overwritten

Medium Severity

When the update sink reports STORE_ERROR and re-raises, the streaming processor's broad Exception handler calls update_status again with UNKNOWN. Because a new error is supplied, status is broadcast again and the more specific STORE_ERROR kind is replaced.

Additional Locations (1)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 73977fd. Configure here.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

This mirrors the sync data source status sink rather than being an async-only issue: sync's __monitor_store_update also re-raises after setting STORE_ERROR, and the sync streaming run() has the same broad except Exception that then calls update_status(INTERRUPTED, UNKNOWN) and overwrites the more specific kind. Keeping async in parity with sync here; if the overwrite is worth correcting it should be fixed in both sync and async as a separate change. Tracked as a follow-up.

Restore an async __monitor_store_update helper mirroring the sync sink,
so the prior-data read during init() is monitored alongside store.init.
A failure in that read now records STORE_ERROR and moves the status to
INTERRUPTED, matching the sync data source status sink.
@jsonbailey
jsonbailey force-pushed the jb/sdk-2743/async-fdv1-streaming branch from 4149c47 to 89de1c5 Compare July 29, 2026 19:31
Comment thread ldclient/impl/datasource/async_streaming.py Outdated
Its stop() is a coroutine, so it must implement the async update processor
interface rather than the sync UpdateProcessor (whose stop is synchronous).
Comment thread ldclient/impl/datasource/async_streaming.py Outdated
Share the FDv1 polling endpoint from the sans-I/O datasource_common module
(next to STREAM_ALL_PATH) so the sync and async feature requesters no longer
each define their own copy.
Defer creating the owned aiohttp session + SSE factory from __init__ to _run(),
so the ClientSession is created on the running event loop (aiohttp's
expectation) rather than at construction time. Behavior is unchanged when a
factory is injected.

@cursor cursor Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

Fix All in Cursor

❌ 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 c7f3f28. Configure here.

if not await self._handle_error(action.error):
break
await self._sse.close()
await self._close_owned_session()

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Owned session can leak on errors

Medium Severity

_run creates an owned aiohttp session but only closes it after a normal loop exit or via stop()/__stop_with_error_info. If session creation succeeds and a later step fails—create(), the SSE iterator, or interrupt() in an except handler—the exception escapes without a finally, so the session is never closed and connections can leak until process exit.

Additional Locations (1)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit c7f3f28. Configure here.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant