Perf(cnr): optimize ComfyRegistry loading using server-time incremental caching.

glob/cnr_utils.py

@@ -19,64 +19,136 @@ lock = asyncio.Lock()
 
 is_cache_loading = False
 
-async def get_cnr_data(cache_mode=True, dont_wait=True):
+async def get_cnr_data(sync_mode=None, dont_wait=True, **kwargs):
+    # For backwards compatibility with keyword argument cache_mode
+    if sync_mode is None:
+        sync_mode = kwargs.get('cache_mode', 'cache')
     try:
-        return await _get_cnr_data(cache_mode, dont_wait)
+        return await _get_cnr_data(sync_mode, dont_wait)
     except asyncio.TimeoutError:
         logging.error(f"[ComfyUI-Manager] A timeout occurred during the fetch process from ComfyRegistry.")
-        return await _get_cnr_data(cache_mode=True, dont_wait=True)  # timeout fallback
+        return await _get_cnr_data(sync_mode='cache', dont_wait=True)  # timeout fallback
 
-async def _get_cnr_data(cache_mode=True, dont_wait=True):
+def get_comfyui_ver():
+    is_desktop = bool(os.environ.get('__COMFYUI_DESKTOP_VERSION__'))
+    if is_desktop:
+        return manager_core.get_current_comfyui_ver() or 'unknown'
+    else:
+        return manager_core.get_comfyui_tag() or 'unknown'
+
+
+def get_form_factor():
+    is_desktop = bool(os.environ.get('__COMFYUI_DESKTOP_VERSION__'))
+    system = platform.system().lower()
+    is_windows = system == 'windows'
+    is_mac = system == 'darwin'
+    is_linux = system == 'linux'
+
+    if is_desktop:
+        if is_windows:
+            return 'desktop-win'
+        elif is_mac:
+            return 'desktop-mac'
+        else:
+            return 'other'
+    else:
+        if is_windows:
+            return 'git-windows'
+        elif is_mac:
+            return 'git-mac'
+        elif is_linux:
+            return 'git-linux'
+        else:
+            return 'other'
+
+
+def get_node_timestamp(node):
+    latest_ver = node.get('latest_version')
+    if isinstance(latest_ver, dict):
+        t = latest_ver.get('createdAt')
+        if t:
+            return t
+    return node.get('created_at')
+
+
+async def _get_cnr_data(sync_mode=None, dont_wait=True, **kwargs):
     global is_cache_loading
 
-    uri = f'{base_url}/nodes'
+    # For backwards compatibility with keyword argument cache_mode
+    if sync_mode is None:
+        sync_mode = kwargs.get('cache_mode', 'cache')
 
-    async def fetch_all():
+    # Normalize sync_mode for backwards
+    # True | 'cache' | 'local' -> cache (just use cache)
+    # 'force' -> force (invalidate and create new cache)
+    # 'remote' -> remote (update comfyui remote with local cache)
+    if sync_mode is True or sync_mode == 'cache' or sync_mode == 'local':
+        normalized_mode = 'cache'
+    elif sync_mode == 'force':
+        normalized_mode = 'force'
+    else:
+        normalized_mode = 'remote'
+
+    uri = f'{base_url}/nodes'
+    cache_path = manager_util.get_cache_path(uri)
+
+    comfyui_ver = get_comfyui_ver()
+    form_factor = get_form_factor()
+
+    cached_data = None
+    last_updated = None
+    full_nodes = {}
+
+    if normalized_mode != 'force' and manager_util.get_cache_state(uri, expired_days=None) == 'cached':
+        try:
+            with open(cache_path, 'r', encoding="UTF-8", errors="ignore") as json_file:
+                cached_data = json.load(json_file)
+
+            if (cached_data.get('comfyui_ver') == comfyui_ver and
+                    cached_data.get('form_factor') == form_factor):
+                last_updated = cached_data.get('last_updated')
+                for node in cached_data.get('nodes', []):
+                    full_nodes[node['id']] = node
+            else:
+                logging.info("[ComfyUI-Manager] Environment change detected. Invalidating local ComfyRegistry cache.")
+                cached_data = None
+                full_nodes = {}
+                last_updated = None
+        except Exception as e:
+            logging.error(f"[ComfyUI-Manager] Failed to read cached data: {e}")
+            cached_data = None
+            full_nodes = {}
+            last_updated = None
+
+    if normalized_mode == 'cache':
+        is_cache_loading = True
+
+        if dont_wait:
+            if cached_data is not None:
+                return cached_data.get('nodes', [])
+            return {}
+
+        if cached_data is not None and manager_util.get_cache_state(uri, expired_days=1) == 'cached':
+            return cached_data.get('nodes', [])
+
+    async def fetch_all(timestamp_filter, existing_nodes):
         remained = True
         page = 1
