refactor: Refine async client start/close lifecycle and hook registration

Set _started before __start_up so a failed or in-progress start still
tears down through close(). Append plugin hooks to those already
registered (config plus any added via add_hook before start) rather
than overwriting. Stop the event processor last on close because it may
still be sending events the other components generated.

Author: jsonbailey

ldclient/async_client.py

@@ -127,11 +127,11 @@ async def variation_eval_fn(key, context):
         )
 
     async def start(self, start_wait: float = 5.0) -> None:
-        """Start the client: create the HTTP session, data system, and event processor.
+        """Start the client's background work: create the shared HTTP session and
+        start the data source, the big-segment status poll, and the event processor.
 
-        Single-shot. Calling start() again after it has started is a no-op. If
-        start() raises, the client is spent: it is marked closed and later
-        start() calls are ignored, so construct a new client to retry. Calling
+        Single-shot. Calling start() again after it has started -- or after a
+        failed start -- is a no-op; construct a new client to retry. Calling
         start() after close() is also a logged no-op.
 
         :param start_wait: seconds to wait for the data source to initialize
@@ -143,48 +143,33 @@ async def start(self, start_wait: float = 5.0) -> None:
             if self._started:
                 return
 
-            # __start_up resets the hook list to config.hooks + plugin hooks;
-            # preserve any hooks registered via add_hook() before start().
-            pre_start_hooks = [h for h in self.__hooks if h not in self._config.hooks]
+            self._started = True
 
             try:
                 await self.__start_up(start_wait)
-                self._started = True
             except BaseException:
-                # Catch BaseException, not Exception, so a cancelled start
-                # (CancelledError) also releases the components __start_up began.
-                # re-raise propagates the error/cancellation; it is not swallowed.
+                # Catch BaseException so a cancelled start (CancelledError) also
+                # releases the components __start_up began; re-raise propagates it.
                 await self._cleanup_partial_start()
-                self._closed = True
                 raise
 
-            for hook in pre_start_hooks:
-                self.add_hook(hook)
-
     async def _cleanup_partial_start(self):
-        """Release any resources that were partially created during a failed __start_up."""
-        try:
-            await self._event_processor.stop()
-        except Exception:
-            pass
-        try:
-            await self._data_system.stop()
-        except Exception:
-            pass
-        try:
-            manager = self.__big_segment_store_manager
-        except AttributeError:
-            manager = None
-        if manager is not None:
+        """Release any resources that were partially created during a failed
+        __start_up."""
+        for name, stop in (
+            ("data system", self._data_system.stop),
+            ("big-segment store manager", self.__big_segment_store_manager.stop),
+            ("event processor", self._event_processor.stop),
+        ):
             try:
-                await manager.stop()
-            except Exception:
-                pass
+                await stop()
+            except Exception as e:
+                log.warning("Error stopping %s during failed start: %s", name, e)
         if self._session is not None:
             try:
                 await self._session.close()
-            except Exception:
-                pass
+            except Exception as e:
+                log.warning("Error closing HTTP session during failed start: %s", e)
             self._session = None
 
     async def close(self, close_timeout: float = 2.0) -> None:
@@ -197,6 +182,7 @@ async def close(self, close_timeout: float = 2.0) -> None:
                 return
             self._closed = True
 
+            # A client that was never started has nothing running to release.
             if self._started:
                 try:
                     await asyncio.wait_for(self._close_components(), timeout=close_timeout)
@@ -215,15 +201,19 @@ async def _close_components(self):
         """Releases the threads and network connections used by the SDK
         components. The public :meth:`close` wraps this with a timeout."""
         log.info("Closing LaunchDarkly client..")
-        await self._event_processor.stop()
         await self._data_system.stop()
         await self.__big_segment_store_manager.stop()
 
+        # The event processor is last because it may still be sending events that were generated by the other components.
+        await self._event_processor.stop()
+
     async def __start_up(self, start_wait: float):
         environment_metadata = get_environment_metadata(self._config, "python-server-sdk-async")
         plugin_hooks = get_plugin_hooks(self._config.plugins, environment_metadata)
 
-        self.__hooks = self._config.hooks + plugin_hooks
+        # Append plugin hooks to those already registered (the config hooks plus
+        # any added via add_hook() before start()), rather than overwriting.
+        self.__hooks = self.__hooks + plugin_hooks
 
         # Start the big-segment status poll now that a loop is running.
         self.__big_segment_store_manager.start()
@@ -325,7 +315,7 @@ def track_migration_op(self, tracker: OpTracker):
         event = tracker.build()
 
         if isinstance(event, str):
-            log.error("error generting migration op event %s; no event will be emitted", event)
+            log.error("error generating migration op event %s; no event will be emitted", event)
             return
 
         self._send_event(event)
@@ -354,7 +344,7 @@ def track(self, event_name: str, context: Context, data: Optional[Any] = None, m
     def identify(self, context: Context):
         """Reports details about an evaluation context.
 
-        This method simply creates an analytics event containing the context properties, to
+        This method simply creates an analytics event containing the context properties, so
         that LaunchDarkly will know about that context if it does not already.
 
         Evaluating a flag, by calling :func:`variation()` or :func:`variation_detail()`, also
@@ -393,7 +383,7 @@ async def flush(self):
         Normally, batches of events are delivered in the background at intervals determined by the
         ``flush_interval`` property of :class:`ldclient.config.Config`. Calling ``flush()``
         schedules the next event delivery to be as soon as possible; however, the delivery still
-        happens asynchronously on a worker thread, so this method will return immediately.
+        happens asynchronously in the background, so this method will return immediately.
         """
         if self._config.offline:
             return

ldclient/testing/test_async_client.py

@@ -384,11 +384,11 @@ def boom(*args, **kwargs):
 
     with pytest.raises(RuntimeError):
         await client.start()
-    assert client._closed is True
+    # A failed start marks the client started (spent); a retry is a no-op.
+    assert client._started is True
 
-    # A retry does not re-run start-up.
+    # Retry does not re-run start-up (it would raise again if it did).
     await client.start()
-    assert client._started is False
 
 
 @pytest.mark.asyncio
@@ -415,10 +415,10 @@ def cancel(*args, **kwargs):
     with pytest.raises(asyncio.CancelledError):
         await client.start()
 
-    # Cleanup (which stops the started components) ran, and the instance is spent.
+    # Cleanup ran (stopping the started components), the instance is spent
+    # (single-shot via _started), and the CancelledError propagated.
     assert cleaned is True
-    assert client._closed is True
-    assert client._started is False
+    assert client._started is True
 
 
 @pytest.mark.asyncio