Wait for all players to accept match before starting game

server/config.py

@@ -146,14 +146,17 @@ def __init__(self):
         # The method for choosing map pool rating
         # Can be "mean", "min", or "max"
         self.MAP_POOL_RATING_SELECTION = "mean"
-        # The maximum amount of time in seconds to wait between pops
+        # The maximum amount of time (in seconds) to wait between pops.
         self.QUEUE_POP_TIME_MAX = 90
         # The number of possible matches we would like to have when the queue
         # pops. The queue pop time will be adjusted based on the current rate of
         # players queuing to try and hit this number.
         self.QUEUE_POP_DESIRED_MATCHES = 2.5
         # How many previous queue sizes to consider
         self.QUEUE_POP_TIME_MOVING_AVG_SIZE = 5
+        # Amount of time (in seconds) that players have to accept a match
+        # before it will time out.
+        self.MATCH_OFFER_TIME = 20
 
         self._defaults = {
             key: value for key, value in vars(self).items() if key.isupper()

server/ladder_service/ladder_service.py

@@ -6,8 +6,10 @@
 import random
 import re
 import statistics
+import weakref
 from collections import defaultdict
-from typing import Awaitable, Callable, Optional
+from datetime import timedelta
+from typing import Awaitable, Callable, Iterable, Optional
 
 import aiocron
 import humanize
@@ -43,11 +45,14 @@
 from server.matchmaker import (
     MapPool,
     MatchmakerQueue,
+    MatchOffer,
+    OfferTimeoutError,
     OnMatchedCallback,
     Search
 )
 from server.metrics import MatchLaunch
 from server.players import Player, PlayerState
+from server.timing import datetime_now
 from server.types import GameLaunchOptions, Map, NeroxisGeneratedMap
 
 
@@ -71,6 +76,7 @@ def __init__(
         self.violation_service = violation_service
 
         self._searches: dict[Player, dict[str, Search]] = defaultdict(dict)
+        self._match_offers = weakref.WeakValueDictionary()
         self._allow_new_searches = True
 
     async def initialize(self) -> None:
@@ -462,13 +468,60 @@ def on_match_found(
 
                 self._clear_search(player, queue.name)
 
-            asyncio.create_task(self.start_game(s1.players, s2.players, queue))
+            asyncio.create_task(self.confirm_match(s1, s2, queue))
         except Exception:
             self._logger.exception(
                 "Error processing match between searches %s, and %s",
                 s1, s2
             )
 
+    async def confirm_match(
+        self,
+        s1: Search,
+        s2: Search,
+        queue: MatchmakerQueue
+    ):
+        try:
+            all_players = s1.players + s2.players
+
+            offer = self.create_match_offer(all_players)
+            offer.write_broadcast_update()
+            await offer.wait_ready()
+        except OfferTimeoutError:
+            unready_players = list(offer.get_unready_players())
+            assert unready_players, "Unready players should be non-empty if offer timed out"
+
+            self._logger.info(
+                "Match failed to start. Some players did not ready up in time: %s",
+                [player.login for player in unready_players]
+            )
+            # TODO: Refactor duplication
+            msg = {"command": "match_cancelled"}
+            for player in all_players:
+                if player.state == PlayerState.STARTING_AUTOMATCH:
+                    player.state = PlayerState.IDLE
+                player.write_message(msg)
+
+                # TODO: Unmatch and return to queue
+
+            self.violation_service.register_violations(unready_players)
+
+        await self.start_game(s1.players, s2.players, queue)
+
+    def create_match_offer(self, players: Iterable[Player]):
+        offer = MatchOffer(
+            players,
+            datetime_now() + timedelta(seconds=config.MATCH_OFFER_TIME)
+        )
+        for player in players:
+            self._match_offers[player] = offer
+        return offer
+
+    def ready_player(self, player: Player):
+        offer = self._match_offers.get(player)
+        if offer is not None:
+            offer.ready_player(player)
+
     def start_game(
         self,
         team1: list[Player],

server/lobbyconnection.py

@@ -951,11 +951,7 @@ async def command_game_host(self, message):
         await self.launch_game(game, is_host=True)
 
     async def command_match_ready(self, message):
-        """
-        Replace with full implementation when implemented in client, see:
-        https://github.com/FAForever/downlords-faf-client/issues/1783
-        """
-        pass
+        self.ladder_service.ready_player(self.player)
 
     async def launch_game(
         self,

server/matchmaker/matchmaker_queue.py

@@ -175,7 +175,7 @@ async def find_matches(self) -> None:
         proposed_matches, unmatched_searches = await loop.run_in_executor(
             None,
             self.matchmaker.find,
-            searches,
+            (search for search in searches if not search.done()),
             self.team_size,
             self.rating_peak,
         )

server/matchmaker/search.py

@@ -226,6 +226,9 @@ def match(self, other: "Search"):
                 )
         self._match.set_result(other)
 
+    def unmatch(self):
+        self._match = asyncio.Future()
+
     async def await_match(self):
         """
         Wait for this search to complete

tests/integration_tests/test_game.py

@@ -155,6 +155,35 @@ async def start_search(proto, queue_name="ladder1v1"):
     )
 
 
+async def accept_match(proto, timeout=5):
+    await read_until_command(proto, "match_info", timeout=timeout)
+    await proto.send_message({"command": "match_ready"})
+    return await read_until_command(
+        proto,
+        "match_info",
+        ready=True,
+        timeout=10
+    )
+
+
+async def read_until_match(proto, timeout=30):
+    # If the players did not match, this will fail due to a timeout error
+    msg = await read_until_command(proto, "match_found", timeout=timeout)
+    # Accept match
+    match_info = await read_until_command(proto, "match_info", timeout=timeout)
+    await proto.send_message({"command": "match_ready"})
+    # Wait for all players to be ready. This could be the same as the message
+    # notifying us that our match ready was received so we can't use
+    # `accept_match` here.
+    await read_until_command(
+        proto,
+        "match_info",
+        players_ready=match_info["players_total"],
+        timeout=10
+    )
+    return msg
+
+
 async def queue_player_for_matchmaking(
     user,
     lobby_server,
@@ -182,10 +211,6 @@ async def queue_players_for_matchmaking(
         queue_name
     )
 
-    # If the players did not match, this will fail due to a timeout error
-    await read_until_command(proto1, "match_found", timeout=30)
-    await read_until_command(proto2, "match_found")
-
     return player1_id, proto1, player2_id, proto2
 
 
@@ -203,18 +228,12 @@ async def queue_temp_players_for_matchmaking(
         tmp_user(queue_name)
         for _ in range(num_players)
     ])
-    protos = await asyncio.gather(*[
+    responses = await asyncio.gather(*[
         queue_player_for_matchmaking(user, lobby_server, queue_name)
         for user in users
     ])
 
-    # If the players did not match, this will fail due to a timeout error
-    await asyncio.gather(*[
-        read_until_command(proto, "match_found", timeout=30)
-        for _, proto in protos
-    ])
-
-    return protos
+    return (proto for _, proto in responses)
 
 
 async def get_player_ratings(proto, *names, rating_type="global"):
@@ -674,9 +693,11 @@ async def test_ladder_game_draw_bug(lobby_server, database):
     instead of a draw.
     """
     player1_id, proto1, player2_id, proto2 = await queue_players_for_matchmaking(lobby_server)
+    protos = (proto1, proto2)
 
+    await asyncio.gather(*[read_until_match(proto) for proto in protos])
     msg1, msg2 = await asyncio.gather(*[
-        client_response(proto) for proto in (proto1, proto2)
+        client_response(proto) for proto in protos
     ])
     game_id = msg1["uid"]
     army1 = msg1["map_position"]
@@ -690,7 +711,7 @@ async def test_ladder_game_draw_bug(lobby_server, database):
             (player_id, "Faction", msg["faction"]),
             (player_id, "Color", msg["map_position"]),
         )
-    for proto in (proto1, proto2):
+    for proto in protos:
         await proto.send_message({
             "target": "game",
             "command": "GameState",
@@ -707,14 +728,14 @@ async def test_ladder_game_draw_bug(lobby_server, database):
         [army1, "score 1"],
         [army2, "defeat -10"]
     ):
-        for proto in (proto1, proto2):
+        for proto in protos:
             await proto.send_message({
                 "target": "game",
                 "command": "GameResult",
                 "args": result
             })
 
-    for proto in (proto1, proto2):
+    for proto in protos:
         await proto.send_message({
             "target": "game",
             "command": "GameEnded",
@@ -757,15 +778,16 @@ async def test_ladder_game_draw_bug(lobby_server, database):
             assert row.score == 0
 
 
-@fast_forward(15)
+@fast_forward(70)
 async def test_ladder_game_not_joinable(lobby_server):
     """
     We should not be able to join AUTO_LOBBY games using the `game_join` command.
     """
     _, _, test_proto = await connect_and_sign_in(
         ("test", "test_password"), lobby_server
     )
-    _, proto1, _, _ = await queue_players_for_matchmaking(lobby_server)
+    _, proto1, _, proto2 = await queue_players_for_matchmaking(lobby_server)
+    await asyncio.gather(*[read_until_match(proto) for proto in (proto1, proto2)])
     await read_until_command(test_proto, "game_info")
 
     msg = await read_until_command(proto1, "game_launch")

tests/integration_tests/test_matchmaker.py

@@ -23,6 +23,7 @@
     queue_players_for_matchmaking,
     queue_temp_players_for_matchmaking,
     read_until_launched,
+    read_until_match,
     send_player_options
 )
 
@@ -31,6 +32,7 @@
 async def test_game_launch_message(lobby_server):
     _, proto1, _, proto2 = await queue_players_for_matchmaking(lobby_server)
 
+    await asyncio.gather(*[read_until_match(proto) for proto in (proto1, proto2)])
     msg1 = await read_until_command(proto1, "game_launch")
     await open_fa(proto1)
     msg2 = await read_until_command(proto2, "game_launch")
@@ -67,6 +69,7 @@ async def test_game_launch_message_map_generator(lobby_server):
         queue_name="neroxis1v1"
     )
 
+    await asyncio.gather(*[read_until_match(proto) for proto in (proto1, proto2)])
     msg1 = await read_until_command(proto1, "game_launch")
     await open_fa(proto1)
     msg2 = await read_until_command(proto2, "game_launch")
@@ -87,6 +90,7 @@ async def test_game_launch_message_game_options(lobby_server, tmp_user):
         queue_name="gameoptions"
     )
 
+    await asyncio.gather(*[read_until_match(proto) for proto in protos])
     msgs = await asyncio.gather(*[
         client_response(proto) for _, proto in protos
     ])
@@ -103,6 +107,7 @@ async def test_game_launch_message_game_options(lobby_server, tmp_user):
 async def test_game_matchmaking_start(lobby_server, database):
     host_id, host, guest_id, guest = await queue_players_for_matchmaking(lobby_server)
 
+    await asyncio.gather(*[read_until_match(proto) for proto in (host, guest)])
     # The player that queued last will be the host
     msg = await read_until_command(host, "game_launch")
     game_id = msg["uid"]
@@ -166,8 +171,9 @@ async def test_game_matchmaking_start(lobby_server, database):
 
 @fast_forward(15)
 async def test_game_matchmaking_start_while_matched(lobby_server):
-    _, proto1, _, _ = await queue_players_for_matchmaking(lobby_server)
+    _, proto1, _, proto2 = await queue_players_for_matchmaking(lobby_server)
 
+    await asyncio.gather(*[read_until_match(proto) for proto in (proto1, proto2)])
     # Trying to queue again after match was found should generate an error
     await proto1.send_message({
         "command": "game_matchmaking",
@@ -182,6 +188,7 @@ async def test_game_matchmaking_start_while_matched(lobby_server):
 async def test_game_matchmaking_timeout(lobby_server, game_service):
     _, proto1, _, proto2 = await queue_players_for_matchmaking(lobby_server)
 
+    await asyncio.gather(*[read_until_match(proto) for proto in (proto1, proto2)])
     msg1 = await idle_response(proto1, timeout=120)
     await read_until_command(proto2, "match_cancelled", timeout=120)
     await read_until_command(proto1, "match_cancelled", timeout=120)
@@ -229,6 +236,7 @@ async def test_game_matchmaking_timeout(lobby_server, game_service):
 async def test_game_matchmaking_timeout_guest(lobby_server, game_service):
     _, proto1, _, proto2 = await queue_players_for_matchmaking(lobby_server)
 
+    await asyncio.gather(*[read_until_match(proto) for proto in (proto1, proto2)])
     msg1, msg2 = await asyncio.gather(
         client_response(proto1),
         client_response(proto2)
@@ -305,23 +313,39 @@ async def test_game_matchmaking_cancel(lobby_server):
         await read_until_command(proto, "search_info", timeout=5)
 
 
-@fast_forward(50)
+@fast_forward(120)
 async def test_game_matchmaking_disconnect(lobby_server):
     _, proto1, _, proto2 = await queue_players_for_matchmaking(lobby_server)
+
     # One player disconnects before the game has launched
+    await asyncio.gather(*[read_until_match(proto) for proto in (proto1, proto2)])
     await proto1.close()
 
     msg = await read_until_command(proto2, "match_cancelled", timeout=120)
 
     assert msg == {"command": "match_cancelled", "game_id": 41956}
 
 
+@fast_forward(120)
+async def test_game_matchmaking_disconnect_no_accept(lobby_server):
+    _, proto1, _, proto2 = await queue_players_for_matchmaking(lobby_server)
+
+    # One player disconnects before the game has launched
+    await read_until_command(proto1, "match_found", timeout=30)
+    await proto1.close()
+
+    msg = await read_until_command(proto2, "match_cancelled", timeout=120)
+
+    assert msg == {"command": "match_cancelled"}
+
+
 @pytest.mark.flaky
 @fast_forward(130)
 async def test_game_matchmaking_close_fa_and_requeue(lobby_server):
     _, proto1, _, proto2 = await queue_players_for_matchmaking(lobby_server)
 
-    _, _ = await asyncio.gather(
+    await asyncio.gather(*[read_until_match(proto) for proto in (proto1, proto2)])
+    await asyncio.gather(
         client_response(proto1),
         client_response(proto2)
     )
@@ -365,6 +389,10 @@ async def test_anti_map_repetition(lobby_server):
     for _ in range(20):
         ret = await queue_players_for_matchmaking(lobby_server)
         player1_id, proto1, player2_id, proto2 = ret
+        await asyncio.gather(
+            read_until_match(proto1),
+            read_until_match(proto2)
+        )
         msg1, msg2 = await asyncio.gather(
             client_response(proto1),
             client_response(proto2)