Implement EstablishedPeers messages

server/game_connection_matrix.py

@@ -0,0 +1,27 @@
+from collections import defaultdict
+
+
+class ConnectionMatrix:
+    def __init__(self, established_peers: dict[int, set[int]]):
+        self.established_peers = established_peers
+
+    def get_unconnected_peer_ids(self) -> set[int]:
+        unconnected_peer_ids: set[int] = set()
+
+        # Group players by number of connected peers
+        players_by_num_peers = defaultdict(list)
+        for player_id, peer_ids in self.established_peers.items():
+            players_by_num_peers[len(peer_ids)].append((player_id, peer_ids))
+
+        # Mark players with least number of connections as unconnected if they
+        # don't meet the connection threshold. Each time a player is marked as
+        # 'unconnected', remaining players need 1 less connection to be
+        # considered connected.
+        connected_peers = dict(self.established_peers)
+        for num_connected, peers in sorted(players_by_num_peers.items()):
+            if num_connected < len(connected_peers) - 1:
+                for player_id, peer_ids in peers:
+                    unconnected_peer_ids.add(player_id)
+                    del connected_peers[player_id]
+
+        return unconnected_peer_ids

server/gameconnection.py

@@ -6,7 +6,7 @@
 import contextlib
 import json
 import logging
-from typing import Any
+from typing import Any, Optional
 
 from sqlalchemy import select
 
@@ -62,6 +62,10 @@
         self.player = player
         player.game_connection = self  # Set up weak reference to self
         self.game = game
+        # None if the EstablishedPeers message is not implemented by the game
+        # version/mode used by the player. For instance, matchmaker might have
+        # it, but custom games might not.
+        self.established_peer_ids: Optional[set[int]] = None
 
         self.setup_timeout = setup_timeout
 
@@ -509,15 +513,20 @@
 
     async def handle_launch_status(self, status: str):
         """
-        Currently is sent with status `Rejected` if a matchmaker game failed
-        to start due to players using differing game settings.
+        Represents the launch status of a peer.
+
+        # Params
+        - `status`: One of "Unknown" | "Connecting" | "Missing local peers" |
+            "Rejoining" | "Ready" | "Ejected" | "Rejected" | "Failed"
         """
         pass
 
     async def handle_bottleneck(self, *args: list[Any]):
         """
         Not sure what this command means. This is currently unused but
         included for documentation purposes.
+
+        Example args: ["ack", "23381", "232191", "64218.4"]
         """
         pass
 
