fix: Enhance core state management and cache synchronization

Author: x0sina

app/core/manager.py

@@ -218,16 +218,18 @@ async def _update_core_local(self, db_core_config: CoreConfig):
         await self._persist_state()
 
     async def _update_core_nats(self, db_core_config: CoreConfig):
-        # Validate core before publishing
-        # Keep local state in sync immediately, while still broadcasting via NATS.
+        # Persist local state (and KV snapshot) before broadcasting.
+        # This lets node workers refresh from KV and avoids reconnect races.
+        await self._update_core_local(db_core_config)
+
+        # Validate payload before publishing the broadcast message.
         self.validate_core(
             db_core_config.config, db_core_config.exclude_inbound_tags, db_core_config.fallbacks_inbound_tags
         )
         try:
             await self._publish_invalidation({"action": "update", "core": self._core_payload_from_db(db_core_config)})
         except Exception as exc:
             self._logger.warning(f"Failed to publish core update via NATS: {exc}")
-        await self._update_core_local(db_core_config)
 
     async def update_core(self, db_core_config: CoreConfig):
         await self._update_core_impl(db_core_config)
@@ -244,11 +246,13 @@ async def _remove_core_local(self, core_id: int):
         await self._persist_state()
 
     async def _remove_core_nats(self, core_id: int):
+        # Persist local removal (and KV snapshot) before broadcasting.
+        await self._remove_core_local(core_id)
+
         try:
             await self._publish_invalidation({"action": "remove", "core_id": core_id})
         except Exception as exc:
             self._logger.warning(f"Failed to publish core remove via NATS: {exc}")
-        await self._remove_core_local(core_id)
 
     async def remove_core(self, core_id: int):
         await self._remove_core_impl(core_id)

app/node/worker.py

@@ -8,6 +8,7 @@
 from PasarGuardNodeBridge.common.service_pb2 import User as ProtoUser
 
 from app import on_shutdown, on_startup
+from app.core.manager import core_manager
 from app.db import GetDB
 from app.db.crud.node import get_node_by_id, get_nodes
 from app.models.node import NodeCoreUpdate, NodeGeoFilesUpdate
@@ -170,12 +171,16 @@ async def _connect_node(self, data: dict):
         node_id = data.get("node_id")
         if not node_id:
             return
+        # Refresh from KV before connecting to avoid stale core cache races.
+        await core_manager._reload_from_cache()
         async with GetDB() as db:
             await self._node_operator.connect_single_node(db, node_id)
 
     async def _connect_nodes_bulk(self, data: dict):
         node_ids = data.get("node_ids")
         core_id = data.get("core_id")
+        # Refresh from KV before bulk reconnect to avoid stale core cache races.
+        await core_manager._reload_from_cache()
         async with GetDB() as db:
             if node_ids:
                 nodes, _ = await get_nodes(db, ids=node_ids)