feast-dev · jyejare · Jul 27, 2026 · Jul 27, 2026
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import asyncio
+import functools
 import os
 import sys
 import threading
@@ -1004,6 +1005,7 @@ def __init__(
                 store=store,
                 registry_ttl_sec=options["registry_ttl_sec"],
             )
+            self._store = store
             self._options = options
             self._metrics_enabled = metrics_enabled
             super().__init__()
@@ -1015,15 +1017,19 @@ def load_config(self):
 
             self.cfg.set("worker_class", "uvicorn_worker.UvicornWorker")
             if self._metrics_enabled:
-                self.cfg.set("post_worker_init", _gunicorn_post_worker_init)
+                self.cfg.set(
+                    "post_worker_init",
+                    functools.partial(_gunicorn_post_worker_init, self._store),
+                )
                 self.cfg.set("child_exit", _gunicorn_child_exit)
 
         def load(self):
             return self._app
 
-    def _gunicorn_post_worker_init(worker):
-        """Start per-worker resource monitoring after Gunicorn forks."""
+    def _gunicorn_post_worker_init(store: "feast.FeatureStore", worker):
+        """Start per-worker resource and freshness monitoring after Gunicorn forks."""
         feast_metrics.init_worker_monitoring()
+        feast_metrics.init_worker_freshness_monitoring(store)
 
     def _gunicorn_child_exit(server, worker):
         """Clean up Prometheus metric files for a dead worker."""
@@ -1061,6 +1067,7 @@ def start_server(
             store,
             metrics_config=flags,
             start_resource_monitoring=not uses_gunicorn,
+            start_freshness_monitoring=not uses_gunicorn,
         )
 
     logger.debug("start_server called")

@@ -572,11 +572,23 @@ def init_worker_monitoring():
         t.start()
 
 
+def init_worker_freshness_monitoring(store: "FeatureStore"):
+    """Start feature-freshness monitoring inside a Gunicorn worker process.
+
+    Called from the ``post_worker_init`` hook so that each worker starts
+    its own freshness monitoring after the fork, not before.
+    """
+    if _config.freshness:
+        t = threading.Thread(target=monitor_freshness, args=(store, 30), daemon=True)
+        t.start()
+
+
 def start_metrics_server(
     store: "FeatureStore",
     port: int = 8000,
     metrics_config: Optional["_MetricsFlags"] = None,
     start_resource_monitoring: bool = True,
+    start_freshness_monitoring: bool = True,
 ):
     """
     Start the Prometheus metrics HTTP server and background monitoring threads.
@@ -593,6 +605,10 @@ def start_metrics_server(
             monitoring thread.  Set to ``False`` when Gunicorn will
             fork workers — the ``post_worker_init`` hook starts
             per-worker monitoring instead.
+        start_freshness_monitoring: Whether to start the feature-freshness
+            thread here.  Set to ``False`` when Gunicorn will fork
+            workers — the ``post_worker_init`` hook starts per-worker
+            freshness monitoring instead.
     """
     global _config
 
@@ -632,7 +648,7 @@ def start_metrics_server(
         )
         resource_thread.start()
 
-    if _config.freshness:
+    if _config.freshness and start_freshness_monitoring:
         freshness_thread = threading.Thread(
             target=monitor_freshness, args=(store, 30), daemon=True
         )

@@ -12,10 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import json
+import sys
 import time
 from collections import Counter
 from types import SimpleNamespace
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from fastapi.testclient import TestClient
@@ -749,3 +750,17 @@ def test_metadata_model_accepts_raw_proto_dict():
     )
     assert full.metadata is not None
     assert full.metadata.feature_names == ["x", "y"]
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="Gunicorn is not used on Windows")
+def test_gunicorn_post_worker_init_starts_resource_and_freshness_monitoring():
+    from feast.feature_server import _gunicorn_post_worker_init
+
+    mock_store = MagicMock()
+    with (
+        patch("feast.feature_server.feast_metrics") as mock_fm,
+    ):
+        _gunicorn_post_worker_init(mock_store, worker=MagicMock())
+
+    mock_fm.init_worker_monitoring.assert_called_once()
+    mock_fm.init_worker_freshness_monitoring.assert_called_once_with(mock_store)
@@ -344,6 +344,51 @@ def test_only_resource_enabled(self):
         )
 
 
+class TestInitWorkerFreshnessMonitoring:
+    """init_worker_freshness_monitoring mirrors init_worker_monitoring's gating."""
+
+    def test_starts_daemon_thread_when_freshness_enabled(self):
+        import feast.metrics as m
+
+        m._config = m._MetricsFlags(
+            enabled=True,
+            resource=False,
+            request=False,
+            online_features=False,
+            push=False,
+            materialization=False,
+            freshness=True,
+        )
+        mock_store = MagicMock()
+
+        with patch("feast.metrics.threading.Thread") as mock_thread_cls:
+            m.init_worker_freshness_monitoring(mock_store)
+
+        mock_thread_cls.assert_called_once_with(
+            target=m.monitor_freshness, args=(mock_store, 30), daemon=True
+        )
+        mock_thread_cls.return_value.start.assert_called_once()
+
+    def test_does_not_start_thread_when_freshness_disabled(self):
+        import feast.metrics as m
+
+        m._config = m._MetricsFlags(
+            enabled=True,
+            resource=False,
+            request=False,
+            online_features=False,
+            push=False,
+            materialization=False,
+            freshness=False,
+        )
+        mock_store = MagicMock()
+
+        with patch("feast.metrics.threading.Thread") as mock_thread_cls:
+            m.init_worker_freshness_monitoring(mock_store)
+
+        mock_thread_cls.assert_not_called()
+
+
 class TestMetricsYamlConfig:
     """Verify metrics config in feature_store.yaml is respected.
 
@@ -438,6 +483,21 @@ def test_metrics_not_started_when_config_is_none(self):
         mock_fm = self._call_start_server(mock_store, cli_metrics=False)
         mock_fm.start_metrics_server.assert_not_called()
 
+    def test_freshness_monitoring_deferred_same_as_resource_monitoring(self):
+        """start_freshness_monitoring must mirror start_resource_monitoring exactly."""
+        from types import SimpleNamespace
+
+        mock_store = MagicMock()
+        mock_store.config = SimpleNamespace(
+            feature_server=SimpleNamespace(metrics=SimpleNamespace(enabled=True)),
+        )
+
+        mock_fm = self._call_start_server(mock_store, cli_metrics=False)
+        _, kwargs = mock_fm.start_metrics_server.call_args
+        assert (
+            kwargs["start_resource_monitoring"] == kwargs["start_freshness_monitoring"]
+        )
+
 
 class TestTrackOnlineFeaturesEntities:
     def test_increments_request_count(self):