fix: Log async persist failures instead of raising them

Also rename AsyncStore.apply_async/commit_async/close_async to
apply/commit/close (the async suffix was a carryover; AsyncStore is a
sibling of Store and inherits no sync methods to disambiguate from), and
fix isort ordering in async_config.py and the wrapper test.

Author: jsonbailey

ldclient/async_config.py

@@ -8,11 +8,10 @@
     compatibility guarantees.
 """
 
+from dataclasses import dataclass
 from typing import Callable, List, Optional, Set
 
 from ldclient.async_feature_store import AsyncInMemoryFeatureStore
-from dataclasses import dataclass
-
 from ldclient.config import (
     DEFAULT_BASE_URI,
     DEFAULT_EVENTS_URI,

ldclient/impl/datasystem/async_store.py

@@ -105,7 +105,7 @@ def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optio
         self._persist = persist
         return collections if self._should_persist() else None
 
-    async def apply_async(self, change_set: ChangeSet, persist: bool) -> None:
+    async def apply(self, change_set: ChangeSet, persist: bool) -> None:
         """
         Apply a changeset to the store using the async persist path.
 
@@ -147,15 +147,18 @@ async def apply_async(self, change_set: ChangeSet, persist: bool) -> None:
             return
 
         async with self._async_persist_lock:
-            if is_full:
-                await store.init(pending)
-            else:
-                for kind in pending:
-                    kind_data = pending[kind]
-                    for key in kind_data:
-                        await store.upsert(kind, kind_data[key])
-
-    async def commit_async(self) -> Optional[Exception]:
+            try:
+                if is_full:
+                    await store.init(pending)
+                else:
+                    for kind in pending:
+                        kind_data = pending[kind]
+                        for key in kind_data:
+                            await store.upsert(kind, kind_data[key])
+            except Exception as e:
+                log.error("Store: couldn't persist changeset: %s", str(e))
+
+    async def commit(self) -> Optional[Exception]:
         """
         Persist the data in the memory store to the async persistent store, if configured.
 
@@ -192,7 +195,7 @@ def __mapping(data: Dict[str, ModelEntity]) -> Dict[str, Dict[str, Any]]:
                 return e
         return None
 
-    async def close_async(self) -> Optional[Exception]:
+    async def close(self) -> Optional[Exception]:
         """
         Close the store and the async persistent store, if configured.
 

ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py

@@ -94,12 +94,12 @@ def _delta_changeset(key: str, version: int, on: bool) -> ChangeSet:
 
 
 @pytest.mark.asyncio
-async def test_apply_async_full_transfer_persists_via_init():
+async def test_apply_full_transfer_persists_via_init():
     async_store = FakeAsyncFeatureStore()
     store = AsyncStore(Listeners(), Listeners())
     store.with_async_persistence(async_store, True, None)
 
-    await store.apply_async(_full_changeset("flag-a", 1, True), True)
+    await store.apply(_full_changeset("flag-a", 1, True), True)
 
     # After a full transfer the memory store is authoritative and serves reads
     assert store.get_active_store() is store._memory_store
@@ -113,30 +113,30 @@ async def test_apply_async_full_transfer_persists_via_init():
 
 
 @pytest.mark.asyncio
-async def test_apply_async_delta_persists_via_upsert():
+async def test_apply_delta_persists_via_upsert():
     async_store = FakeAsyncFeatureStore()
     store = AsyncStore(Listeners(), Listeners())
     store.with_async_persistence(async_store, True, None)
 
-    await store.apply_async(_full_changeset("flag-a", 1, True), True)
+    await store.apply(_full_changeset("flag-a", 1, True), True)
     async_store.init_called_count = 0
     async_store.upsert_calls = []
 
-    await store.apply_async(_delta_changeset("flag-a", 2, False), True)
+    await store.apply(_delta_changeset("flag-a", 2, False), True)
 
     assert any(call[1] == "flag-a" and call[2] == 2 for call in async_store.upsert_calls)
     assert async_store.snapshot()[FEATURES]["flag-a"]["on"] is False
 
 
 @pytest.mark.asyncio
