refactor: Simplify the FDv2 store persist path and clean up docstrings

Remove the _stage_persist_full/_stage_persist_delta hook: _set_basis and
_apply_delta now return the collections to persist, and each apply does its
own persist (sync inline under the lock, async awaited under the persist
lock). Make AsyncStore.commit take its snapshot and write under the persist
lock so the flush is atomic. Tidy the _StoreBase and read-only view
docstrings.

Author: jsonbailey

ldclient/impl/datasystem/async_fdv1.py

@@ -36,11 +36,11 @@
 
 
 class _AsyncReadOnlyFeatureStoreView(AsyncReadOnlyStore):
-    """Exposes only async ``get``/``all`` over a stable async store, decoding dict items.
+    """Read-only view of an async feature store.
 
-    The wrapped store is always async and never swaps, so it is read directly.
-    Items stored as dicts are decoded into model objects; items already decoded
-    are returned unchanged.
+    Serves every read from the wrapped store. Items that a custom feature store
+    keeps as raw dicts are decoded into model objects; items that are already
+    models pass through unchanged.
     """
 
     def __init__(self, store: AsyncReadOnlyStore):

ldclient/impl/datasystem/async_store.py

@@ -43,9 +43,6 @@ def __init__(
         self._persistent_store_status_provider: Optional[DataStoreStatusProvider] = None
         self._persistent_store_writable = False
 
-        # True if the data in the memory store may be persisted to the persistent store
-        self._persist = False
-
         # Serializes async store writes; held only across the awaited I/O, never with self._lock.
         self._async_persist_lock = AsyncLock()
 
@@ -97,23 +94,10 @@ def _on_memory_store_active(self) -> None:
             except Exception as e:
                 log.warning("Failed to disable persistent store cache: %s", e)
 
-    def _stage_persist_full(self, collections: Collections, persist: bool) -> Optional[Collections]:
-        self._persist = persist
-        return collections if self._should_persist() else None
-
-    def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optional[Collections]:
-        self._persist = persist
-        return collections if self._should_persist() else None
-
     async def apply(self, change_set: ChangeSet, persist: bool) -> None:
         """
-        Apply a changeset to the store using the async persist path.
-
-        The in-memory update and change-set notification run under the
-        synchronous lock with no awaits inside it. The persistent-store write is
-        awaited afterwards, outside the lock, serialized by the async persist lock.
-        The in-memory store is authoritative, so change listeners fire before the
-        awaited store write completes.
+        Apply a changeset to the in-memory store and, if configured, the async
+        persistent store.
 
         Args:
             change_set: The changeset to apply
@@ -160,10 +144,8 @@ async def apply(self, change_set: ChangeSet, persist: bool) -> None:
 
     async def commit(self) -> Optional[Exception]:
         """
-        Persist the data in the memory store to the async persistent store, if configured.
-
-        The memory read happens under the synchronous lock; the store write is
-        awaited afterwards, serialized by the async persist lock.
+        Persist the contents of the memory store to the async persistent store,
+        if configured.
 
         Returns:
             Exception if the commit failed, None otherwise
@@ -174,21 +156,21 @@ def __mapping(data: Dict[str, ModelEntity]) -> Dict[str, Dict[str, Any]]:
 
             return __mapping
 
-        all_data: Optional[Collections] = None
-        with self._lock:
-            if self._should_persist():
-                all_data = {}
-                for kind in [FEATURES, SEGMENTS]:
-                    all_data[kind] = self._memory_store.all(kind, __mapping_from_kind(kind))
-
-        if all_data is None:
-            return None
-
         store = self._persistent_store
         if store is None:
             return None
 
         async with self._async_persist_lock:
+            all_data: Optional[Collections] = None
+            with self._lock:
+                if self._should_persist():
+                    all_data = {}
+                    for kind in [FEATURES, SEGMENTS]:
+                        all_data[kind] = self._memory_store.all(kind, __mapping_from_kind(kind))
+
+            if all_data is None:
+                return None
+
             try:
                 await store.init(all_data)
             except Exception as e:

ldclient/impl/datasystem/fdv1.py

@@ -147,11 +147,11 @@ def is_monitoring_enabled(self) -> bool:
 
 
 class _ReadOnlyFeatureStoreView(ReadOnlyStore):
-    """Exposes only ``get``/``all`` over a feature store, decoding dict items.
+    """Read-only view of a feature store.
 
-    The wrapped store is stable and never swaps, so it is read directly. Items
-    stored as dicts are decoded into model objects; items already decoded are
-    returned unchanged, then the caller's ``callback`` is applied.
+    Serves every read from the wrapped store. Items that a custom feature store
+    keeps as raw dicts are decoded into model objects; items that are already
+    models pass through unchanged.
     """
 
     def __init__(self, store: FeatureStore):

