feat: improve prompt synchronization in Netmiko driver and replace Redis KEYS with scan_iter to avoid blocking

Author: tkspuk

netpulse/plugins/drivers/netmiko/__init__.py

@@ -422,9 +422,6 @@ def config(self, session: BaseConnection, config: list[str]) -> list[DriverExecu
                                 )
                             )
                         break
-                    finally:
-                        # Maintain prompt synchronization to handle sub-view changes
-                        session.set_base_prompt()
 
                 # 3. Post-execution operations (Commit/Save)
                 if commit := self._commit(session):

netpulse/services/manager.py

@@ -292,15 +292,17 @@ def get_node(self, node: str) -> NodeInfo | None:
 
     def get_all_nodes(self) -> list[NodeInfo]:
         """
-        Get all nodes from the redis
+        Get all nodes from the redis using non-blocking scan.
+        Guarantees unique list of NodeInfo.
         """
-        # check the map in redis
-        nodes: dict[str, str] = self.rdb.hgetall(self.node_info_map)  # type: ignore
-        if not nodes:
-            return []
-
-        # key: hostname of the node, value: node info
-        return [NodeInfo.model_validate_json(node) for node in nodes.values()]
+        # Collect into dict first to match original deduplication behavior
+        nodes_dict = {}
+        for hostname_bin, node_json in self.rdb.hscan_iter(self.node_info_map):
+            if node_json:
+                hostname = hostname_bin.decode() if isinstance(hostname_bin, bytes) else hostname_bin
+                nodes_dict[hostname] = NodeInfo.model_validate_json(node_json)
+        
+        return list(nodes_dict.values())
 
     def dispatch_rpc_job(
         self,
@@ -652,8 +654,10 @@ def get_command_list(r):
         )
 
     def _get_all_job_id(self):
-        keys: list[bytes] = self.rdb.keys(f"{Job.redis_job_namespace_prefix}*")  # type: ignore
-        return [k.decode().split(":")[-1] for k in keys]
+        # Use scan_iter instead of keys() to avoid blocking Redis main thread
+        # Add set() to ensure unique IDs if Redis returns duplicates during scanning
+        keys = self.rdb.scan_iter(match=f"{Job.redis_job_namespace_prefix}*")
+        return list(set(k.decode().split(":")[-1] for k in keys))
 
     def _get_job_id_by_status(self, state: str, q_name: str):
         """

netpulse/services/rediz.py

@@ -192,12 +192,12 @@ def unregister(self, task_id: str):
         log.info(f"Detached Task {task_id} removed from Registry.")
 
     def list_all(self) -> dict:
-        """List all registered tasks."""
+        """List all registered tasks using non-blocking scan."""
         import json
 
-        raw = self.rdb.hgetall(self.KEY)
         result = {}
-        for k, v in raw.items():
+        # Use hscan_iter to avoid blocking Redis with large registries
+        for k, v in self.rdb.hscan_iter(self.KEY):
             key = k.decode("utf-8") if isinstance(k, bytes) else k
             val = v.decode("utf-8") if isinstance(v, bytes) else v
             result[key] = json.loads(val)