fix: Skip refresh if already in progress or if lock is already held

ntkathole · ntkathole · commit 37337db4f79c · 2025-02-18T19:08:04.000+05:30
Signed-off-by: ntkathole &lt;nikhilkathole2683@gmail.com&gt;
diff --git a/sdk/python/feast/infra/registry/caching_registry.py b/sdk/python/feast/infra/registry/caching_registry.py
@@ -37,6 +37,7 @@ def __init__(self, project: str, cache_ttl_seconds: int, cache_mode: str):
         )
         self.cached_registry_proto = self.proto()
         self.cached_registry_proto_created = _utc_now()
+        self._refreshing = False  # Prevents reentrant calls to refresh
         if cache_mode == "thread":
             self._start_thread_async_refresh(cache_ttl_seconds)
             atexit.register(self._exit_handler)
@@ -425,12 +426,26 @@ def list_projects(
         return self._list_projects(tags)
 
     def refresh(self, project: Optional[str] = None):
-        self.cached_registry_proto = self.proto()
-        self.cached_registry_proto_created = _utc_now()
+        if self._refreshing:  # Skipping refresh if already in progress
+            return
+        self._refreshing = True
+        try:
+            self.cached_registry_proto = self.proto()
+            self.cached_registry_proto_created = _utc_now()
+            logger.info("Registry cache refreshed successfully")
+        except Exception as e:
+            logger.error(f"Error while refreshing registry: {e}", exc_info=True)
+        finally:
+            self._refreshing = False  # Always reset flag
 
     def _refresh_cached_registry_if_necessary(self):
         if self.cache_mode == "sync":
-            with self._refresh_lock:
+            # Try acquiring the lock without blocking
+            if not self._refresh_lock.acquire(
+                blocking=False
+            ):  # Skipping refresh if lock is already held by another thread
+                return
+            try:
                 if self.cached_registry_proto == RegistryProto():
                     # Avoids the need to refresh the registry when cache is not populated yet
                     # Specially during the __init__ phase
@@ -454,6 +469,13 @@ def _refresh_cached_registry_if_necessary(self):
                 if expired:
                     logger.info("Registry cache expired, so refreshing")
                     self.refresh()
+            except Exception as e:
+                logger.error(
+                    f"Error in _refresh_cached_registry_if_necessary: {e}",
+                    exc_info=True,
+                )
+            finally:
+                self._refresh_lock.release()  # Always release the lock safely
 
     def _start_thread_async_refresh(self, cache_ttl_seconds):
         self.refresh()