-async def test_apply_async_read_only_does_not_persist():
+async def test_apply_read_only_does_not_persist():
     async_store = FakeAsyncFeatureStore()
     store = AsyncStore(Listeners(), Listeners())
     # writable=False -> READ_ONLY: never write to the store
     store.with_async_persistence(async_store, False, None)
 
-    await store.apply_async(_full_changeset("flag-a", 1, True), True)
-    await store.apply_async(_delta_changeset("flag-a", 2, False), True)
+    await store.apply(_full_changeset("flag-a", 1, True), True)
+    await store.apply(_delta_changeset("flag-a", 2, False), True)
 
     assert async_store.init_called_count == 0
     assert async_store.upsert_calls == []
@@ -145,7 +145,7 @@ async def test_apply_async_read_only_does_not_persist():
 
 
 @pytest.mark.asyncio
-async def test_apply_async_fires_change_set_listeners():
+async def test_apply_fires_change_set_listeners():
     async_store = FakeAsyncFeatureStore()
     received: List[ChangeSet] = []
     change_set_listeners = Listeners()
@@ -155,29 +155,29 @@ async def test_apply_async_fires_change_set_listeners():
     store.with_async_persistence(async_store, True, None)
 
     cs = _full_changeset("flag-a", 1, True)
-    await store.apply_async(cs, True)
+    await store.apply(cs, True)
 
     assert received == [cs]
 
 
 @pytest.mark.asyncio
-async def test_commit_async_writes_memory_to_store():
+async def test_commit_writes_memory_to_store():
     async_store = FakeAsyncFeatureStore()
     store = AsyncStore(Listeners(), Listeners())
     store.with_async_persistence(async_store, True, None)
 
     # Populate memory without persisting yet (read-only apply through memory)
-    await store.apply_async(_full_changeset("flag-a", 1, True), True)
+    await store.apply(_full_changeset("flag-a", 1, True), True)
     async_store.init_called_count = 0
 
-    err = await store.commit_async()
+    err = await store.commit()
     assert err is None
     assert async_store.init_called_count == 1
     assert "flag-a" in async_store.snapshot()[FEATURES]
 
 
 @pytest.mark.asyncio
-async def test_commit_async_returns_error_on_failure():
+async def test_commit_returns_error_on_failure():
     class FailingStore(FakeAsyncFeatureStore):
         async def init(self, all_data):
             raise RuntimeError("boom")
@@ -186,22 +186,22 @@ async def init(self, all_data):
     store = AsyncStore(Listeners(), Listeners())
     # Read-only so the deferred persist is skipped and memory is populated first.
     store.with_async_persistence(async_store, False, None)
-    await store.apply_async(_full_changeset("flag-a", 1, True), True)
+    await store.apply(_full_changeset("flag-a", 1, True), True)
 
     # Now make it writable and commit, which triggers the failing init.
     store._persistent_store_writable = True
-    err = await store.commit_async()
+    err = await store.commit()
     assert isinstance(err, RuntimeError)
     assert str(err) == "boom"
 
 
 @pytest.mark.asyncio
-async def test_close_async_closes_async_store():
+async def test_close_closes_async_store():
     async_store = FakeAsyncFeatureStore()
     store = AsyncStore(Listeners(), Listeners())
     store.with_async_persistence(async_store, True, None)
 
-    err = await store.close_async()
+    err = await store.close()
     assert err is None
     assert async_store.closed is True
 

ldclient/testing/test_feature_store_client_wrapper.py

@@ -2,8 +2,8 @@
 from typing import Callable, List
 from unittest.mock import Mock
 
-from ldclient.impl.datasystem.fdv1 import _FeatureStoreClientWrapper
 from ldclient.impl.datastore.status import DataStoreUpdateSinkImpl
+from ldclient.impl.datasystem.fdv1 import _FeatureStoreClientWrapper
 from ldclient.impl.listeners import Listeners