ldclient/impl/datasystem/fdv2.py

@@ -37,13 +37,12 @@
 
 
 class _ReadOnlyStoreView(ReadOnlyStore):
-    """Exposes only ``get``/``all`` over a store, decoding dict items.
+    """Read-only view of the data system store.
 
-    Resolves the active store on each read rather than at construction, so a held
-    instance follows the active-store swap: reads hit the persistent store before
-    the in-memory store has data, and the in-memory store afterwards. Items stored
-    as dicts are decoded into model objects; items already decoded are returned
-    unchanged, then the caller's ``callback`` is applied.
+    Serves every read from the store's active store, so a held instance follows
+    the swap from the persistent store to the in-memory store once it has data.
+    Items that a custom persistent store keeps as raw dicts are decoded into
+    model objects; items that are already models pass through unchanged.
     """
 
     def __init__(self, store: Store):

ldclient/impl/datasystem/store.py

@@ -160,11 +160,8 @@ class _StoreBase:
     synchronizer), reads serve from memory and the persistent store is no longer
     read from.
 
-    This base holds the in-memory store, dependency tracking, listeners, the
-    active-store swap, and the changeset-to-memory apply. It never references a
-    persistent store: subclasses own their concretely-typed store and supply the
-    persist step through the ``_stage_persist_full``/``_stage_persist_delta``
-    hooks and the ``_on_memory_store_active`` hook.
+    This base owns the in-memory store, dependency tracking, listeners, and the
+    active-store swap. It holds no persistent store of its own.
     """
 
     def __init__(
@@ -198,6 +195,10 @@ def __init__(
         # Thread synchronization
         self._lock = threading.RLock()
 
+        # True if the data in the memory store may be written to the persistent
+        # store. Set on each apply from its persist flag.
+        self._persist = False
+
     def selector(self) -> Selector:
         """Returns the current selector."""
         with self._lock:
@@ -211,25 +212,13 @@ def _on_memory_store_active(self) -> None:
         """
         pass
 
-    def _stage_persist_full(self, collections: Collections, persist: bool) -> Optional[Collections]:
-        """
-        Persist a full data set. Subclasses supply the persist step.
-
-        A subclass either writes the data synchronously and returns None, or
-        returns the collections for the caller to persist afterwards. The
-        subclass owns the decision of whether to persist at all.
-        """
-        raise NotImplementedError
-
-    def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optional[Collections]:
+    def _should_persist(self) -> bool:
         """
-        Persist a delta update. Subclasses supply the persist step.
-
-        A subclass either writes the data synchronously and returns None, or
-        returns the collections for the caller to persist afterwards. The
-        subclass owns the decision of whether to persist at all.
+        Returns whether the current data should be written to the persistent
+        store. The base engine has no persistent store, so it never persists;
+        subclasses override this based on their store.
         """
-        raise NotImplementedError
+        return False
 
     def _set_basis(
         self, collections: Collections, selector: Selector, persist: bool
@@ -243,8 +232,10 @@ def _set_basis(
             persist: Whether to persist the data to the persistent store
 
         Returns:
-            The collections the subclass deferred for later persistence, or None.
+            The collections to persist, or None if there is nothing to persist.
         """
+        self._persist = persist
+
         # Take snapshot for change detection if we have flag listeners
         old_data: Optional[Collections] = None
         if self._flag_change_listeners.has_listeners():
@@ -266,20 +257,17 @@ def _set_basis(
         self._active_store = self._memory_store
 
         # In-memory store is now authoritative. The subclass reacts here (e.g. by
-        # disabling the persistent-store cache) before the persist step below.
+        # disabling the persistent-store cache) before the caller persists.
         self._on_memory_store_active()
 
-        # Persist through the subclass hook
-        pending = self._stage_persist_full(collections, persist)
-
         # Send change events if we had listeners
         if old_data is not None:
             affected_items = self._compute_changed_items_for_full_data_set(
                 old_data, collections
             )
             self._send_change_events(affected_items)
 
-        return pending
+        return collections if self._should_persist() else None
 
     def _apply_delta(
         self, collections: Collections, selector: Selector, persist: bool
@@ -293,8 +281,10 @@ def _apply_delta(
             persist: Whether to persist the changes to the persistent store
 
         Returns:
-            The collections the subclass deferred for later persistence, or None.
+            The collections to persist, or None if there is nothing to persist.
         """
+        self._persist = persist
+
         ok = self._memory_store.apply_delta(collections)
         if ok is False:
             return None
@@ -317,13 +307,11 @@ def _apply_delta(
         # Update state
         self._selector = selector if selector is not None else Selector.no_selector()
 
-        pending = self._stage_persist_delta(collections, persist)
-
         # Send change events
         if affected_items:
             self._send_change_events(affected_items)
 
-        return pending
+        return collections if self._should_persist() else None
 
     def _changes_to_store_data(self, changes: List[Change]) -> Collections:
         """
@@ -424,9 +412,6 @@ def __init__(
         self._persistent_store_status_provider: Optional[DataStoreStatusProvider] = None
         self._persistent_store_writable = False
 
-        # True if the data in the memory store may be persisted to the persistent store
-        self._persist = False
-
     def with_persistence(
         self,
         persistent_store: FeatureStore,
@@ -456,27 +441,44 @@ def with_persistence(
 
     def apply(self, change_set: ChangeSet, persist: bool) -> None:
         """
-        Apply a changeset to the store.
+        Apply a changeset to the in-memory store and, if configured, the
+        persistent store.
 
         Args:
             change_set: The changeset to apply
             persist: Whether the changes should be persisted to the persistent store
         """
         collections = self._changes_to_store_data(change_set.changes)
 
+        pending: Optional[Collections] = None
+        is_full = False
+
         with self._lock:
             try:
                 if change_set.intent_code == IntentCode.TRANSFER_FULL:
-                    self._set_basis(collections, change_set.selector, persist)
+                    pending = self._set_basis(collections, change_set.selector, persist)
+                    is_full = True
                 elif change_set.intent_code == IntentCode.TRANSFER_CHANGES:
-                    self._apply_delta(collections, change_set.selector, persist)
+                    pending = self._apply_delta(collections, change_set.selector, persist)
                 elif change_set.intent_code == IntentCode.TRANSFER_NONE:
                     # No-op, no changes to apply
                     return
 
                 # Notify changeset listeners
                 self._change_set_listeners.notify(change_set)
 
+                # Persist synchronously, inline under the lock
+                if pending is not None:
+                    store = self._persistent_store
+                    assert store is not None
+                    if is_full:
+                        store.init(pending)
+                    else:
+                        for kind in pending:
+                            kind_data = pending[kind]
+                            for key in kind_data:
+                                store.upsert(kind, kind_data[key])
+
             except Exception as e:
                 # Log error but don't re-raise - matches Go behavior
                 log.error("Store: couldn't apply changeset: %s", str(e))
@@ -502,27 +504,6 @@ def _on_memory_store_active(self) -> None:
             except Exception as e:
                 log.warning("Failed to disable persistent store cache: %s", e)
 
-    def _stage_persist_full(self, collections: Collections, persist: bool) -> Optional[Collections]:
-        self._persist = persist
-        if not self._should_persist():
-            return None
-        store = self._persistent_store
-        assert store is not None
-        store.init(collections)
-        return None
-
-    def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optional[Collections]:
-        self._persist = persist
-        if not self._should_persist():
-            return None
-        store = self._persistent_store
-        assert store is not None
-        for kind in collections:
-            kind_data = collections[kind]
-            for key in kind_data:
-                store.upsert(kind, kind_data[key])
-        return None
-
     def commit(self) -> Optional[Exception]:
         """
         Commit persists the data in the memory store to the persistent store, if configured.