+        nodes_map = dict(existing_nodes)
 
-        full_nodes = {}
-
-        
-        # Determine form factor based on environment and platform
-        is_desktop = bool(os.environ.get('__COMFYUI_DESKTOP_VERSION__'))
-        system = platform.system().lower()
-        is_windows = system == 'windows'
-        is_mac = system == 'darwin'
-        is_linux = system == 'linux'
-
-        # Get ComfyUI version tag
-        if is_desktop:
-            # extract version from pyproject.toml instead of git tag
-            comfyui_ver = manager_core.get_current_comfyui_ver() or 'unknown'
-        else:
-            comfyui_ver = manager_core.get_comfyui_tag() or 'unknown'
-
-        if is_desktop:
-            if is_windows:
-                form_factor = 'desktop-win'
-            elif is_mac:
-                form_factor = 'desktop-mac'
-            else:
-                form_factor = 'other'
-        else:
-            if is_windows:
-                form_factor = 'git-windows'
-            elif is_mac:
-                form_factor = 'git-mac'
-            elif is_linux:
-                form_factor = 'git-linux'
-            else:
-                form_factor = 'other'
-        
         while remained:
-            # Add comfyui_version and form_factor to the API request
             sub_uri = f'{base_url}/nodes?page={page}&limit=30&comfyui_version={comfyui_ver}&form_factor={form_factor}'
-            sub_json_obj = await asyncio.wait_for(manager_util.get_data_with_cache(sub_uri, cache_mode=False, silent=True, dont_cache=True), timeout=30)
+            if timestamp_filter:
+                sub_uri += f'&timestamp={timestamp_filter}'
+
+            sub_json_obj = await asyncio.wait_for(
+                manager_util.get_data_with_cache(sub_uri, cache_mode=False, silent=True, dont_cache=True),
+                timeout=30
+            )
             remained = page < sub_json_obj['totalPages']
 
             for x in sub_json_obj['nodes']:
-                full_nodes[x['id']] = x
+                nodes_map[x['id']] = x
 
             if page % 5 == 0:
                 logging.info(f"[ComfyUI-Manager] FETCH ComfyRegistry Data: {page}/{sub_json_obj['totalPages']}")
@@ -86,41 +158,47 @@ async def _get_cnr_data(cache_mode=True, dont_wait=True):
 
         logging.info(f"[ComfyUI-Manager] FETCH ComfyRegistry Data [DONE]")
 
-        for v in full_nodes.values():
+        for v in nodes_map.values():
             if 'latest_version' not in v:
                 v['latest_version'] = dict(version='nightly')
 
-        return {'nodes': list(full_nodes.values())}
-
-    if cache_mode:
-        is_cache_loading = True
-        cache_state = manager_util.get_cache_state(uri)
-        
-
-        if dont_wait:
-            if cache_state == 'not-cached':
-                return {}
-            else:
-                logging.warning(f"[ComfyUI-Manager] The ComfyRegistry cache update is still in progress, so an outdated cache is being used.")
-                with open(manager_util.get_cache_path(uri), 'r', encoding="UTF-8", errors="ignore") as json_file:
-                    return json.load(json_file)['nodes']
-
-        if cache_state == 'cached':
-            with open(manager_util.get_cache_path(uri), 'r', encoding="UTF-8", errors="ignore") as json_file:
-                return json.load(json_file)['nodes']
+        return {'nodes': list(nodes_map.values())}
 
     try:
-        json_obj = await fetch_all()
-        manager_util.save_to_cache(uri, json_obj)
-        return json_obj['nodes']
-    except:
-        res = {}
-        logging.error(f"[ComfyUI-Manager] Cannot connect to comfyregistry.")
-    finally:
-        if cache_mode:
-            is_cache_loading = False
+        from datetime import datetime, timezone, timedelta
+        json_obj = await fetch_all(last_updated, full_nodes)
 
-    return res
+        # Set cache's timestamp as the maximum timestamp from fetched nodes.
+        # This way, in the next run, only the latest updates will be fetched.
+        timestamps = [get_node_timestamp(node) for node in json_obj['nodes'] if get_node_timestamp(node)]
+        max_timestamp = max(timestamps) if timestamps else None
+
+        if max_timestamp:
+            try:
+                ts_str = max_timestamp.replace('Z', '+00:00')
+                dt = datetime.fromisoformat(ts_str) - timedelta(seconds=10)
+                new_timestamp = dt.strftime("%Y-%m-%dT%H:%M:%SZ")
+            except Exception:
+                new_timestamp = max_timestamp
+        else:
+            new_timestamp = last_updated if last_updated else datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+        cache_to_save = {
+            'nodes': json_obj['nodes'],
+            'comfyui_ver': comfyui_ver,
+            'form_factor': form_factor,
+            'last_updated': new_timestamp
+        }
+        manager_util.save_to_cache(uri, cache_to_save)
+        return json_obj['nodes']
+    except Exception as e:
+        logging.error(f"[ComfyUI-Manager] Cannot connect to comfyregistry or failed sync: {e}")
+        if cached_data is not None:
+            return cached_data.get('nodes', [])
+        return []
+    finally:
+        if normalized_mode == 'cache':
+            is_cache_loading = False
 
 
 @dataclass

glob/manager_core.py

@@ -811,7 +811,7 @@ class UnifiedManager:
             dont_wait = True
 
         # reload 'cnr_map' and 'repo_cnr_map'
-        cnrs = await cnr_utils.get_cnr_data(cache_mode=cache_mode=='cache', dont_wait=dont_wait)
+        cnrs = await cnr_utils.get_cnr_data(sync_mode=cache_mode, dont_wait=dont_wait)
 
         for x in cnrs:
             self.cnr_map[x['id']] = x

glob/manager_util.py

@@ -166,8 +166,10 @@ def simple_hash(input_string):
 
     return hash_value
 
+def is_file_created_within_days(file_path, days):
+    if days is None:
+        return True
 
-def is_file_created_within_one_day(file_path):
     if not os.path.exists(file_path):
         return False
 
@@ -175,8 +177,10 @@ def is_file_created_within_one_day(file_path):
     current_time = datetime.now().timestamp()
     time_difference = current_time - file_creation_time
 
-    return time_difference <= 86400
+    return time_difference <= (days * 86400)
 
+def is_file_created_within_one_day(file_path):
+    return is_file_created_within_days(file_path, 1)
 
 async def get_data(uri, silent=False):
     if not silent:
@@ -214,12 +218,12 @@ def get_cache_path(uri):
     return os.path.join(cache_dir, cache_uri+'.json')
 
 
-def get_cache_state(uri):
+def get_cache_state(uri, expired_days=1):
     cache_uri = get_cache_path(uri)
 
     if not os.path.exists(cache_uri):
         return "not-cached"
-    elif is_file_created_within_one_day(cache_uri):
+    elif is_file_created_within_days(cache_uri, expired_days):
         return "cached"
 
     return "expired"