feat: Add async FDv1 polling data source and feature requester

[jsonbailey]

Overview

PR 7 of the SDK-60 async epic: the async FDv1 polling data source and feature requester.

• async_polling.py — async FDv1 polling update processor. Polls the feature requester on an interval and pushes flag/segment data into the data source update sink, updating data source status (VALID / OFF) as appropriate.
• async_feature_requester.py — async FDv1 feature requester that fetches the full flag/segment payload over HTTP.
• test_async_polling.py — unit tests for the async polling update processor.

Stacking

This PR is stacked on #464 (base branch jb/sdk-2743/async-fdv1-streaming), which provides the shared datasource_common module these files import. Until #464 merges, this PR will also show #464's commits in its diff; a rebase after #464 merges will drop them, leaving only the three files here.

SDK-2825

---

Note

Medium Risk
New async data-source path affects client initialization and flag freshness; shutdown and HTTP error handling are security-adjacent but largely mirror existing sync logic with solid test coverage.

Overview
Adds the async FDv1 polling path for the Python SDK: periodic HTTP fetches of flags/segments, store initialization, and data-source status updates aligned with the sync polling processor.

AsyncFeatureRequesterImpl performs GETs to the FDv1 polling endpoint (optional payload filter query param), uses ETag / 304 caching, and only closes its HTTP transport when it created it.

AsyncPollingUpdateProcessor runs polls on AsyncRepeatingTask, writes via sink_or_store, sets ready when the store is initialized, and reports VALID, INTERRUPTED, or OFF (unrecoverable HTTP errors stop polling and unblock init wait). stop() waits for the repeating task to finish before closing the requester so the transport is not torn down mid-request.

AsyncRepeatingTask.wait_stopped() is new so callers can await background task unwind after stop().

Broad unit tests cover requester caching/errors, processor lifecycle, error recovery, sink status, and shutdown ordering.

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

[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 5053397. Configure here.}

[cursor]

VALID status requires store initialized

Medium Severity

After a successful poll, async polling only emits DataSourceState.VALID when _store.initialized is true. Sync polling and async streaming update to VALID whenever the sink is present after a successful init. With a custom sink that does not mark the shared store initialized, status can stay non-VALID even though data was delivered.

 

^{Reviewed by Cursor Bugbot for commit 5053397. Configure here.}

[kinyoklion]

Note

This is a comment from Claude, an AI tool. @rlamb ran a multi-agent review of this PR and asked Claude to post this finding with a test.

Problem: wait_stopped() does not correctly handle cancellation of its caller

AsyncRepeatingTask.wait_stopped() uses await task (ldclient/impl/aio/concurrency.py, line 232). This statement makes the polling task the _fut_waiter of the caller. If asyncio cancels the caller of stop(), two unwanted effects occur:

1. asyncio sends the cancellation into the polling task. This stops the cleanup of the poll (for example, a store write or the aiohttp connection teardown).
2. The except asyncio.CancelledError: block catches the cancellation of the caller. stop() then continues. It closes the transport while the poll is not complete. It returns with no error.

Effect 2 breaks the guarantee this PR adds. The comment in async_polling.py (lines 49–51) says the transport does not close while a request uses it.

This condition occurs when an application sets a time limit on shutdown. Examples:

• await asyncio.wait_for(client.close(), timeout=5) returns with no TimeoutError. But the poll is not complete, and the transport closed under it.
• A TaskGroup (or a lifespan teardown) cancels the task that runs stop(). The cancellation is lost. The code after await stop() runs.

Suggested fix

Do not await the task directly. Use asyncio.wait:

async def wait_stopped(self):
    """Waits for the task to finish unwinding after ``stop()``. A no-op if
    the task never started or is the current task."""
    task = self.__task
    if task is not None and task is not asyncio.current_task():
        # asyncio.wait does not cancel the task and does not raise the
        # task's exception. Cancellation of the caller propagates normally.
        await asyncio.wait({task})

join_handle() in the same module uses this pattern. Its comment gives the reason.

Tests

The two tests below show the problem. They assert the correct behavior:

• On the current code (ef5803c), the two tests fail: DID NOT RAISE TimeoutError and DID NOT RAISE CancelledError. The result is stable across 5 of 5 runs.
• With the fix above, the two tests pass (5 of 5 runs). The related suites also pass: 55 of 55 tests across this file, test_async_polling.py, and test_aio.py.

Add the tests to ldclient/testing/impl/datasource/ (as a new file, or move the class into test_async_polling.py):

