From ec7746abe0e65ca457ca3fdd7d0c04535acbd109 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 27 Jul 2026 16:48:46 -0500 Subject: [PATCH 1/8] feat: Add async migrations --- ldclient/migrations/__init__.py | 7 + ldclient/migrations/async_migrator.py | 362 ++++++++++ ldclient/migrations/migrator.py | 46 +- ldclient/migrations/types.py | 98 ++- .../testing/migrations/test_async_migrator.py | 629 ++++++++++++++++++ 5 files changed, 1083 insertions(+), 59 deletions(-) create mode 100644 ldclient/migrations/async_migrator.py create mode 100644 ldclient/testing/migrations/test_async_migrator.py diff --git a/ldclient/migrations/__init__.py b/ldclient/migrations/__init__.py index 53a6cec0..2e1269b3 100644 --- a/ldclient/migrations/__init__.py +++ b/ldclient/migrations/__init__.py @@ -1,8 +1,15 @@ +# async_migrator is import-cheap (asyncio stdlib only, no aiohttp), so it is +# exported eagerly alongside the sync surface and keeps `import ldclient` cheap. +from .async_migrator import * from .migrator import * from .tracker import * from .types import * __all__ = [ + 'AsyncMigrationConfig', + 'AsyncMigrator', + 'AsyncMigratorBuilder', + 'AsyncMigratorFn', 'Migrator', 'MigratorBuilder', 'MigratorCompareFn', diff --git a/ldclient/migrations/async_migrator.py b/ldclient/migrations/async_migrator.py new file mode 100644 index 00000000..f1525651 --- /dev/null +++ b/ldclient/migrations/async_migrator.py @@ -0,0 +1,362 @@ +from __future__ import annotations + +import asyncio +from abc import ABC, abstractmethod +from datetime import datetime +from random import Random +from typing import ( + TYPE_CHECKING, + Any, + Awaitable, + Callable, + Optional, + Tuple, + Union +) + +from ldclient.impl.sampler import Sampler +from ldclient.impl.util import Result +from ldclient.migrations.tracker import OpTracker +from ldclient.migrations.types import ( + ExecutionOrder, + MigrationConfig, + MigratorCompareFn, + Operation, + OperationResult, + Origin, + Stage, + WriteResult, + _MigrationConfigBase, + _MigratorBuilderBase +) + +if TYPE_CHECKING: + from ldclient import Context + from ldclient.async_client import AsyncLDClient + +__all__ = [ + 'AsyncMigrator', + 'AsyncMigratorBuilder', + 'AsyncMigratorImpl', + 'AsyncMigrationConfig', + 'AsyncExecutor', + 'AsyncMigratorFn', +] + +AsyncMigratorFn = Callable[[Optional[Any]], Awaitable[Any]] +""" +The async counterpart to :data:`ldclient.migrations.MigratorFn`. When an async +migration wishes to execute a read or write operation, it must delegate that +call to a consumer defined coroutine function. This function must accept an +optional payload value, and return a :class:`ldclient.Result`. +""" + + +class AsyncMigrator(ABC): + """ + An async migrator is the interface through which migration support is + executed for the async SDK. An async migrator is configured through the + :class:`AsyncMigratorBuilder`. + + .. caution:: + This feature is experimental and should NOT be considered ready for production + use. It may change or be removed without notice and is not subject to backwards + compatibility guarantees. Pin to a specific minor version and review the changelog + before upgrading. + """ + + @abstractmethod + async def read(self, key: str, context: Context, default_stage: Stage, payload: Optional[Any] = None) -> OperationResult: + """ + Uses the provided flag key and context to execute a migration-backed read operation. + + :param key: The migration flag key to use when determining the current stage + :param context: The context to use when evaluating the flag + :param default_stage: A default stage to fallback to if one cannot be determined + :param payload: An optional payload to be passed through to the appropriate read method + """ + + @abstractmethod + async def write(self, key: str, context: Context, default_stage: Stage, payload: Optional[Any] = None) -> WriteResult: + """ + Uses the provided flag key and context to execute a migration-backed write operation. + + :param key: The migration flag key to use when determining the current stage + :param context: The context to use when evaluating the flag + :param default_stage: A default stage to fallback to if one cannot be determined + :param payload: An optional payload to be passed through to the appropriate write method + """ + + +class AsyncMigratorImpl(AsyncMigrator): + """ + An implementation of the :class:`ldclient.migrations.AsyncMigrator` + interface, capable of supporting feature-flag backed technology migrations + for the async SDK. + """ + + def __init__( + self, + sampler: Sampler, + client: AsyncLDClient, + read_execution_order: ExecutionOrder, + read_config: AsyncMigrationConfig, + write_config: AsyncMigrationConfig, + measure_latency: bool, + measure_errors: bool, + ): + self._sampler = sampler + self._client = client + self._read_execution_order = read_execution_order + self._read_config = read_config + self._write_config = write_config + self._measure_latency = measure_latency + self._measure_errors = measure_errors + + async def read(self, key: str, context: Context, default_stage: Stage, payload: Optional[Any] = None) -> OperationResult: + stage, tracker = await self._client.migration_variation(key, context, default_stage) + tracker.operation(Operation.READ) + + old = AsyncExecutor(Origin.OLD, self._read_config.old, tracker, self._measure_latency, self._measure_errors, payload) + new = AsyncExecutor(Origin.NEW, self._read_config.new, tracker, self._measure_latency, self._measure_errors, payload) + + if stage == Stage.OFF: + result = await old.run() + elif stage == Stage.DUALWRITE: + result = await old.run() + elif stage == Stage.SHADOW: + result = await self.__read_both(old, new, tracker) + elif stage == Stage.LIVE: + result = await self.__read_both(new, old, tracker) + elif stage == Stage.RAMPDOWN: + result = await new.run() + else: + result = await new.run() + + # track_migration_op is synchronous on the async client; do not await it. + self._client.track_migration_op(tracker) + + return result + + async def write(self, key: str, context: Context, default_stage: Stage, payload: Optional[Any] = None) -> WriteResult: + stage, tracker = await self._client.migration_variation(key, context, default_stage) + tracker.operation(Operation.WRITE) + + old = AsyncExecutor(Origin.OLD, self._write_config.old, tracker, self._measure_latency, self._measure_errors, payload) + new = AsyncExecutor(Origin.NEW, self._write_config.new, tracker, self._measure_latency, self._measure_errors, payload) + + if stage == Stage.OFF: + result = await old.run() + write_result = WriteResult(result) + elif stage == Stage.DUALWRITE: + authoritative_result, nonauthoritative_result = await self.__write_both(old, new, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + elif stage == Stage.SHADOW: + authoritative_result, nonauthoritative_result = await self.__write_both(old, new, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + elif stage == Stage.LIVE: + authoritative_result, nonauthoritative_result = await self.__write_both(new, old, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + elif stage == Stage.RAMPDOWN: + authoritative_result, nonauthoritative_result = await self.__write_both(new, old, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + else: + result = await new.run() + write_result = WriteResult(result) + + # track_migration_op is synchronous on the async client; do not await it. + self._client.track_migration_op(tracker) + + return write_result + + async def __read_both(self, authoritative: AsyncExecutor, nonauthoritative: AsyncExecutor, tracker: OpTracker) -> OperationResult: + if self._read_execution_order == ExecutionOrder.PARALLEL: + authoritative_result, nonauthoritative_result = await asyncio.gather( + authoritative.run(), + nonauthoritative.run(), + ) + elif self._read_execution_order == ExecutionOrder.RANDOM and self._sampler.sample(2): + nonauthoritative_result = await nonauthoritative.run() + authoritative_result = await authoritative.run() + else: + authoritative_result = await authoritative.run() + nonauthoritative_result = await nonauthoritative.run() + + if self._read_config.comparison is None: + return authoritative_result + + compare = self._read_config.comparison + if authoritative_result.is_success() and nonauthoritative_result.is_success(): + tracker.consistent(lambda: compare(authoritative_result.value, nonauthoritative_result.value)) + + return authoritative_result + + async def __write_both(self, authoritative: AsyncExecutor, nonauthoritative: AsyncExecutor, tracker: OpTracker) -> Tuple[OperationResult, Optional[OperationResult]]: + authoritative_result = await authoritative.run() + tracker.invoked(authoritative.origin) + + if not authoritative_result.is_success(): + return authoritative_result, None + + nonauthoritative_result = await nonauthoritative.run() + tracker.invoked(nonauthoritative.origin) + + return authoritative_result, nonauthoritative_result + + +class AsyncMigrationConfig(_MigrationConfigBase[AsyncMigratorFn]): + """ + The async counterpart to :class:`ldclient.migrations.MigrationConfig`. It + stores references to coroutine functions which execute customer defined + read or write operations on old or new origins of information. For read + operations, an optional (synchronous) comparison function can also be + defined. + + .. caution:: + This feature is experimental and should NOT be considered ready for production + use. It may change or be removed without notice and is not subject to backwards + compatibility guarantees. + """ + + +class AsyncMigratorBuilder(_MigratorBuilderBase): + """ + The async migration builder is used to configure and construct an instance + of an :class:`AsyncMigrator`. This migrator can be used to perform + LaunchDarkly assisted technology migrations through the use of + migration-based feature flags. + + .. caution:: + This feature is experimental and should NOT be considered ready for production + use. It may change or be removed without notice and is not subject to backwards + compatibility guarantees. Pin to a specific minor version and review the changelog + before upgrading. + """ + + def __init__(self, client: AsyncLDClient): + # Single _ to prevent mangling; useful for testing + self._client = client + + # Default settings as required by the spec + self._read_execution_order = ExecutionOrder.PARALLEL + self._measure_latency = True + self._measure_errors = True + + self.__read_config: Optional[AsyncMigrationConfig] = None + self.__write_config: Optional[AsyncMigrationConfig] = None + + def read(self, old: AsyncMigratorFn, new: AsyncMigratorFn, comparison: Optional[MigratorCompareFn] = None) -> 'AsyncMigratorBuilder': + """ + Read can be used to configure the migration-read behavior of the + resulting :class:`AsyncMigrator` instance. + + Users are required to provide two different read coroutine functions -- + one to read from the old migration origin, and one to read from the new + origin. Additionally, customers can opt-in to consistency tracking by + providing a comparison function. + + Depending on the migration stage, one or both of these read methods may + be called. + + The read methods should accept a single nullable parameter. This + parameter is a payload passed through the :func:`AsyncMigrator.read` + method. This method should return a :class:`ldclient.Result` instance. + + The consistency method should accept 2 parameters of any type. These + parameters are the results of executing the read operation against the + old and new origins. If both operations were successful, the + consistency method will be invoked. This method should return true if + the two parameters are equal, or false otherwise. The comparison + function is synchronous. + + :param old: The coroutine function to execute when reading from the old origin + :param new: The coroutine function to execute when reading from the new origin + :param comparison: An optional function to use for comparing the results from two origins + """ + self.__read_config = AsyncMigrationConfig(old, new, comparison) + return self + + def write(self, old: AsyncMigratorFn, new: AsyncMigratorFn) -> 'AsyncMigratorBuilder': + """ + Write can be used to configure the migration-write behavior of the + resulting :class:`AsyncMigrator` instance. + + Users are required to provide two different write coroutine functions -- + one to write to the old migration origin, and one to write to the new + origin. + + Depending on the migration stage, one or both of these write methods + may be called. + + The write methods should accept a single nullable parameter. This + parameter is a payload passed through the :func:`AsyncMigrator.write` + method. This method should return a :class:`ldclient.Result` instance. + + :param old: The coroutine function to execute when writing to the old origin + :param new: The coroutine function to execute when writing to the new origin + """ + self.__write_config = AsyncMigrationConfig(old, new) + return self + + def build(self) -> Union[AsyncMigrator, str]: + """ + Build constructs an :class:`AsyncMigrator` instance to support + migration-based reads and writes. A string describing any failure + conditions will be returned if the build fails. + """ + if self.__read_config is None: + return "read configuration not provided" + + if self.__write_config is None: + return "write configuration not provided" + + return AsyncMigratorImpl( + Sampler(Random()), + self._client, + self._read_execution_order, + self.__read_config, + self.__write_config, + self._measure_latency, + self._measure_errors, + ) + + +class AsyncExecutor: + """ + Utility class for executing async migration operations while also tracking + our built-in migration measurements. + """ + + def __init__(self, origin: Origin, fn: AsyncMigratorFn, tracker: OpTracker, measure_latency: bool, measure_errors: bool, payload: Any): + self.__origin = origin + self.__fn = fn + self.__tracker = tracker + self.__measure_latency = measure_latency + self.__measure_errors = measure_errors + self.__payload = payload + + @property + def origin(self) -> Origin: + return self.__origin + + async def run(self) -> OperationResult: + """ + Execute the configured operation and track any available measurements. + """ + start = datetime.now() + + try: + result = await self.__fn(self.__payload) + except Exception as e: + result = Result.fail(f"'{self.__origin.value} operation raised an exception", e) + + # Record required tracker measurements + if self.__measure_latency: + self.__tracker.latency(self.__origin, datetime.now() - start) + + if self.__measure_errors and not result.is_success(): + self.__tracker.error(self.__origin) + + self.__tracker.invoked(self.__origin) + + return OperationResult(self.__origin, result) diff --git a/ldclient/migrations/migrator.py b/ldclient/migrations/migrator.py index fde023f0..1b23e464 100644 --- a/ldclient/migrations/migrator.py +++ b/ldclient/migrations/migrator.py @@ -18,7 +18,8 @@ OperationResult, Origin, Stage, - WriteResult + WriteResult, + _MigratorBuilderBase ) if TYPE_CHECKING: @@ -168,7 +169,7 @@ def __write_both(self, authoritative: Executor, nonauthoritative: Executor, trac return authoritative_result, nonauthoritative_result -class MigratorBuilder: +class MigratorBuilder(_MigratorBuilderBase): """ The migration builder is used to configure and construct an instance of a :class:`Migrator`. This migrator can be used to perform LaunchDarkly @@ -181,42 +182,13 @@ def __init__(self, client: LDClient): self._client = client # Default settings as required by the spec - self.__read_execution_order = ExecutionOrder.PARALLEL - self.__measure_latency = True - self.__measure_errors = True + self._read_execution_order = ExecutionOrder.PARALLEL + self._measure_latency = True + self._measure_errors = True self.__read_config: Optional[MigrationConfig] = None self.__write_config: Optional[MigrationConfig] = None - def read_execution_order(self, order: ExecutionOrder) -> 'MigratorBuilder': - """ - The read execution order influences the parallelism and execution order - for read operations involving multiple origins. - """ - if order not in ExecutionOrder: - return self - - self.__read_execution_order = order - return self - - def track_latency(self, enabled: bool) -> 'MigratorBuilder': - """ - Enable or disable latency tracking for migration operations. This - latency information can be sent upstream to LaunchDarkly to enhance - migration visibility. - """ - self.__measure_latency = enabled - return self - - def track_errors(self, enabled: bool) -> 'MigratorBuilder': - """ - Enable or disable error tracking for migration operations. This error - information can be sent upstream to LaunchDarkly to enhance migration - visibility. - """ - self.__measure_errors = enabled - return self - def read(self, old: MigratorFn, new: MigratorFn, comparison: Optional[MigratorCompareFn] = None) -> 'MigratorBuilder': """ Read can be used to configure the migration-read behavior of the @@ -283,11 +255,11 @@ def build(self) -> Union[Migrator, str]: return MigratorImpl( Sampler(Random()), self._client, - self.__read_execution_order, + self._read_execution_order, self.__read_config, self.__write_config, - self.__measure_latency, - self.__measure_errors, + self._measure_latency, + self._measure_errors, ) diff --git a/ldclient/migrations/types.py b/ldclient/migrations/types.py index 295731ea..54b2ce7c 100644 --- a/ldclient/migrations/types.py +++ b/ldclient/migrations/types.py @@ -1,5 +1,5 @@ from enum import Enum -from typing import Any, Callable, Optional +from typing import Any, Callable, Generic, Optional, TypeVar from ldclient.impl.util import Result @@ -196,41 +196,41 @@ def nonauthoritative(self) -> Optional[OperationResult]: return self.__nonauthoritative -class MigrationConfig: +_MigratorFnT = TypeVar('_MigratorFnT') + + +class _MigrationConfigBase(Generic[_MigratorFnT]): """ - A migration config stores references to callable methods which execute - customer defined read or write operations on old or new origins of - information. For read operations, an optional comparison function also be - defined. + Shared implementation backing :class:`MigrationConfig` and its async + counterpart. It stores references to the customer defined read or write + functions for the old and new origins, along with an optional synchronous + comparison function used for read consistency tracking. """ - def __init__(self, old: MigratorFn, new: MigratorFn, comparison: Optional[MigratorCompareFn] = None): - self.__old = old - self.__new = new - self.__comparison = comparison + def __init__(self, old: _MigratorFnT, new: _MigratorFnT, comparison: Optional[MigratorCompareFn] = None): + self._old = old + self._new = new + self._comparison = comparison @property - def old(self) -> MigratorFn: + def old(self) -> _MigratorFnT: """ Callable which receives a nullable payload parameter and returns an :class:`ldclient.Result`. This function call should affect the old migration origin when called. - - @return [#call] """ - return self.__old + return self._old @property - def new(self) -> MigratorFn: + def new(self) -> _MigratorFnT: """ - # Callable which receives a nullable payload parameter and returns an - # :class:`ldclient.Result`. - # - # This function call should affect the new migration origin when - # called. + Callable which receives a nullable payload parameter and returns an + :class:`ldclient.Result`. + + This function call should affect the new migration origin when called. """ - return self.__new + return self._new @property def comparison(self) -> Optional[MigratorCompareFn]: @@ -241,4 +241,58 @@ def comparison(self) -> Optional[MigratorCompareFn]: The result of this comparison can be sent upstream to LaunchDarkly to enhance migration observability. """ - return self.__comparison + return self._comparison + + +class MigrationConfig(_MigrationConfigBase[MigratorFn]): + """ + A migration config stores references to callable methods which execute + customer defined read or write operations on old or new origins of + information. For read operations, an optional comparison function also be + defined. + """ + + +_MigratorBuilderT = TypeVar('_MigratorBuilderT', bound='_MigratorBuilderBase') + + +class _MigratorBuilderBase: + """ + Shared setter implementation for :class:`MigratorBuilder` and its async + counterpart. It holds the fluent configuration methods common to both + builders. The ``read``, ``write``, and ``build`` methods differ between the + sync and async builders and are defined on each subclass. + """ + + _read_execution_order: ExecutionOrder + _measure_latency: bool + _measure_errors: bool + + def read_execution_order(self: _MigratorBuilderT, order: ExecutionOrder) -> _MigratorBuilderT: + """ + The read execution order influences the parallelism and execution order + for read operations involving multiple origins. + """ + if order not in ExecutionOrder: + return self + + self._read_execution_order = order + return self + + def track_latency(self: _MigratorBuilderT, enabled: bool) -> _MigratorBuilderT: + """ + Enable or disable latency tracking for migration operations. This + latency information can be sent upstream to LaunchDarkly to enhance + migration visibility. + """ + self._measure_latency = enabled + return self + + def track_errors(self: _MigratorBuilderT, enabled: bool) -> _MigratorBuilderT: + """ + Enable or disable error tracking for migration operations. This error + information can be sent upstream to LaunchDarkly to enhance migration + visibility. + """ + self._measure_errors = enabled + return self diff --git a/ldclient/testing/migrations/test_async_migrator.py b/ldclient/testing/migrations/test_async_migrator.py new file mode 100644 index 00000000..9fcbfb44 --- /dev/null +++ b/ldclient/testing/migrations/test_async_migrator.py @@ -0,0 +1,629 @@ +import asyncio +from datetime import datetime, timedelta +from typing import List + +import pytest + +from ldclient import Result +from ldclient.context import Context +from ldclient.evaluation import EvaluationDetail +from ldclient.impl.events.types import EventInputEvaluation +from ldclient.impl.model import FeatureFlag +from ldclient.impl.util import timedelta_millis +from ldclient.migrations import ( + AsyncMigrator, + AsyncMigratorBuilder, + AsyncMigratorFn +) +from ldclient.migrations.tracker import MigrationOpEvent, OpTracker +from ldclient.migrations.types import ExecutionOrder, Origin, Stage +from ldclient.testing.builders import FlagBuilder + +user = Context.from_dict({u'key': u'xyz', u'kind': u'user', u'bizzle': u'def'}) + + +# --------------------------------------------------------------------------- +# Test doubles +# --------------------------------------------------------------------------- + +class FakeEventProcessor: + """Records events the way the real (async/sync) event processors expose them + so tests can assert on the emitted MigrationOpEvent, mirroring the sync + migrator tests' use of ``client._event_processor._events``.""" + + def __init__(self): + self._events: List = [] + + def send_event(self, event): + self._events.append(event) + + +class FakeAsyncClient: + """A minimal stand-in for AsyncLDClient that exposes the exact surface the + AsyncMigrator depends on: an async ``migration_variation`` returning + ``(Stage, OpTracker)`` and a synchronous ``track_migration_op``. + + The flag key passed to ``migration_variation`` is interpreted as a Stage + value (matching the sync migrator tests), and a real OpTracker is built from + a real FeatureFlag so the consistency/error/latency/event-build semantics + are genuinely exercised. + """ + + def __init__(self): + self._event_processor = FakeEventProcessor() + self._flags = {} + for stage in Stage: + flag = FlagBuilder(stage.value).on(True).variations(stage.value).fallthrough_variation(0).build() + self._flags[stage.value] = flag + + async def migration_variation(self, key: str, context: Context, default_stage: Stage): + # Yield once to confirm callers truly await this coroutine. + await asyncio.sleep(0) + flag: FeatureFlag = self._flags[key] + stage = Stage(key) + detail = EvaluationDetail(stage.value, 0, {'kind': 'FALLTHROUGH'}) + tracker = OpTracker(key, flag, context, detail, default_stage) + return stage, tracker + + def track_migration_op(self, tracker: OpTracker): + # Synchronous on the real async client; must NOT be awaited. + event = tracker.build() + if isinstance(event, str): + raise AssertionError("tracker.build() failed: %s" % event) + # Emulate the EventInputEvaluation that migration_variation would have + # queued, so index [1] is the MigrationOpEvent (as in the sync tests). + self._event_processor.send_event(_FakeEvalEvent()) + self._event_processor.send_event(event) + + +class _FakeEvalEvent(EventInputEvaluation): + def __init__(self): + pass + + +async def async_success(payload) -> Result: + return Result.success(True) + + +def raises_exception(msg) -> AsyncMigratorFn: + async def inner(payload): + raise Exception(msg) + + return inner + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def builder() -> AsyncMigratorBuilder: + client = FakeAsyncClient() + builder = AsyncMigratorBuilder(client) # type: ignore[arg-type] + builder.track_latency(False) + builder.track_errors(False) + + builder.read(async_success, async_success, None) + builder.write(async_success, async_success) + + return builder + + +# --------------------------------------------------------------------------- +# Builder validation +# --------------------------------------------------------------------------- + +class TestBuilder: + def test_can_build_successfully(self): + client = FakeAsyncClient() + builder = AsyncMigratorBuilder(client) # type: ignore[arg-type] + builder.read(async_success, async_success, None) + builder.write(async_success, async_success) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + @pytest.mark.parametrize( + "order", + [ + pytest.param(ExecutionOrder.SERIAL, id="serial"), + pytest.param(ExecutionOrder.RANDOM, id="random"), + pytest.param(ExecutionOrder.PARALLEL, id="parallel"), + ], + ) + def test_can_modify_execution_order(self, order): + client = FakeAsyncClient() + builder = AsyncMigratorBuilder(client) # type: ignore[arg-type] + builder.read(async_success, async_success, None) + builder.write(async_success, async_success) + builder.read_execution_order(order) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + def test_build_fails_without_read(self): + client = FakeAsyncClient() + builder = AsyncMigratorBuilder(client) # type: ignore[arg-type] + builder.write(async_success, async_success) + migrator = builder.build() + assert isinstance(migrator, str) + assert migrator == "read configuration not provided" + + def test_build_fails_without_write(self): + client = FakeAsyncClient() + builder = AsyncMigratorBuilder(client) # type: ignore[arg-type] + builder.read(async_success, async_success) + migrator = builder.build() + assert isinstance(migrator, str) + assert migrator == "write configuration not provided" + + +# --------------------------------------------------------------------------- +# Payload passthrough +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +class TestPassingPayloadThrough: + @pytest.mark.parametrize( + "stage,count", + [ + pytest.param(Stage.OFF, 1, id="off"), + pytest.param(Stage.DUALWRITE, 1, id="dualwrite"), + pytest.param(Stage.SHADOW, 2, id="shadow"), + pytest.param(Stage.LIVE, 2, id="live"), + pytest.param(Stage.RAMPDOWN, 1, id="rampdown"), + pytest.param(Stage.COMPLETE, 1, id="complete"), + ], + ) + async def test_passes_through_read(self, builder: AsyncMigratorBuilder, stage: Stage, count: int): + payloads = [] + + async def capture_payloads(payload): + payloads.append(payload) + return Result.success(None) + + builder.read(capture_payloads, capture_payloads) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE, "payload") + + assert result.is_success() + assert len(payloads) == count + assert all("payload" == p for p in payloads) + + @pytest.mark.parametrize( + "stage,count", + [ + pytest.param(Stage.OFF, 1, id="off"), + pytest.param(Stage.DUALWRITE, 2, id="dualwrite"), + pytest.param(Stage.SHADOW, 2, id="shadow"), + pytest.param(Stage.LIVE, 2, id="live"), + pytest.param(Stage.RAMPDOWN, 2, id="rampdown"), + pytest.param(Stage.COMPLETE, 1, id="complete"), + ], + ) + async def test_passes_through_write(self, builder: AsyncMigratorBuilder, stage: Stage, count: int): + payloads = [] + + async def capture_payloads(payload): + payloads.append(payload) + return Result.success(None) + + builder.write(capture_payloads, capture_payloads) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE, "payload") + + assert result.authoritative.is_success() + if result.nonauthoritative is not None: + assert result.nonauthoritative.is_success() + + assert len(payloads) == count + assert all("payload" == p for p in payloads) + + +# --------------------------------------------------------------------------- +# Invoked tracking +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +class TestTrackingInvoked: + @pytest.mark.parametrize( + "stage,origins", + [ + pytest.param(Stage.OFF, [Origin.OLD], id="off"), + pytest.param(Stage.DUALWRITE, [Origin.OLD], id="dualwrite"), + pytest.param(Stage.SHADOW, [Origin.OLD, Origin.NEW], id="shadow"), + pytest.param(Stage.LIVE, [Origin.OLD, Origin.NEW], id="live"), + pytest.param(Stage.RAMPDOWN, [Origin.NEW], id="rampdown"), + pytest.param(Stage.COMPLETE, [Origin.NEW], id="complete"), + ], + ) + async def test_reads(self, builder: AsyncMigratorBuilder, stage: Stage, origins: List[Origin]): + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + + assert result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert len(origins) == len(event.invoked) + assert all(o in event.invoked for o in origins) + + @pytest.mark.parametrize( + "stage,origins", + [ + pytest.param(Stage.OFF, [Origin.OLD], id="off"), + pytest.param(Stage.DUALWRITE, [Origin.OLD, Origin.NEW], id="dualwrite"), + pytest.param(Stage.SHADOW, [Origin.OLD, Origin.NEW], id="shadow"), + pytest.param(Stage.LIVE, [Origin.OLD, Origin.NEW], id="live"), + pytest.param(Stage.RAMPDOWN, [Origin.OLD, Origin.NEW], id="rampdown"), + pytest.param(Stage.COMPLETE, [Origin.NEW], id="complete"), + ], + ) + async def test_writes(self, builder: AsyncMigratorBuilder, stage: Stage, origins: List[Origin]): + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert result.authoritative.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert len(origins) == len(event.invoked) + assert all(o in event.invoked for o in origins) + + +# --------------------------------------------------------------------------- +# Latency tracking +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +class TestTrackingLatency: + @pytest.mark.parametrize( + "stage,origins", + [ + pytest.param(Stage.OFF, [Origin.OLD], id="off"), + pytest.param(Stage.DUALWRITE, [Origin.OLD], id="dualwrite"), + pytest.param(Stage.SHADOW, [Origin.OLD, Origin.NEW], id="shadow"), + pytest.param(Stage.LIVE, [Origin.OLD, Origin.NEW], id="live"), + pytest.param(Stage.RAMPDOWN, [Origin.NEW], id="rampdown"), + pytest.param(Stage.COMPLETE, [Origin.NEW], id="complete"), + ], + ) + async def test_reads(self, builder: AsyncMigratorBuilder, stage: Stage, origins: List[Origin]): + async def delay(payload): + await asyncio.sleep(0.1) + return Result.success("success") + + builder.track_latency(True) + builder.read(delay, delay) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + + assert result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert len(origins) == len(event.latencies) + for o in origins: + assert o in event.latencies + assert event.latencies[o] >= timedelta(milliseconds=100) + + @pytest.mark.parametrize( + "stage,origins", + [ + pytest.param(Stage.OFF, [Origin.OLD], id="off"), + pytest.param(Stage.DUALWRITE, [Origin.OLD, Origin.NEW], id="dualwrite"), + pytest.param(Stage.SHADOW, [Origin.OLD, Origin.NEW], id="shadow"), + pytest.param(Stage.LIVE, [Origin.OLD, Origin.NEW], id="live"), + pytest.param(Stage.RAMPDOWN, [Origin.OLD, Origin.NEW], id="rampdown"), + pytest.param(Stage.COMPLETE, [Origin.NEW], id="complete"), + ], + ) + async def test_writes(self, builder: AsyncMigratorBuilder, stage: Stage, origins: List[Origin]): + async def delay(payload): + await asyncio.sleep(0.1) + return Result.success("success") + + builder.track_latency(True) + builder.write(delay, delay) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert result.authoritative.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert len(origins) == len(event.latencies) + for o in origins: + assert o in event.latencies + assert event.latencies[o] >= timedelta(milliseconds=100) + + +# --------------------------------------------------------------------------- +# Error tracking +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +class TestTrackingErrors: + @pytest.mark.parametrize( + "stage,origins", + [ + pytest.param(Stage.OFF, [Origin.OLD], id="off"), + pytest.param(Stage.DUALWRITE, [Origin.OLD], id="dualwrite"), + pytest.param(Stage.SHADOW, [Origin.OLD, Origin.NEW], id="shadow"), + pytest.param(Stage.LIVE, [Origin.OLD, Origin.NEW], id="live"), + pytest.param(Stage.RAMPDOWN, [Origin.NEW], id="rampdown"), + pytest.param(Stage.COMPLETE, [Origin.NEW], id="complete"), + ], + ) + async def test_reads(self, builder: AsyncMigratorBuilder, stage: Stage, origins: List[Origin]): + async def fail(_): + return Result.fail("fail") + + builder.track_errors(True) + builder.read(fail, fail) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + + assert not result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert len(origins) == len(event.errors) + assert all(o in event.errors for o in origins) + + @pytest.mark.parametrize( + "stage,origin", + [ + pytest.param(Stage.OFF, Origin.OLD, id="off"), + pytest.param(Stage.DUALWRITE, Origin.OLD, id="dualwrite"), + pytest.param(Stage.SHADOW, Origin.OLD, id="shadow"), + pytest.param(Stage.LIVE, Origin.NEW, id="live"), + pytest.param(Stage.RAMPDOWN, Origin.NEW, id="rampdown"), + pytest.param(Stage.COMPLETE, Origin.NEW, id="complete"), + ], + ) + async def test_authoritative_writes(self, builder: AsyncMigratorBuilder, stage: Stage, origin: Origin): + async def fail(_): + return Result.fail("fail") + + builder.track_errors(True) + builder.write(fail, fail) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert not result.authoritative.is_success() + assert result.nonauthoritative is None + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert 1 == len(event.errors) + assert origin in event.errors + + @pytest.mark.parametrize( + "stage,fail_old,fail_new,origin", + [ + pytest.param(Stage.DUALWRITE, False, True, Origin.NEW, id="dualwrite"), + pytest.param(Stage.SHADOW, False, True, Origin.NEW, id="shadow"), + pytest.param(Stage.LIVE, True, False, Origin.OLD, id="live"), + pytest.param(Stage.RAMPDOWN, True, False, Origin.OLD, id="rampdown"), + ], + ) + async def test_nonauthoritative_writes(self, builder: AsyncMigratorBuilder, stage: Stage, fail_old: bool, fail_new: bool, origin: Origin): + async def success(_): + return Result.success(None) + + async def fail(_): + return Result.fail("fail") + + builder.track_errors(True) + builder.write(fail if fail_old else success, fail if fail_new else success) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert result.authoritative.is_success() + assert result.nonauthoritative is not None + assert not result.nonauthoritative.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert 1 == len(event.errors) + assert origin in event.errors + + +# --------------------------------------------------------------------------- +# Consistency tracking +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +class TestTrackingConsistency: + @pytest.mark.parametrize( + "stage", + [ + pytest.param(Stage.OFF, id="off"), + pytest.param(Stage.DUALWRITE, id="dualwrite"), + pytest.param(Stage.RAMPDOWN, id="rampdown"), + pytest.param(Stage.COMPLETE, id="complete"), + ], + ) + async def test_consistency_is_not_run_in_most_stages(self, builder: AsyncMigratorBuilder, stage: Stage): + async def value(_): + return Result.success("value") + + builder.read(value, value, lambda lhs, rhs: lhs == rhs) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + assert result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert event.consistent is None + + @pytest.mark.parametrize( + "stage,old,new,expected", + [ + pytest.param(Stage.SHADOW, "value", "value", True, id="shadow matches"), + pytest.param(Stage.LIVE, "value", "value", True, id="live matches"), + pytest.param(Stage.SHADOW, "old", "new", False, id="shadow does not match"), + pytest.param(Stage.LIVE, "old", "new", False, id="live does not match"), + ], + ) + async def test_consistency_is_tracked_correctly(self, builder: AsyncMigratorBuilder, stage: Stage, old: str, new: str, expected: bool): + async def old_fn(_): + return Result.success(old) + + async def new_fn(_): + return Result.success(new) + + builder.read(old_fn, new_fn, lambda lhs, rhs: lhs == rhs) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + assert result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert event.consistent is expected + + @pytest.mark.parametrize( + "stage,old,new", + [ + pytest.param(Stage.SHADOW, "value", "value", id="shadow"), + pytest.param(Stage.LIVE, "value", "value", id="live"), + ], + ) + async def test_consistency_handles_exceptions(self, builder: AsyncMigratorBuilder, stage: Stage, old: str, new: str): + def raise_exception(lhs, rhs): + raise Exception("error") + + async def old_fn(_): + return Result.success(old) + + async def new_fn(_): + return Result.success(new) + + builder.read(old_fn, new_fn, raise_exception) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + assert result.is_success() + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert event.consistent is None + + +# --------------------------------------------------------------------------- +# Exceptions in migrator functions +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +class TestHandlesExceptionsInMigratorFn: + @pytest.mark.parametrize( + "stage,expected_msg", + [ + pytest.param(Stage.OFF, "old read", id="off"), + pytest.param(Stage.DUALWRITE, "old read", id="dualwrite"), + pytest.param(Stage.SHADOW, "old read", id="shadow"), + pytest.param(Stage.LIVE, "new read", id="live"), + pytest.param(Stage.RAMPDOWN, "new read", id="rampdown"), + pytest.param(Stage.COMPLETE, "new read", id="complete"), + ], + ) + async def test_reads(self, builder: AsyncMigratorBuilder, stage: Stage, expected_msg: str): + builder.read(raises_exception("old read"), raises_exception("new read")) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.read(stage.value, user, Stage.LIVE) + + assert result.is_success() is False + assert str(result.exception) == expected_msg + + @pytest.mark.parametrize( + "stage,expected_msg", + [ + pytest.param(Stage.OFF, "old write", id="off"), + pytest.param(Stage.DUALWRITE, "old write", id="dualwrite"), + pytest.param(Stage.SHADOW, "old write", id="shadow"), + pytest.param(Stage.LIVE, "new write", id="live"), + pytest.param(Stage.RAMPDOWN, "new write", id="rampdown"), + pytest.param(Stage.COMPLETE, "new write", id="complete"), + ], + ) + async def test_exception_in_authoritative_write(self, builder: AsyncMigratorBuilder, stage: Stage, expected_msg: str): + builder.write(raises_exception("old write"), raises_exception("new write")) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert result.authoritative.is_success() is False + assert str(result.authoritative.exception) == expected_msg + assert result.nonauthoritative is None + + @pytest.mark.parametrize( + "stage,expected_msg,fail_old", + [ + pytest.param(Stage.DUALWRITE, "new write", False, id="dualwrite"), + pytest.param(Stage.SHADOW, "new write", False, id="shadow"), + pytest.param(Stage.LIVE, "old write", True, id="live"), + pytest.param(Stage.RAMPDOWN, "old write", True, id="rampdown"), + ], + ) + async def test_exception_in_nonauthoritative_write(self, builder: AsyncMigratorBuilder, stage: Stage, expected_msg: str, fail_old: bool): + old_fn = raises_exception("old write") if fail_old else async_success + new_fn = async_success if fail_old else raises_exception("new write") + + builder.write(old_fn, new_fn) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + result = await migrator.write(stage.value, user, Stage.LIVE) + + assert result.authoritative.is_success() + assert result.nonauthoritative is not None + assert not result.nonauthoritative.is_success() + assert str(result.nonauthoritative.exception) == expected_msg + + +# --------------------------------------------------------------------------- +# Execution order (parallel via asyncio.gather vs serial) +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +class TestSupportsExecutionOrder: + @pytest.mark.parametrize( + "order,min_time", + [ + pytest.param(ExecutionOrder.PARALLEL, 300, id="parallel"), + pytest.param(ExecutionOrder.SERIAL, 600, id="serial"), + pytest.param(ExecutionOrder.RANDOM, 600, id="random"), + ], + ) + async def test_parallel(self, builder: AsyncMigratorBuilder, order: ExecutionOrder, min_time: int): + async def delay(payload): + await asyncio.sleep(0.3) + return Result.success("success") + + builder.read_execution_order(order) + builder.read(delay, delay) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + start = datetime.now() + result = await migrator.read('live', user, Stage.LIVE) + delta = datetime.now() - start + ms = timedelta_millis(delta) + + assert result.is_success() + assert ms >= min_time From 510689e2d147b0c5a01155a86089305f44a77e42 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 12:34:10 -0500 Subject: [PATCH 2/8] refactor: Drop section-heading comments in async migrator tests --- .../testing/migrations/test_async_migrator.py | 40 ------------------- 1 file changed, 40 deletions(-) diff --git a/ldclient/testing/migrations/test_async_migrator.py b/ldclient/testing/migrations/test_async_migrator.py index 9fcbfb44..db6e1f73 100644 --- a/ldclient/testing/migrations/test_async_migrator.py +++ b/ldclient/testing/migrations/test_async_migrator.py @@ -22,10 +22,6 @@ user = Context.from_dict({u'key': u'xyz', u'kind': u'user', u'bizzle': u'def'}) -# --------------------------------------------------------------------------- -# Test doubles -# --------------------------------------------------------------------------- - class FakeEventProcessor: """Records events the way the real (async/sync) event processors expose them so tests can assert on the emitted MigrationOpEvent, mirroring the sync @@ -92,10 +88,6 @@ async def inner(payload): return inner -# --------------------------------------------------------------------------- -# Fixtures -# --------------------------------------------------------------------------- - @pytest.fixture def builder() -> AsyncMigratorBuilder: client = FakeAsyncClient() @@ -109,10 +101,6 @@ def builder() -> AsyncMigratorBuilder: return builder -# --------------------------------------------------------------------------- -# Builder validation -# --------------------------------------------------------------------------- - class TestBuilder: def test_can_build_successfully(self): client = FakeAsyncClient() @@ -156,10 +144,6 @@ def test_build_fails_without_write(self): assert migrator == "write configuration not provided" -# --------------------------------------------------------------------------- -# Payload passthrough -# --------------------------------------------------------------------------- - @pytest.mark.asyncio class TestPassingPayloadThrough: @pytest.mark.parametrize( @@ -222,10 +206,6 @@ async def capture_payloads(payload): assert all("payload" == p for p in payloads) -# --------------------------------------------------------------------------- -# Invoked tracking -# --------------------------------------------------------------------------- - @pytest.mark.asyncio class TestTrackingInvoked: @pytest.mark.parametrize( @@ -275,10 +255,6 @@ async def test_writes(self, builder: AsyncMigratorBuilder, stage: Stage, origins assert all(o in event.invoked for o in origins) -# --------------------------------------------------------------------------- -# Latency tracking -# --------------------------------------------------------------------------- - @pytest.mark.asyncio class TestTrackingLatency: @pytest.mark.parametrize( @@ -344,10 +320,6 @@ async def delay(payload): assert event.latencies[o] >= timedelta(milliseconds=100) -# --------------------------------------------------------------------------- -# Error tracking -# --------------------------------------------------------------------------- - @pytest.mark.asyncio class TestTrackingErrors: @pytest.mark.parametrize( @@ -439,10 +411,6 @@ async def fail(_): assert origin in event.errors -# --------------------------------------------------------------------------- -# Consistency tracking -# --------------------------------------------------------------------------- - @pytest.mark.asyncio class TestTrackingConsistency: @pytest.mark.parametrize( @@ -522,10 +490,6 @@ async def new_fn(_): assert event.consistent is None -# --------------------------------------------------------------------------- -# Exceptions in migrator functions -# --------------------------------------------------------------------------- - @pytest.mark.asyncio class TestHandlesExceptionsInMigratorFn: @pytest.mark.parametrize( @@ -596,10 +560,6 @@ async def test_exception_in_nonauthoritative_write(self, builder: AsyncMigratorB assert str(result.nonauthoritative.exception) == expected_msg -# --------------------------------------------------------------------------- -# Execution order (parallel via asyncio.gather vs serial) -# --------------------------------------------------------------------------- - @pytest.mark.asyncio class TestSupportsExecutionOrder: @pytest.mark.parametrize( From 9de05e276d5c1e8065485150710af161a509c382 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 16:40:49 -0500 Subject: [PATCH 3/8] refactor: Make migration configs standalone instead of a shared generic base Drop _MigrationConfigBase and give MigrationConfig and AsyncMigrationConfig their own bodies. The shared base saved only a few trivial passthrough properties while adding generic indirection, and full duplication matches how Config/AsyncConfig are handled. MigrationConfig reverts to its pre-async standalone form. --- ldclient/migrations/async_migrator.py | 39 +++++++++++++++++++- ldclient/migrations/types.py | 53 +++++++++++---------------- 2 files changed, 59 insertions(+), 33 deletions(-) diff --git a/ldclient/migrations/async_migrator.py b/ldclient/migrations/async_migrator.py index f1525651..9853100b 100644 --- a/ldclient/migrations/async_migrator.py +++ b/ldclient/migrations/async_migrator.py @@ -26,7 +26,6 @@ Origin, Stage, WriteResult, - _MigrationConfigBase, _MigratorBuilderBase ) @@ -204,7 +203,7 @@ async def __write_both(self, authoritative: AsyncExecutor, nonauthoritative: Asy return authoritative_result, nonauthoritative_result -class AsyncMigrationConfig(_MigrationConfigBase[AsyncMigratorFn]): +class AsyncMigrationConfig: """ The async counterpart to :class:`ldclient.migrations.MigrationConfig`. It stores references to coroutine functions which execute customer defined @@ -218,6 +217,42 @@ class AsyncMigrationConfig(_MigrationConfigBase[AsyncMigratorFn]): compatibility guarantees. """ + def __init__(self, old: AsyncMigratorFn, new: AsyncMigratorFn, comparison: Optional[MigratorCompareFn] = None): + self.__old = old + self.__new = new + self.__comparison = comparison + + @property + def old(self) -> AsyncMigratorFn: + """ + Coroutine function which receives a nullable payload parameter and + returns an awaitable resolving to an :class:`ldclient.Result`. + + This function call should affect the old migration origin when called. + """ + return self.__old + + @property + def new(self) -> AsyncMigratorFn: + """ + Coroutine function which receives a nullable payload parameter and + returns an awaitable resolving to an :class:`ldclient.Result`. + + This function call should affect the new migration origin when called. + """ + return self.__new + + @property + def comparison(self) -> Optional[MigratorCompareFn]: + """ + Optional (synchronous) callable which receives two objects of any kind + and returns a boolean representing equality. + + The result of this comparison can be sent upstream to LaunchDarkly to + enhance migration observability. + """ + return self.__comparison + class AsyncMigratorBuilder(_MigratorBuilderBase): """ diff --git a/ldclient/migrations/types.py b/ldclient/migrations/types.py index 54b2ce7c..2fcffbc6 100644 --- a/ldclient/migrations/types.py +++ b/ldclient/migrations/types.py @@ -1,5 +1,5 @@ from enum import Enum -from typing import Any, Callable, Generic, Optional, TypeVar +from typing import Any, Callable, Optional, TypeVar from ldclient.impl.util import Result @@ -196,41 +196,41 @@ def nonauthoritative(self) -> Optional[OperationResult]: return self.__nonauthoritative -_MigratorFnT = TypeVar('_MigratorFnT') - - -class _MigrationConfigBase(Generic[_MigratorFnT]): +class MigrationConfig: """ - Shared implementation backing :class:`MigrationConfig` and its async - counterpart. It stores references to the customer defined read or write - functions for the old and new origins, along with an optional synchronous - comparison function used for read consistency tracking. + A migration config stores references to callable methods which execute + customer defined read or write operations on old or new origins of + information. For read operations, an optional comparison function also be + defined. """ - def __init__(self, old: _MigratorFnT, new: _MigratorFnT, comparison: Optional[MigratorCompareFn] = None): - self._old = old - self._new = new - self._comparison = comparison + def __init__(self, old: MigratorFn, new: MigratorFn, comparison: Optional[MigratorCompareFn] = None): + self.__old = old + self.__new = new + self.__comparison = comparison @property - def old(self) -> _MigratorFnT: + def old(self) -> MigratorFn: """ Callable which receives a nullable payload parameter and returns an :class:`ldclient.Result`. This function call should affect the old migration origin when called. + + @return [#call] """ - return self._old + return self.__old @property - def new(self) -> _MigratorFnT: + def new(self) -> MigratorFn: """ - Callable which receives a nullable payload parameter and returns an - :class:`ldclient.Result`. - - This function call should affect the new migration origin when called. + # Callable which receives a nullable payload parameter and returns an + # :class:`ldclient.Result`. + # + # This function call should affect the new migration origin when + # called. """ - return self._new + return self.__new @property def comparison(self) -> Optional[MigratorCompareFn]: @@ -241,16 +241,7 @@ def comparison(self) -> Optional[MigratorCompareFn]: The result of this comparison can be sent upstream to LaunchDarkly to enhance migration observability. """ - return self._comparison - - -class MigrationConfig(_MigrationConfigBase[MigratorFn]): - """ - A migration config stores references to callable methods which execute - customer defined read or write operations on old or new origins of - information. For read operations, an optional comparison function also be - defined. - """ + return self.__comparison _MigratorBuilderT = TypeVar('_MigratorBuilderT', bound='_MigratorBuilderBase') From 6245fa210d2102833b52a21d28e1327ac5feb287 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 16:52:23 -0500 Subject: [PATCH 4/8] refactor: Drop _MigratorBuilderBase; inline the builder setters Give MigratorBuilder and AsyncMigratorBuilder their own read_execution_order/ track_latency/track_errors methods instead of sharing a base for three trivial property setters. MigratorBuilder and types.py revert to their pre-async form. --- ldclient/migrations/async_migrator.py | 34 +++++++++++++++++-- ldclient/migrations/migrator.py | 46 +++++++++++++++++++++----- ldclient/migrations/types.py | 47 +-------------------------- 3 files changed, 69 insertions(+), 58 deletions(-) diff --git a/ldclient/migrations/async_migrator.py b/ldclient/migrations/async_migrator.py index 9853100b..37dfb050 100644 --- a/ldclient/migrations/async_migrator.py +++ b/ldclient/migrations/async_migrator.py @@ -25,8 +25,7 @@ OperationResult, Origin, Stage, - WriteResult, - _MigratorBuilderBase + WriteResult ) if TYPE_CHECKING: @@ -254,7 +253,7 @@ def comparison(self) -> Optional[MigratorCompareFn]: return self.__comparison -class AsyncMigratorBuilder(_MigratorBuilderBase): +class AsyncMigratorBuilder: """ The async migration builder is used to configure and construct an instance of an :class:`AsyncMigrator`. This migrator can be used to perform @@ -280,6 +279,35 @@ def __init__(self, client: AsyncLDClient): self.__read_config: Optional[AsyncMigrationConfig] = None self.__write_config: Optional[AsyncMigrationConfig] = None + def read_execution_order(self, order: ExecutionOrder) -> 'AsyncMigratorBuilder': + """ + The read execution order influences the parallelism and execution order + for read operations involving multiple origins. + """ + if order not in ExecutionOrder: + return self + + self._read_execution_order = order + return self + + def track_latency(self, enabled: bool) -> 'AsyncMigratorBuilder': + """ + Enable or disable latency tracking for migration operations. This + latency information can be sent upstream to LaunchDarkly to enhance + migration visibility. + """ + self._measure_latency = enabled + return self + + def track_errors(self, enabled: bool) -> 'AsyncMigratorBuilder': + """ + Enable or disable error tracking for migration operations. This error + information can be sent upstream to LaunchDarkly to enhance migration + visibility. + """ + self._measure_errors = enabled + return self + def read(self, old: AsyncMigratorFn, new: AsyncMigratorFn, comparison: Optional[MigratorCompareFn] = None) -> 'AsyncMigratorBuilder': """ Read can be used to configure the migration-read behavior of the diff --git a/ldclient/migrations/migrator.py b/ldclient/migrations/migrator.py index 1b23e464..fde023f0 100644 --- a/ldclient/migrations/migrator.py +++ b/ldclient/migrations/migrator.py @@ -18,8 +18,7 @@ OperationResult, Origin, Stage, - WriteResult, - _MigratorBuilderBase + WriteResult ) if TYPE_CHECKING: @@ -169,7 +168,7 @@ def __write_both(self, authoritative: Executor, nonauthoritative: Executor, trac return authoritative_result, nonauthoritative_result -class MigratorBuilder(_MigratorBuilderBase): +class MigratorBuilder: """ The migration builder is used to configure and construct an instance of a :class:`Migrator`. This migrator can be used to perform LaunchDarkly @@ -182,13 +181,42 @@ def __init__(self, client: LDClient): self._client = client # Default settings as required by the spec - self._read_execution_order = ExecutionOrder.PARALLEL - self._measure_latency = True - self._measure_errors = True + self.__read_execution_order = ExecutionOrder.PARALLEL + self.__measure_latency = True + self.__measure_errors = True self.__read_config: Optional[MigrationConfig] = None self.__write_config: Optional[MigrationConfig] = None + def read_execution_order(self, order: ExecutionOrder) -> 'MigratorBuilder': + """ + The read execution order influences the parallelism and execution order + for read operations involving multiple origins. + """ + if order not in ExecutionOrder: + return self + + self.__read_execution_order = order + return self + + def track_latency(self, enabled: bool) -> 'MigratorBuilder': + """ + Enable or disable latency tracking for migration operations. This + latency information can be sent upstream to LaunchDarkly to enhance + migration visibility. + """ + self.__measure_latency = enabled + return self + + def track_errors(self, enabled: bool) -> 'MigratorBuilder': + """ + Enable or disable error tracking for migration operations. This error + information can be sent upstream to LaunchDarkly to enhance migration + visibility. + """ + self.__measure_errors = enabled + return self + def read(self, old: MigratorFn, new: MigratorFn, comparison: Optional[MigratorCompareFn] = None) -> 'MigratorBuilder': """ Read can be used to configure the migration-read behavior of the @@ -255,11 +283,11 @@ def build(self) -> Union[Migrator, str]: return MigratorImpl( Sampler(Random()), self._client, - self._read_execution_order, + self.__read_execution_order, self.__read_config, self.__write_config, - self._measure_latency, - self._measure_errors, + self.__measure_latency, + self.__measure_errors, ) diff --git a/ldclient/migrations/types.py b/ldclient/migrations/types.py index 2fcffbc6..295731ea 100644 --- a/ldclient/migrations/types.py +++ b/ldclient/migrations/types.py @@ -1,5 +1,5 @@ from enum import Enum -from typing import Any, Callable, Optional, TypeVar +from typing import Any, Callable, Optional from ldclient.impl.util import Result @@ -242,48 +242,3 @@ def comparison(self) -> Optional[MigratorCompareFn]: enhance migration observability. """ return self.__comparison - - -_MigratorBuilderT = TypeVar('_MigratorBuilderT', bound='_MigratorBuilderBase') - - -class _MigratorBuilderBase: - """ - Shared setter implementation for :class:`MigratorBuilder` and its async - counterpart. It holds the fluent configuration methods common to both - builders. The ``read``, ``write``, and ``build`` methods differ between the - sync and async builders and are defined on each subclass. - """ - - _read_execution_order: ExecutionOrder - _measure_latency: bool - _measure_errors: bool - - def read_execution_order(self: _MigratorBuilderT, order: ExecutionOrder) -> _MigratorBuilderT: - """ - The read execution order influences the parallelism and execution order - for read operations involving multiple origins. - """ - if order not in ExecutionOrder: - return self - - self._read_execution_order = order - return self - - def track_latency(self: _MigratorBuilderT, enabled: bool) -> _MigratorBuilderT: - """ - Enable or disable latency tracking for migration operations. This - latency information can be sent upstream to LaunchDarkly to enhance - migration visibility. - """ - self._measure_latency = enabled - return self - - def track_errors(self: _MigratorBuilderT, enabled: bool) -> _MigratorBuilderT: - """ - Enable or disable error tracking for migration operations. This error - information can be sent upstream to LaunchDarkly to enhance migration - visibility. - """ - self._measure_errors = enabled - return self From d62ed3c8d42712d1ca97020b3f3d58c13e62ae46 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 3 Aug 2026 08:04:22 -0600 Subject: [PATCH 5/8] chore: Retrigger PR head sync From d0c8d6bd836cee293de4ad9e1459d10d98c70c5e Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 3 Aug 2026 16:55:44 -0600 Subject: [PATCH 6/8] feat: Report migration op event when an async write is cancelled Move track_migration_op into a finally so a write cancelled mid-operation still emits its migration op event. The CancelledError propagates and no WriteResult is returned, but the event now records which origins were written, so a partial write is reported rather than lost. Reads are left as-is: a cancelled read has no side effect and loses its consistency measurement, so a partial read event carries no useful signal. --- ldclient/migrations/async_migrator.py | 57 ++++++++++++------- .../testing/migrations/test_async_migrator.py | 28 +++++++++ 2 files changed, 64 insertions(+), 21 deletions(-) diff --git a/ldclient/migrations/async_migrator.py b/ldclient/migrations/async_migrator.py index 37dfb050..ad398052 100644 --- a/ldclient/migrations/async_migrator.py +++ b/ldclient/migrations/async_migrator.py @@ -83,6 +83,16 @@ async def write(self, key: str, context: Context, default_stage: Stage, payload: :param context: The context to use when evaluating the flag :param default_stage: A default stage to fallback to if one cannot be determined :param payload: An optional payload to be passed through to the appropriate write method + + Writes run serially, authoritative origin first. If the authoritative + write fails, the non-authoritative write does not run. See + :class:`WriteResult`. + + .. note:: + Cancelling a write mid-operation raises ``CancelledError`` instead + of returning a :class:`WriteResult`. A completed authoritative write + is not undone, and the migration event still records which origins + were written. """ @@ -143,27 +153,32 @@ async def write(self, key: str, context: Context, default_stage: Stage, payload: old = AsyncExecutor(Origin.OLD, self._write_config.old, tracker, self._measure_latency, self._measure_errors, payload) new = AsyncExecutor(Origin.NEW, self._write_config.new, tracker, self._measure_latency, self._measure_errors, payload) - if stage == Stage.OFF: - result = await old.run() - write_result = WriteResult(result) - elif stage == Stage.DUALWRITE: - authoritative_result, nonauthoritative_result = await self.__write_both(old, new, tracker) - write_result = WriteResult(authoritative_result, nonauthoritative_result) - elif stage == Stage.SHADOW: - authoritative_result, nonauthoritative_result = await self.__write_both(old, new, tracker) - write_result = WriteResult(authoritative_result, nonauthoritative_result) - elif stage == Stage.LIVE: - authoritative_result, nonauthoritative_result = await self.__write_both(new, old, tracker) - write_result = WriteResult(authoritative_result, nonauthoritative_result) - elif stage == Stage.RAMPDOWN: - authoritative_result, nonauthoritative_result = await self.__write_both(new, old, tracker) - write_result = WriteResult(authoritative_result, nonauthoritative_result) - else: - result = await new.run() - write_result = WriteResult(result) - - # track_migration_op is synchronous on the async client; do not await it. - self._client.track_migration_op(tracker) + try: + if stage == Stage.OFF: + result = await old.run() + write_result = WriteResult(result) + elif stage == Stage.DUALWRITE: + authoritative_result, nonauthoritative_result = await self.__write_both(old, new, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + elif stage == Stage.SHADOW: + authoritative_result, nonauthoritative_result = await self.__write_both(old, new, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + elif stage == Stage.LIVE: + authoritative_result, nonauthoritative_result = await self.__write_both(new, old, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + elif stage == Stage.RAMPDOWN: + authoritative_result, nonauthoritative_result = await self.__write_both(new, old, tracker) + write_result = WriteResult(authoritative_result, nonauthoritative_result) + else: + result = await new.run() + write_result = WriteResult(result) + finally: + # Emit the event even if the write is cancelled mid-operation. On + # cancellation the CancelledError propagates and no WriteResult is + # returned, but the tracker still shows which origins were written, + # so a partial write is reported instead of lost. track_migration_op + # is synchronous; do not await it. + self._client.track_migration_op(tracker) return write_result diff --git a/ldclient/testing/migrations/test_async_migrator.py b/ldclient/testing/migrations/test_async_migrator.py index db6e1f73..b3a3f1fe 100644 --- a/ldclient/testing/migrations/test_async_migrator.py +++ b/ldclient/testing/migrations/test_async_migrator.py @@ -587,3 +587,31 @@ async def delay(payload): assert result.is_success() assert ms >= min_time + + +def raises_cancelled() -> AsyncMigratorFn: + async def inner(payload): + raise asyncio.CancelledError() + + return inner + + +@pytest.mark.asyncio +class TestReportsCancelledWrite: + async def test_cancelled_write_still_reports_completed_origins(self, builder: AsyncMigratorBuilder): + # In SHADOW the old origin is authoritative. It succeeds, then the new + # write is cancelled mid-operation. + builder.write(async_success, raises_cancelled()) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + with pytest.raises(asyncio.CancelledError): + await migrator.write(Stage.SHADOW.value, user, Stage.LIVE) + + # The finally still emitted the migration op event, and it records only + # the origin that completed. The cancelled write leaves the new origin + # uninvoked and unerrored. + event = builder._client._event_processor._events[1] # type: ignore + assert isinstance(event, MigrationOpEvent) + assert event.invoked == {Origin.OLD} + assert event.errors == set() From 14b33c68a11876bc1b3b839175e6cd2fa1c6f5eb Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 3 Aug 2026 17:03:38 -0600 Subject: [PATCH 7/8] docs: Remove unnecessary comment above async migrations export --- ldclient/migrations/__init__.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/ldclient/migrations/__init__.py b/ldclient/migrations/__init__.py index 2e1269b3..a5529187 100644 --- a/ldclient/migrations/__init__.py +++ b/ldclient/migrations/__init__.py @@ -1,5 +1,3 @@ -# async_migrator is import-cheap (asyncio stdlib only, no aiohttp), so it is -# exported eagerly alongside the sync surface and keeps `import ldclient` cheap. from .async_migrator import * from .migrator import * from .tracker import * From ef02557c946bdeb3ce8b368b002010ceb4b9fe02 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 3 Aug 2026 17:17:03 -0600 Subject: [PATCH 8/8] fix: Skip async migration event when no origin was written On an early cancel (before any origin write completes) OpTracker.build returned "no origins were invoked" and the client logged a spurious error. Guard the write finally with OpTracker.has_invocations so the event is only emitted when at least one origin ran. Drop the now-redundant finally comment. --- ldclient/migrations/async_migrator.py | 8 ++------ ldclient/migrations/tracker.py | 5 +++++ ldclient/testing/migrations/test_async_migrator.py | 14 ++++++++++++++ 3 files changed, 21 insertions(+), 6 deletions(-) diff --git a/ldclient/migrations/async_migrator.py b/ldclient/migrations/async_migrator.py index ad398052..c444f2fc 100644 --- a/ldclient/migrations/async_migrator.py +++ b/ldclient/migrations/async_migrator.py @@ -173,12 +173,8 @@ async def write(self, key: str, context: Context, default_stage: Stage, payload: result = await new.run() write_result = WriteResult(result) finally: - # Emit the event even if the write is cancelled mid-operation. On - # cancellation the CancelledError propagates and no WriteResult is - # returned, but the tracker still shows which origins were written, - # so a partial write is reported instead of lost. track_migration_op - # is synchronous; do not await it. - self._client.track_migration_op(tracker) + if tracker.has_invocations(): + self._client.track_migration_op(tracker) return write_result diff --git a/ldclient/migrations/tracker.py b/ldclient/migrations/tracker.py index 89f12425..ea9696e4 100644 --- a/ldclient/migrations/tracker.py +++ b/ldclient/migrations/tracker.py @@ -134,6 +134,11 @@ def invoked(self, origin: Origin) -> 'OpTracker': self.__invoked.add(origin) return self + def has_invocations(self) -> bool: + """Returns whether any origin has been recorded as invoked.""" + with self.__mutex: + return len(self.__invoked) > 0 + def consistent(self, is_consistent: Callable[[], bool]) -> 'OpTracker': """ Allows recording the results of a consistency check. diff --git a/ldclient/testing/migrations/test_async_migrator.py b/ldclient/testing/migrations/test_async_migrator.py index b3a3f1fe..652d19c1 100644 --- a/ldclient/testing/migrations/test_async_migrator.py +++ b/ldclient/testing/migrations/test_async_migrator.py @@ -615,3 +615,17 @@ async def test_cancelled_write_still_reports_completed_origins(self, builder: As assert isinstance(event, MigrationOpEvent) assert event.invoked == {Origin.OLD} assert event.errors == set() + + async def test_cancelled_before_any_write_reports_nothing(self, builder: AsyncMigratorBuilder): + # In SHADOW the old origin is authoritative. It is cancelled before it + # completes, so no origin is written. + builder.write(raises_cancelled(), async_success) + migrator = builder.build() + assert isinstance(migrator, AsyncMigrator) + + with pytest.raises(asyncio.CancelledError): + await migrator.write(Stage.SHADOW.value, user, Stage.LIVE) + + # Nothing was written, so no migration event is emitted (and the client + # does not log a spurious "no origins were invoked" error). + assert builder._client._event_processor._events == [] # type: ignore