Fix concurrency issue with NodePeer update

packages/syft/src/syft/client/registry.py

@@ -13,7 +13,8 @@
 
 # relative
 from ..service.metadata.node_metadata import NodeMetadataJSON
-from ..service.network.network_service import NodePeer
+from ..service.network.node_peer import NodePeer
+from ..service.network.node_peer import NodePeerConnectionStatus
 from ..service.response import SyftException
 from ..types.grid_url import GridURL
 from ..util.constants import DEFAULT_TIMEOUT
@@ -120,13 +121,40 @@ def _repr_html_(self) -> str:
         on = self.online_networks
         if len(on) == 0:
             return "(no gateways online - try syft.gateways.all_networks to see offline gateways)"
-        return pd.DataFrame(on)._repr_html_()  # type: ignore
+        df = pd.DataFrame(on)
+        total_df = pd.DataFrame(
+            [
+                [
+                    f"{len(on)} / {len(self.all_networks)} (online networks / all networks)"
+                ]
+                + [""] * (len(df.columns) - 1)
+            ],
+            columns=df.columns,
+            index=["Total"],
+        )
+        df = pd.concat([df, total_df])
+        return df._repr_html_()  # type: ignore
 
     def __repr__(self) -> str:
         on = self.online_networks
         if len(on) == 0:
             return "(no gateways online - try syft.gateways.all_networks to see offline gateways)"
-        return pd.DataFrame(on).to_string()
+        df = pd.DataFrame(on)
+        total_df = pd.DataFrame(
+            [
+                [
+                    f"{len(on)} / {len(self.all_networks)} (online networks / all networks)"
+                ]
+                + [""] * (len(df.columns) - 1)
+            ],
+            columns=df.columns,
+            index=["Total"],
+        )
+        df = pd.concat([df, total_df])
+        return df.to_string()
+
+    def __len__(self) -> int:
+        return len(self.all_networks)
 
     @staticmethod
     def create_client(network: dict[str, Any]) -> Client:
@@ -228,32 +256,25 @@ def check_network(network: dict) -> dict[Any, Any] | None:
 
     @property
     def online_domains(self) -> list[tuple[NodePeer, NodeMetadataJSON | None]]:
-        def check_domain(
-            peer: NodePeer,
-        ) -> tuple[NodePeer, NodeMetadataJSON | None] | None:
-            try:
-                guest_client = peer.guest_client
-                metadata = guest_client.metadata
-                return peer, metadata
-            except Exception as e:  # nosec
-                print(f"Error in checking domain with exception {e}")
-            return None
-
         networks = self.online_networks
 
-        # We can use a with statement to ensure threads are cleaned up promptly
-        with futures.ThreadPoolExecutor(max_workers=20) as executor:
-            # map
-            _all_online_domains = []
-            for network in networks:
+        _all_online_domains = []
+        for network in networks:
+            try:
                 network_client = NetworkRegistry.create_client(network)
-                domains: list[NodePeer] = network_client.domains.retrieve_nodes()
-                for domain in domains:
-                    self.all_domains[str(domain.id)] = domain
-                _online_domains = list(
-                    executor.map(lambda domain: check_domain(domain), domains)
-                )
-                _all_online_domains += _online_domains
+            except Exception as e:
+                print(f"Error in creating network client with exception {e}")
+                continue
+
+            domains: list[NodePeer] = network_client.domains.retrieve_nodes()
+            for domain in domains:
+                self.all_domains[str(domain.id)] = domain
+
+            _all_online_domains += [
+                (domain, domain.guest_client.metadata)
+                for domain in domains
+                if domain.ping_status == NodePeerConnectionStatus.ACTIVE
+            ]
 
         return [domain for domain in _all_online_domains if domain is not None]
 
@@ -281,13 +302,33 @@ def _repr_html_(self) -> str:
         on: list[dict[str, Any]] = self.__make_dict__()
         if len(on) == 0:
             return "(no domains online - try syft.domains.all_domains to see offline domains)"
-        return pd.DataFrame(on)._repr_html_()  # type: ignore
+        df = pd.DataFrame(on)
+        total_df = pd.DataFrame(
+            [
+                [f"{len(on)} / {len(self.all_domains)} (online domains / all domains)"]
+                + [""] * (len(df.columns) - 1)
+            ],
+            columns=df.columns,
+            index=["Total"],
+        )
+        df = pd.concat([df, total_df])
+        return df._repr_html_()  # type: ignore
 
     def __repr__(self) -> str:
         on: list[dict[str, Any]] = self.__make_dict__()
         if len(on) == 0:
             return "(no domains online - try syft.domains.all_domains to see offline domains)"
-        return pd.DataFrame(on).to_string()
+        df = pd.DataFrame(on)
+        total_df = pd.DataFrame(
+            [
+                [f"{len(on)} / {len(self.all_domains)} (online domains / all domains)"]
+                + [""] * (len(df.columns) - 1)
+            ],
+            columns=df.columns,
+            index=["Total"],
+        )
+        df = pd.concat([df, total_df])
+        return df._repr_html_()  # type: ignore
 
     def create_client(self, peer: NodePeer) -> Client:
         try:

packages/syft/src/syft/service/network/network_service.py

@@ -50,6 +50,7 @@
 from ..warnings import CRUDWarning
 from .association_request import AssociationRequestChange
 from .node_peer import NodePeer
+from .node_peer import NodePeerUpdate
 from .routes import HTTPNodeRoute
 from .routes import NodeRoute
 from .routes import NodeRouteType
@@ -87,13 +88,18 @@ def get_by_name(
     def update(
         self,
         credentials: SyftVerifyKey,
-        peer: NodePeer,
+        peer: NodePeer | NodePeerUpdate,
         has_permission: bool = False,
     ) -> Result[NodePeer, str]:
-        valid = self.check_type(peer, NodePeer)
-        if valid.is_err():
-            return Err(SyftError(message=valid.err()))
-        return super().update(credentials, peer)
+        valid_node_peer = self.check_type(peer, NodePeer)
+        valid_node_peer_update = self.check_type(peer, NodePeerUpdate)
+        if valid_node_peer.is_err() and valid_node_peer_update.is_err():
+            return Err(
+                SyftError(
+                    message=f"{type(peer)} does not match required type: NodePeer or NodePeerUpdate"
+                )
+            )
+        return super().update(credentials, peer, has_permission=has_permission)
 
     def create_or_update_peer(
         self, credentials: SyftVerifyKey, peer: NodePeer
@@ -113,9 +119,7 @@ def create_or_update_peer(
         valid = self.check_type(peer, NodePeer)
         if valid.is_err():
             return SyftError(message=valid.err())
-        existing: Result | NodePeer = self.get_by_uid(
-            credentials=credentials, uid=peer.id
-        )
+        existing = self.get_by_uid(credentials=credentials, uid=peer.id)
         if existing.is_ok() and existing.ok():
             existing = existing.ok()
             existing.update_routes(peer.node_routes)
@@ -150,8 +154,6 @@ def __init__(self, store: DocumentStore) -> None:
         self.store = store
         self.stash = NetworkStash(store=store)
 
-    # TODO: Check with MADHAVA, can we even allow guest user to introduce routes to
-    # domain nodes?
     @service_method(
         path="network.exchange_credentials_with",
         name="exchange_credentials_with",

packages/syft/src/syft/service/network/node_peer.py

@@ -17,6 +17,8 @@
 from ...service.response import SyftError
 from ...types.datetime import DateTime
 from ...types.syft_migration import migrate
+from ...types.syft_object import PartialSyftObject
+from ...types.syft_object import SYFT_OBJECT_VERSION_1
 from ...types.syft_object import SYFT_OBJECT_VERSION_2
 from ...types.syft_object import SYFT_OBJECT_VERSION_3
 from ...types.syft_object import SyftObject
@@ -71,7 +73,7 @@ class NodePeer(SyftObject):
         "name",
         "node_type",
         "admin_email",
-        "ping_status.value",
+        "ping_status",
         "ping_status_message",
         "pinged_timestamp",
     ]
@@ -117,9 +119,9 @@ def existed_route(
 
         return (False, None)
 
-    def assign_highest_priority(self, route: NodeRoute) -> NodeRoute:
+    def update_route_priority(self, route: NodeRoute) -> NodeRoute:
         """
-        Assign the new_route's to have the highest priority
+        Assign the new_route's priority to be current max + 1
 
         Args:
             route (NodeRoute): The new route whose priority is to be updated.
@@ -131,15 +133,39 @@ def assign_highest_priority(self, route: NodeRoute) -> NodeRoute:
         route.priority = current_max_priority + 1
         return route
 
+    def pick_highest_priority_route(self, oldest: bool = True) -> NodeRoute:
+        """
+        Picks the route with the highest priority from the list of node routes.
+
+        Args:
+            oldest (bool):
+                If True, picks the oldest route to have the highest priority,
+                    meaning the route with min priority value.
+                If False, picks the most recent route with the highest priority,
+                    meaning the route with max priority value.
+
+        Returns:
+            NodeRoute: The route with the highest priority.
+
+        """
+        highest_priority_route: NodeRoute = self.node_routes[-1]
+        for route in self.node_routes[:-1]:
+            if oldest:
+                if route.priority < highest_priority_route.priority:
+                    highest_priority_route = route
+            else:
+                if route.priority > highest_priority_route.priority:
+                    highest_priority_route = route
+        return highest_priority_route
+
     def update_route(self, route: NodeRoute) -> NodeRoute | None:
         """
         Update the route for the node.
         If the route already exists, return it.
-        If the route is new, assign it to have the highest priority
-        before appending it to the peer's list of node routes.
+        If the route is new, assign it to have the priority of (current_max + 1)
 
         Args:
-            route (NodeRoute): The new route to be added to the peer.
+            route (NodeRoute): The new route to be added to the peer's node route list
 
         Returns:
             NodeRoute | None: if the route already exists, return it, else returns None
@@ -148,7 +174,7 @@ def update_route(self, route: NodeRoute) -> NodeRoute | None:
         if existed:
             return route
         else:
-            new_route = self.assign_highest_priority(route)
+            new_route = self.update_route_priority(route)
             self.node_routes.append(new_route)
             return None
 
@@ -199,7 +225,7 @@ def update_existed_route_priority(
         if priority is not None:
             self.node_routes[index].priority = priority
         else:
-            self.node_routes[index].priority = self.assign_highest_priority(
+            self.node_routes[index].priority = self.update_route_priority(
                 route
             ).priority
 
@@ -223,7 +249,7 @@ def client_with_context(
 
         if len(self.node_routes) < 1:
             raise ValueError(f"No routes to peer: {self}")
-        # select the highest priority route (i.e. added or updated the latest)
+        # select the route with highest priority to connect to the peer
         final_route: NodeRoute = self.pick_highest_priority_route()
         connection: NodeConnection = route_to_connection(route=final_route)
         try:
@@ -244,7 +270,7 @@ def client_with_context(
     def client_with_key(self, credentials: SyftSigningKey) -> SyftClient | SyftError:
         if len(self.node_routes) < 1:
             raise ValueError(f"No routes to peer: {self}")
-        # select the latest added route
+
         final_route: NodeRoute = self.pick_highest_priority_route()
 
         connection = route_to_connection(route=final_route)
@@ -262,13 +288,6 @@ def guest_client(self) -> SyftClient:
     def proxy_from(self, client: SyftClient) -> SyftClient:
         return client.proxy_to(self)
 
-    def pick_highest_priority_route(self) -> NodeRoute:
-        highest_priority_route: NodeRoute = self.node_routes[-1]
-        for route in self.node_routes:
-            if route.priority > highest_priority_route.priority:
-                highest_priority_route = route
-        return highest_priority_route
-
     def delete_route(
         self, route: NodeRouteType | None = None, route_id: UID | None = None
     ) -> SyftError | None:
@@ -302,6 +321,19 @@ def delete_route(
         return None
 
 
+class NodePeerUpdate(PartialSyftObject):
+    __canonical_name__ = "NodePeerUpdate"
+    __version__ = SYFT_OBJECT_VERSION_1
+
+    id: UID
+    name: str
+    node_routes: list[NodeRouteType]
+    admin_email: str
+    ping_status: NodePeerConnectionStatus
+    ping_status_message: str
+    pinged_timestamp: DateTime
+
+
 def drop_veilid_route() -> Callable:
     def _drop_veilid_route(context: TransformContext) -> TransformContext:
         if context.output:

packages/syft/src/syft/service/network/utils.py

@@ -15,6 +15,7 @@
 from .network_service import NodePeerAssociationStatus
 from .node_peer import NodePeer
 from .node_peer import NodePeerConnectionStatus
+from .node_peer import NodePeerUpdate
 
 
 @serializable(without=["thread"])
@@ -51,42 +52,47 @@ def peer_route_heathcheck(self, context: AuthedServiceContext) -> SyftError | No
         all_peers: list[NodePeer] = result.ok()
 
         for peer in all_peers:
-            peer.pinged_timestamp = DateTime.now()
+            peer_update = NodePeerUpdate(id=peer.id)
+            peer_update.pinged_timestamp = DateTime.now()
             try:
                 peer_client = peer.client_with_context(context=context)
                 if peer_client.is_err():
                     logger.error(
                         f"Failed to create client for peer: {peer}: {peer_client.err()}"
                     )
-                    peer.ping_status = NodePeerConnectionStatus.TIMEOUT
+                    peer_update.ping_status = NodePeerConnectionStatus.TIMEOUT
                     peer_client = None
             except Exception as e:
                 logger.error(
                     f"Failed to create client for peer: {peer} with exception {e}"
                 )
-                peer.ping_status = NodePeerConnectionStatus.TIMEOUT
+
+                peer_update.ping_status = NodePeerConnectionStatus.TIMEOUT
                 peer_client = None
 
             if peer_client is not None:
                 peer_client = peer_client.ok()
                 peer_status = peer_client.api.services.network.check_peer_association(
                     peer_id=context.node.id
                 )
-                peer.ping_status = (
+                peer_update.ping_status = (
                     NodePeerConnectionStatus.ACTIVE
                     if peer_status == NodePeerAssociationStatus.PEER_ASSOCIATED
                     else NodePeerConnectionStatus.INACTIVE
                 )
                 if isinstance(peer_status, SyftError):
-                    peer.ping_status_message = (
+                    peer_update.ping_status_message = (
                         f"Error `{peer_status.message}` when pinging peer '{peer.name}'"
                     )
                 else:
-                    peer.ping_status_message = f"Peer '{peer.name}''s ping status: {peer.ping_status.value.lower()}"
+                    peer_update.ping_status_message = (
+                        f"Peer '{peer.name}''s ping status: "
+                        f"{peer_update.ping_status.value.lower()}"
+                    )
 
             result = network_stash.update(
                 credentials=context.node.verify_key,
-                peer=peer,
+                peer=peer_update,
                 has_permission=True,
             )