"""
Tests demonstrating that AsyncRepeatingTask.wait_stopped() mishandles
cancellation of its *caller*.

`await task` makes the polling task the awaiting coroutine's `_fut_waiter`, so
cancelling the caller of stop() (an ``asyncio.wait_for`` deadline, an
``asyncio.timeout`` block, a TaskGroup tearing down) has two effects:

1. the cancellation is forwarded *into* the polling task, aborting whatever
   cancellation cleanup it was doing (e.g. a persistent store finishing a
   write, aiohttp connection teardown), and
2. the caller's own cancellation is then absorbed by the blanket
   ``except asyncio.CancelledError``, so stop() keeps going, closes the
   transport out from under the still-unwinding poll, and reports success.

Both tests assert the *desired* behavior, so they FAIL on the current code and
pass once wait_stopped() waits without forwarding cancellation, e.g.::

    async def wait_stopped(self):
        task = self.__task
        if task is not None and task is not asyncio.current_task():
            await asyncio.wait({task})

(the same pattern join_handle() in this module already uses, for the reason
its comment explains).
"""

import asyncio
from unittest.mock import AsyncMock, MagicMock

import pytest

from ldclient.testing.impl.datasource.test_async_polling import make_processor


class TestStopUnderExternalCancellation:
    @pytest.mark.asyncio
    async def test_stop_under_deadline_reports_timeout_and_does_not_abandon_cleanup(self):
        # An application shutting down under a deadline:
        #   await asyncio.wait_for(client.close(), timeout=...)
        # If the in-flight poll's cancellation cleanup outlives the deadline,
        # the caller must see TimeoutError; the cleanup must not be aborted,
        # and the transport must not be closed under the live poll.
        events = []
        started = asyncio.Event()

        async def slow_poll():
            started.set()
            try:
                await asyncio.sleep(60)
            except asyncio.CancelledError:
                events.append('cleanup_started')
                # Simulates a store commit / connection teardown that takes
                # longer than the shutdown deadline below.
                await asyncio.sleep(0.3)
                events.append('cleanup_finished')
                raise

        async def close():
            events.append('transport_closed')

        requester = MagicMock()
        requester.get_all_data = slow_poll
        requester.close = close

        processor = make_processor(requester=requester)
        processor.start()
        await asyncio.wait_for(started.wait(), timeout=1.0)

        # Desired: the missed deadline is reported. Today wait_for() returns
        # normally, because wait_stopped() swallows the CancelledError that
        # wait_for delivers to stop().
        with pytest.raises(asyncio.TimeoutError):
            await asyncio.wait_for(processor.stop(), timeout=0.1)

        # Give the polling task time to finish unwinding on its own.
        await asyncio.sleep(0.4)

        # Desired: the cleanup ran to completion instead of being aborted by
        # the forwarded cancellation...
        assert events[:2] == ['cleanup_started', 'cleanup_finished']
        # ...and the transport was never closed while the poll was live.
        if 'transport_closed' in events:
            assert events.index('transport_closed') > events.index('cleanup_finished')

    @pytest.mark.asyncio
    async def test_cancelling_a_task_blocked_in_stop_actually_cancels_it(self):
        # A TaskGroup sibling failure or lifespan teardown cancels the task
        # that is running stop(). Desired: that task ends cancelled, and the
        # cancellation is not forwarded into the polling task's cleanup.
        cleanup_completed = asyncio.Event()
        started = asyncio.Event()

        async def slow_poll():
            started.set()
            try:
                await asyncio.sleep(60)
            except asyncio.CancelledError:
                await asyncio.sleep(0.2)
                cleanup_completed.set()
                raise

        requester = MagicMock()
        requester.get_all_data = slow_poll
        requester.close = AsyncMock()

        processor = make_processor(requester=requester)
        processor.start()
        await asyncio.wait_for(started.wait(), timeout=1.0)

        shutdown = asyncio.ensure_future(processor.stop())
        await asyncio.sleep(0.05)  # shutdown is now blocked inside wait_stopped()
        shutdown.cancel()

        # Desired: the cancellation propagates. Today stop() swallows it and
        # returns normally, so the application's cancellation is lost.
        with pytest.raises(asyncio.CancelledError):
            await shutdown
        assert shutdown.cancelled()

        # Desired: the polling task's cleanup still ran to completion.
        await asyncio.wait_for(cleanup_completed.wait(), timeout=1.0)

Note: with the fix, stop() can now stop before await self._requester.close() when its caller cancels it. A try/finally around lines 52–53 of async_polling.py makes sure the transport closes in that path too.