@@ -547,6 +556,31 @@
         """
         pass
 
+    async def handle_established_peer(self, peer_id: str):
+        """
+        Sent by the lobby when the player connectes to another peer. Can be
+        send multiple times.
+
+        # Params
+        - `peer_id`: The identifier of the peer that this connection received
+            the message from
+        """
+        if self.established_peer_ids is None:
+            self.established_peer_ids = set()
+
+        self.established_peer_ids.add(int(peer_id))
+
+    async def handle_disconnected_peer(self, peer_id: str):
+        """
+        Sent by the lobby when a player disconnects from a peer. This can happen
+        when a peer is rejoining in which case that peer will have reported a
+        "Rejoining" status, or if the peer has exited the game.
+        """
+        if self.established_peer_ids is None:
+            self.established_peer_ids = set()
+
+        self.established_peer_ids.discard(int(peer_id))
+
     def _mark_dirty(self):
         if self.game:
             self.game_service.mark_dirty(self.game)
@@ -623,26 +657,28 @@
 
 
 COMMAND_HANDLERS = {
-    "AIOption":             GameConnection.handle_ai_option,
-    "Bottleneck":           GameConnection.handle_bottleneck,
-    "BottleneckCleared":    GameConnection.handle_bottleneck_cleared,
-    "Chat":                 GameConnection.handle_chat,
-    "ClearSlot":            GameConnection.handle_clear_slot,
-    "Desync":               GameConnection.handle_desync,
-    "Disconnected":         GameConnection.handle_disconnected,
-    "EnforceRating":        GameConnection.handle_enforce_rating,
-    "GameEnded":            GameConnection.handle_game_ended,
-    "GameFull":             GameConnection.handle_game_full,
-    "GameMods":             GameConnection.handle_game_mods,
-    "GameOption":           GameConnection.handle_game_option,
-    "GameResult":           GameConnection.handle_game_result,
-    "GameState":            GameConnection.handle_game_state,
-    "IceMsg":               GameConnection.handle_ice_message,
-    "JsonStats":            GameConnection.handle_json_stats,
-    "LaunchStatus":         GameConnection.handle_launch_status,
-    "OperationComplete":    GameConnection.handle_operation_complete,
-    "PlayerOption":         GameConnection.handle_player_option,
-    "Rehost":               GameConnection.handle_rehost,
-    "TeamkillHappened":     GameConnection.handle_teamkill_happened,
-    "TeamkillReport":       GameConnection.handle_teamkill_report,
+    "AIOption":             GameConnection.handle_ai_option,            # Lobby message
+    "Bottleneck":           GameConnection.handle_bottleneck,           # Lobby/game message
+    "BottleneckCleared":    GameConnection.handle_bottleneck_cleared,   # Lobby/game message
+    "Chat":                 GameConnection.handle_chat,                 # Lobby message
+    "ClearSlot":            GameConnection.handle_clear_slot,           # Lobby message
+    "Desync":               GameConnection.handle_desync,               # Game message
+    "Disconnected":         GameConnection.handle_disconnected,         # Lobby message
+    "EnforceRating":        GameConnection.handle_enforce_rating,       # Game message
+    "EstablishedPeer":      GameConnection.handle_established_peer,     # Lobby message
+    "DisconnectedPeer":     GameConnection.handle_disconnected_peer,    # Lobby message
+    "GameEnded":            GameConnection.handle_game_ended,           # Game message
+    "GameFull":             GameConnection.handle_game_full,            # Lobby message
+    "GameMods":             GameConnection.handle_game_mods,            # Lobby message
+    "GameOption":           GameConnection.handle_game_option,          # Lobby message
+    "GameResult":           GameConnection.handle_game_result,          # Game message
+    "GameState":            GameConnection.handle_game_state,           # Lobby/game message
+    "IceMsg":               GameConnection.handle_ice_message,          # Lobby/Game message
+    "JsonStats":            GameConnection.handle_json_stats,           # Game message
+    "LaunchStatus":         GameConnection.handle_launch_status,        # Lobby message
+    "OperationComplete":    GameConnection.handle_operation_complete,   # Coop message
+    "PlayerOption":         GameConnection.handle_player_option,        # Lobby message
+    "Rehost":               GameConnection.handle_rehost,               # Game message
+    "TeamkillHappened":     GameConnection.handle_teamkill_happened,    # Game message
+    "TeamkillReport":       GameConnection.handle_teamkill_report,      # Game message
 }

server/games/game.py

@@ -17,6 +17,7 @@
     game_stats,
     matchmaker_queue_game
 )
+from server.game_connection_matrix import ConnectionMatrix
 from server.games.game_results import (
     ArmyOutcome,
     ArmyReportedOutcome,
@@ -211,13 +212,41 @@
 
     def get_connected_players(self) -> list[Player]:
         """
-        Get a collection of all players currently connected to the game.
+        Get a collection of all players currently connected to the host.
         """
         return [
             player for player in self._connections.keys()
             if player.id in self._configured_player_ids
         ]
 
+    def get_unconnected_players_from_peer_matrix(
+        self,
+    ) -> Optional[list[Player]]:
+        """
+        Get a list of players who are not fully connected to the game based on
+        the established peers matrix if possible. The EstablishedPeers messages
+        might not be implemented by the game in which case this returns None.
+        """
+        if any(
+            conn.established_peer_ids is None
+            for conn in self._connections.values()
+        ):
+            return None
+
+        matrix = ConnectionMatrix(
+            established_peers={
+                player.id: conn.established_peer_ids
+                for player, conn in self._connections.items()
+            }
+        )
+        unconnected_peer_ids = matrix.get_unconnected_peer_ids()
+
+        return [
+            player
+            for player in self._connections.keys()
+            if player.id in unconnected_peer_ids
+        ]
+
     def _is_observer(self, player: Player) -> bool:
         army = self.get_player_option(player.id, "Army")
         return army is None or army < 0

server/ladder_service/ladder_service.py

@@ -667,6 +667,12 @@
         try:
             await game.wait_launched(60 + 10 * len(guests))
         except asyncio.TimeoutError:
+            unconnected_players = game.get_unconnected_players_from_peer_matrix()
+            if unconnected_players is not None:
+                raise NotConnectedError(unconnected_players)
+
+            # If the connection matrix was not available, fall back to looking
+            # at who was connected to the host only.
             connected_players = game.get_connected_players()
             raise NotConnectedError([
                 player for player in guests