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 experimental async flag-ingestion path affects client initialization and data-source status; shutdown and HTTP error handling must stay correct to avoid leaks or stuck waits.

Overview
Adds experimental async FDv1 polling: an AsyncFeatureRequester contract plus AsyncFeatureRequesterImpl that GETs the poll endpoint with gzip, optional payload filter query param, and ETag / 304 caching before returning flags and segments.

AsyncPollingUpdateProcessor runs polls on AsyncRepeatingTask, writes via sink_or_store, sets ready when the store initializes, and reports VALID / INTERRUPTED / OFF on success, recoverable errors, and fatal HTTP failures (matching sync polling semantics, including unblocking init on unrecoverable errors).

AsyncRepeatingTask.wait_stopped() lets shutdown wait for the in-flight poll to finish; stop() on the processor uses that (with requester.close() in finally) so transports are not closed mid-request.

Broad unit tests cover caching, transport ownership, error recovery, sink status, and shutdown ordering.

^{Reviewed by Cursor Bugbot for commit 305f2b0. Bugbot is set up for automated code reviews on this repo. 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.

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