work around data race during game join

src/Client.hh

@@ -191,6 +191,12 @@ struct Client : public std::enable_shared_from_this<Client> {
   std::weak_ptr<Episode3::Tournament::Team> ep3_tournament_team;
   std::shared_ptr<Episode3::BattleRecord> ep3_prev_battle_record;
   std::shared_ptr<const Menu> last_menu_sent;
+  struct JoinCommand {
+    uint16_t command;
+    uint32_t flag;
+    std::string data;
+  };
+  std::unique_ptr<std::deque<JoinCommand>> game_join_command_queue;
 
   // Miscellaneous (used by chat commands)
   uint32_t next_exp_value; // next EXP value to give

src/CommandFormats.hh

@@ -4446,7 +4446,7 @@ struct G_CreateTelepipe_6x68 {
   G_UnusedHeader header;
   le_uint16_t client_id2 = 0;
   le_uint16_t unknown_a1 = 0;
-  le_uint16_t unused1 = 0;
+  le_uint16_t unknown_a2 = 0;
   parray<uint8_t, 2> unused2;
   le_float x = 0.0f;
   le_float y = 0.0f;
@@ -4504,8 +4504,26 @@ struct G_SyncObjectState_6x6C_Entry_Decompressed {
 } __packed__;
 
 // 6x6D: Sync item state (used while loading into game; same header format as 6E)
+// Internal name: RcvItemCondition
 // Compressed format is the same as 6x6B.
 
+// There is a bug in the client that can cause desync between players' item IDs
+// if the 6x6D command is sent too quickly. 6x6D is the first command sent by
+// the leader when a player joins a game, so the joining player often receives
+// it immediately after the 64 command. One of the first steps of the game join
+// process is to reset the next item ID variables for each player in the game,
+// which under normal operation, the 6x6D command overwrites with the values in
+// the next_item_id_per_player array. However, the loading process doesn't
+// actually start until the next frame after the 64 command is received, so if
+// the 6x6D command is received on the same frame as the 64 command, it will
+// set the next item ID variables correctly and the loading process will then
+// clear all of them on the next frame. The client will then assign its own
+// inventory item IDs based on the default base item ID, which will result in
+// incorrect IDs if another player had previously been in the game in the same
+// slot. To prevent this bug, we delay the 6x6D command and everything that
+// depends on it until the client responds to a ping command (sent along with
+// the 64 command).
+
 struct G_SyncItemState_6x6D_Decompressed {
   // TODO: Verify this format on DC and PC. It appears correct for GC and BB.
   // Note: 16 vs. 15 is not a bug here - there really is an extra field in the

src/ReceiveCommands.cc

@@ -329,6 +329,18 @@ static void on_1D(shared_ptr<Client> c, uint16_t, uint32_t, string&) {
     double ping_ms = static_cast<double>(ping_usecs) / 1000.0;
     send_text_message_printf(c, "To server: %gms", ping_ms);
   }
+
+  // See the comment on the 6x6D command in CommandFormats.hh to understand why
+  // we do this.
+  if (c->game_join_command_queue) {
+    c->log.info("Sending %zu queued command(s)", c->game_join_command_queue->size());
+    while (!c->game_join_command_queue->empty()) {
+      const auto& cmd = c->game_join_command_queue->front();
+      send_command(c, cmd.command, cmd.flag, cmd.data);
+      c->game_join_command_queue->pop_front();
+    }
+    c->game_join_command_queue.reset();
+  }
 }
 
 static void on_05_XB(shared_ptr<Client> c, uint16_t, uint32_t, string&) {

src/ReceiveSubcommands.cc

@@ -132,6 +132,62 @@ static void on_forward_check_game(shared_ptr<Client> c, uint8_t command, uint8_t
   forward_subcommand(c, command, flag, data, size);
 }
 
+template <typename CmdT>
+static void send_or_enqueue_joining_player_command(shared_ptr<Client> c, uint8_t command, uint8_t flag, const CmdT& data) {
+  if (c->game_join_command_queue) {
+    c->log.info("Client not ready to receive join commands; adding to queue");
+    auto cmd = c->game_join_command_queue->emplace_back();
+    cmd.command = command;
+    cmd.flag = flag;
+    cmd.data.assign(reinterpret_cast<const char*>(&data), sizeof(data));
+  } else {
+    send_command(c, command, flag, &data, sizeof(data));
+  }
+}
+
+static void send_or_enqueue_joining_player_command(shared_ptr<Client> c, uint8_t command, uint8_t flag, string&& data) {
+  if (c->game_join_command_queue) {
+    c->log.info("Client not ready to receive join commands; adding to queue");
+    auto cmd = c->game_join_command_queue->emplace_back();
+    cmd.command = command;
+    cmd.flag = flag;
+    cmd.data = std::move(data);
+  } else {
+    send_command(c, command, flag, data.data(), data.size());
+  }
+}
+
+static void send_or_enqueue_joining_player_command(shared_ptr<Client> c, uint8_t command, uint8_t flag, const void* data, size_t size) {
+  if (c->game_join_command_queue) {
+    c->log.info("Client not ready to receive join commands; adding to queue");
+    auto cmd = c->game_join_command_queue->emplace_back();
+    cmd.command = command;
+    cmd.flag = flag;
+    cmd.data.assign(reinterpret_cast<const char*>(data), size);
+  } else {
+    send_command(c, command, flag, data, size);
+  }
+}
+
+static void on_forward_check_game_loading(shared_ptr<Client> c, uint8_t command, uint8_t flag, const void* data, size_t size) {
+  auto l = c->require_lobby();
+  if (!l->is_game() || !l->any_client_loading()) {
+    return;
+  }
+  if (command_is_private(command)) {
+    auto target = l->clients.at(flag);
+    if (target) {
+      send_or_enqueue_joining_player_command(target, command, flag, data, size);
+    }
+  } else {
+    for (auto lc : l->clients) {
+      if (lc) {
+        send_or_enqueue_joining_player_command(lc, command, flag, data, size);
+      }
+    }
+  }
+}
+
 static void on_forward_sync_joining_player_state(shared_ptr<Client> c, uint8_t command, uint8_t flag, const void* data, size_t size) {
   auto l = c->require_lobby();
   if (!l->is_game() || !l->any_client_loading()) {
@@ -150,7 +206,7 @@ static void on_forward_sync_joining_player_state(shared_ptr<Client> c, uint8_t c
     print_data(stderr, decompressed);
   }
 
-  forward_subcommand(c, command, flag, data, size);
+  on_forward_check_game_loading(c, command, flag, data, size);
 }
 
 static void on_sync_joining_player_item_state(shared_ptr<Client> c, uint8_t command, uint8_t flag, const void* data, size_t size) {
@@ -164,26 +220,26 @@ static void on_sync_joining_player_item_state(shared_ptr<Client> c, uint8_t comm
   if (!command_is_private(command)) {
     throw runtime_error("6x6D sent via public command");
   }
+  if (flag >= l->max_clients) {
+    return;
+  }
+  auto target = l->clients[flag];
+  if (!target) {
+    return;
+  }
 
   // For non-V3 versions, just forward the data verbatim. For V3, we need to
   // byteswap mags' data2 fields if exactly one of the sender and recipient is
   // PSO GC
   bool sender_is_gc = (c->version() == GameVersion::GC);
   if (!sender_is_gc && (c->version() != GameVersion::XB)) {
-    forward_subcommand(c, command, flag, data, size);
+    send_or_enqueue_joining_player_command(target, command, flag, data, size);
 
   } else {
-    if (flag >= l->max_clients) {
-      return;
-    }
-    auto target = l->clients[flag];
-    if (!target) {
-      return;
-    }
     bool target_is_gc = (target->version() == GameVersion::GC);
 
     if (target_is_gc == sender_is_gc) {
-      send_command(target, command, flag, data, size);
+      send_or_enqueue_joining_player_command(target, command, flag, data, size);
 
     } else {
       const auto& cmd = check_size_t<G_SyncGameStateHeader_6x6B_6x6C_6x6D_6x6E>(data, size, 0xFFFF);
@@ -239,10 +295,10 @@ static void on_sync_joining_player_item_state(shared_ptr<Client> c, uint8_t comm
         c->log.info("Byteswapped and recompressed item sync data (%zX bytes)", out_compressed_data.size());
       }
 
-      vector<pair<const void*, size_t>> blocks;
-      blocks.emplace_back(make_pair(&out_cmd, sizeof(out_cmd)));
-      blocks.emplace_back(make_pair(out_compressed_data.data(), out_compressed_data.size()));
-      send_command(target, command, flag, blocks);
+      StringWriter w;
+      w.write(&out_cmd, sizeof(out_cmd));
+      w.write(out_compressed_data);
+      send_or_enqueue_joining_player_command(target, command, flag, std::move(w.str()));
     }
   }
 }
@@ -284,7 +340,7 @@ static void on_sync_joining_player_disp_and_inventory(
   bool sender_is_gc = (c->version() == GameVersion::GC);
   bool target_is_gc = (target->version() == GameVersion::GC);
   if (target_is_gc == sender_is_gc) {
-    send_command(target, command, flag, data, size);
+    send_or_enqueue_joining_player_command(target, command, flag, data, size);
 
   } else if (sender_is_gc) {
     // Convert GC command to XB command
@@ -300,7 +356,7 @@ static void on_sync_joining_player_disp_and_inventory(
       out_cmd.inventory.items[z].data.decode_for_version(GameVersion::GC);
       out_cmd.inventory.items[z].data.encode_for_version(GameVersion::XB, s->item_parameter_table_for_version(GameVersion::XB));
     }
-    send_command_t(target, command, flag, out_cmd);
+    send_or_enqueue_joining_player_command(target, command, flag, out_cmd);
 
   } else {
     // Convert XB command to GC command
@@ -312,18 +368,10 @@ static void on_sync_joining_player_disp_and_inventory(
       out_cmd.inventory.items[z].data.decode_for_version(GameVersion::XB);
       out_cmd.inventory.items[z].data.encode_for_version(GameVersion::GC, s->item_parameter_table_for_version(GameVersion::GC));
     }
-    send_command(target, command, flag, &out_cmd, sizeof(G_SyncPlayerDispAndInventory_DC_PC_GC_6x70));
+    send_or_enqueue_joining_player_command(target, command, flag, out_cmd);
   }
 }
 
-static void on_forward_check_game_loading(shared_ptr<Client> c, uint8_t command, uint8_t flag, const void* data, size_t size) {
-  auto l = c->require_lobby();
-  if (!l->is_game() || !l->any_client_loading()) {
-    return;
-  }
-  forward_subcommand(c, command, flag, data, size);
-}
-
 static void on_forward_check_size_client(shared_ptr<Client> c, uint8_t command, uint8_t flag, const void* data, size_t size) {
   const auto& cmd = check_size_t<G_ClientIDHeader>(data, size, 0xFFFF);
   if (cmd.client_id != c->lobby_client_id) {

src/SendCommands.cc

@@ -1691,6 +1691,10 @@ void send_join_game(shared_ptr<Client> c, shared_ptr<Lobby> l) {
     default:
       throw logic_error("invalid game version");
   }
+
+  c->log.info("Creating game join command queue");
+  c->game_join_command_queue.reset(new deque<Client::JoinCommand>());
+  send_command(c, 0x1D, 0x00);
 }
 
 template <typename LobbyDataT, typename DispDataT, typename RecordsT, bool UseLanguageMarkerInName>