fix: restart local-eval polling thread after fork (#77)

The polling thread that refreshes the local evaluation environment
document is created in the parent process. Threads do not survive
os.fork(), so pre-fork servers (gunicorn, uwsgi, multiprocessing)
end up with worker processes whose Python Thread object is intact
but whose underlying OS thread is dead. Workers then serve a frozen
environment-document snapshot from the moment of fork for their
entire lifetime.

EnvironmentDataPollingManager now composes a threading.Thread instead
of inheriting from it, and registers an os.register_at_fork hook so a
fresh polling thread is started in each child. The hook also closes
the parent's requests.Session so connection-pool sockets are not
shared across processes.

Stream and analytics threads are not yet covered; tracking
separately.

Author: khvn26

flagsmith/flagsmith.py

@@ -235,7 +235,6 @@ def _initialise_local_evaluation(self) -> None:
                 EnvironmentDataPollingManager(
                     main=self,
                     refresh_interval_seconds=self.environment_refresh_interval_seconds,
-                    daemon=True,
                 )
             )
 

flagsmith/polling_manager.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import os
 import threading
 import time
 import typing
@@ -11,26 +12,64 @@
 logger = logging.getLogger(__name__)
 
 
-class EnvironmentDataPollingManager(threading.Thread):
+class EnvironmentDataPollingManager:
+    """Owns the worker thread that periodically refreshes the local
+    evaluation environment document.
+
+    Composes (rather than extends) :class:`threading.Thread` so the
+    worker can be replaced — most importantly after :func:`os.fork`,
+    where threads do not survive into the child. We register an at-fork
+    hook so a forked worker (gunicorn, uwsgi, multiprocessing) gets a
+    freshly-started polling thread for its own PID. See issue #77.
+    """
+
     def __init__(
         self,
-        *args: typing.Any,
+        *,
         main: Flagsmith,
         refresh_interval_seconds: typing.Union[int, float] = 10,
-        **kwargs: typing.Any,
-    ):
-        super(EnvironmentDataPollingManager, self).__init__(*args, **kwargs)
+    ) -> None:
+        self._main = main
+        self._refresh_interval_seconds = refresh_interval_seconds
         self._stop_event = threading.Event()
-        self.main = main
-        self.refresh_interval_seconds = refresh_interval_seconds
+        self._thread = self._build_thread()
+        self._at_fork_registered = False
 
-    def run(self) -> None:
+    def _build_thread(self) -> threading.Thread:
+        return threading.Thread(target=self._run, daemon=True)
+
+    def _run(self) -> None:
         while not self._stop_event.is_set():
-            self.main.update_environment()
-            time.sleep(self.refresh_interval_seconds)
+            self._main.update_environment()
+            time.sleep(self._refresh_interval_seconds)
+
+    def start(self) -> None:
+        self._thread.start()
+        if not self._at_fork_registered and hasattr(os, "register_at_fork"):
+            os.register_at_fork(after_in_child=self._restart_after_fork)
+            self._at_fork_registered = True
+
+    def _restart_after_fork(self) -> None:
+        if self._thread.is_alive():
+            return
+        # Sockets in the parent's connection pool are inherited as
+        # shared FDs across fork; reusing them would interleave bytes
+        # between processes. Drop them so the new thread opens fresh.
+        if session := getattr(self._main, "session", None):
+            session.close()
+        self._stop_event = threading.Event()
+        self._thread = self._build_thread()
+        self._thread.start()
 
     def stop(self) -> None:
         self._stop_event.set()
 
+    def is_alive(self) -> bool:
+        return self._thread.is_alive()
+
+    @property
+    def ident(self) -> typing.Optional[int]:
+        return self._thread.ident
+
     def __del__(self) -> None:
         self._stop_event.set()

poetry.lock

@@ -30,6 +30,7 @@ pre-commit = { version = "^4.2.0", python = ">=3.9,<4" }
 responses = "^0.24.1"
 types-requests = "^2.32"
 pyfakefs = "^5.9.2"
+pytest-httpserver = {version = "^1.1.0", python = ">=3.10,<4"}
 
 [tool.mypy]
 exclude = ["example/*"]