Propagate peer ID change to outgoing reliables (#15680)

Otherwise a desync could ocurr since the server does strict checking.
fixes #15627

src/network/mtp/impl.cpp

@@ -53,6 +53,15 @@ u16 BufferedPacket::getSeqnum() const
 	return readU16(&data[BASE_HEADER_SIZE + 1]);
 }
 
+void BufferedPacket::setSenderPeerId(session_t id)
+{
+	if (size() < BASE_HEADER_SIZE) {
+		assert(false); // should never happen
+		return;
+	}
+	writeU16(&data[4], id);
+}
+
 BufferedPacketPtr makePacket(const Address &address, const SharedBuffer<u8> &data,
 		u32 protocol_id, session_t sender_peer_id, u8 channel)
 {
@@ -337,6 +346,13 @@ void ReliablePacketBuffer::insert(BufferedPacketPtr &p_ptr, u16 next_expected)
 	m_oldest_non_answered_ack = m_list.front()->getSeqnum();
 }
 
+void ReliablePacketBuffer::fixPeerId(session_t new_id)
+{
+	MutexAutoLock listlock(m_list_mutex);
+	for (auto &packet : m_list)
+		packet->setSenderPeerId(new_id);
+}
+
 void ReliablePacketBuffer::incrementTimeouts(float dtime)
 {
 	MutexAutoLock listlock(m_list_mutex);
@@ -569,6 +585,13 @@ ConnectionCommandPtr ConnectionCommand::resend_one(session_t peer_id)
 	return c;
 }
 
+ConnectionCommandPtr ConnectionCommand::peer_id_set(session_t own_peer_id)
+{
+	auto c = create(CONNCMD_PEER_ID_SET);
+	c->peer_id = own_peer_id;
+	return c;
+}
+
 ConnectionCommandPtr ConnectionCommand::send(session_t peer_id, u8 channelnum,
 	NetworkPacket *pkt, bool reliable)
 {
@@ -1615,6 +1638,14 @@ void Connection::DisconnectPeer(session_t peer_id)
 	putCommand(ConnectionCommand::disconnect_peer(peer_id));
 }
 
+void Connection::SetPeerID(session_t id)
+{
+	m_peer_id = id;
+	// fix peer id in existing queued reliable packets
+	if (id != PEER_ID_INEXISTENT)
+		putCommand(ConnectionCommand::peer_id_set(id));
+}
+
 void Connection::doResendOne(session_t peer_id)
 {
 	assert(peer_id != PEER_ID_INEXISTENT);

src/network/mtp/impl.h

@@ -256,7 +256,7 @@ protected:
 	UDPPeer*  createServerPeer(const Address& sender);
 	bool deletePeer(session_t peer_id, bool timeout);
 
-	void SetPeerID(session_t id) { m_peer_id = id; }
+	void SetPeerID(session_t id);
 
 	void doResendOne(session_t peer_id);
 

src/network/mtp/internal.h

@@ -187,6 +187,7 @@ struct BufferedPacket {
 	DISABLE_CLASS_COPY(BufferedPacket)
 
 	u16 getSeqnum() const;
+	void setSenderPeerId(session_t id);
 
 	inline size_t size() const { return m_data.size(); }
 
@@ -250,6 +251,8 @@ public:
 	BufferedPacketPtr popFirst();
 	BufferedPacketPtr popSeqnum(u16 seqnum);
 	void insert(BufferedPacketPtr &p_ptr, u16 next_expected);
+	/// Adjusts the sender peer ID for all packets
+	void fixPeerId(session_t id);
 
 	void incrementTimeouts(float dtime);
 	u32 getTimedOuts(float timeout);
@@ -307,7 +310,8 @@ enum ConnectionCommandType{
 	CONNCMD_SEND_TO_ALL,
 	CONCMD_ACK,
 	CONCMD_CREATE_PEER,
-	CONNCMD_RESEND_ONE
+	CONNCMD_RESEND_ONE,
+	CONNCMD_PEER_ID_SET
 };
 
 // This is very similar to ConnectionEvent
@@ -328,6 +332,7 @@ struct ConnectionCommand
 	static ConnectionCommandPtr disconnect();
 	static ConnectionCommandPtr disconnect_peer(session_t peer_id);
 	static ConnectionCommandPtr resend_one(session_t peer_id);
+	static ConnectionCommandPtr peer_id_set(session_t own_peer_id);
 	static ConnectionCommandPtr send(session_t peer_id, u8 channelnum, NetworkPacket *pkt, bool reliable);
 	static ConnectionCommandPtr ack(session_t peer_id, u8 channelnum, const Buffer<u8> &data);
 	static ConnectionCommandPtr createPeer(session_t peer_id, const Buffer<u8> &data);

src/network/mtp/threads.cpp

@@ -476,6 +476,11 @@ void ConnectionSendThread::processNonReliableCommand(ConnectionCommandPtr &c_ptr
 				<< " UDP processing CONNCMD_DISCONNECT_PEER" << std::endl);
 			disconnect_peer(c.peer_id);
 			return;
+		case CONNCMD_PEER_ID_SET:
+			LOG(dout_con << m_connection->getDesc()
+				<< " UDP processing CONNCMD_PEER_ID_SET" << std::endl);
+			fix_peer_id(c.peer_id);
+			return;
 		case CONNCMD_SEND:
 			LOG(dout_con << m_connection->getDesc()
 				<< " UDP processing CONNCMD_SEND" << std::endl);
@@ -579,6 +584,26 @@ void ConnectionSendThread::disconnect_peer(session_t peer_id)
 	dynamic_cast<UDPPeer *>(&peer)->m_pending_disconnect = true;
 }
 
+void ConnectionSendThread::fix_peer_id(session_t own_peer_id)
+{
+	auto peer_ids = m_connection->getPeerIDs();
+	for (const session_t peer_id : peer_ids) {
+		PeerHelper peer = m_connection->getPeerNoEx(peer_id);
+		if (!peer)
+			continue;
+
+		auto *udp_peer = dynamic_cast<UDPPeer*>(&peer);
+		if (!udp_peer)
+			continue;
+
+		for (int ch = 0; ch < CHANNEL_COUNT; ch++) {
+			auto &channel = udp_peer->channels[ch];
+
+			channel.outgoing_reliables_sent.fixPeerId(own_peer_id);
+		}
+	}
+}
+
 void ConnectionSendThread::send(session_t peer_id, u8 channelnum,
 	const SharedBuffer<u8> &data)
 {

src/network/mtp/threads.h

@@ -70,6 +70,7 @@ private:
 	void connect(Address address);
 	void disconnect();
 	void disconnect_peer(session_t peer_id);
+	void fix_peer_id(session_t own_peer_id);
 	void send(session_t peer_id, u8 channelnum, const SharedBuffer<u8> &data);
 	void sendReliable(ConnectionCommandPtr &c);
 	void sendToAll(u8 channelnum, const SharedBuffer<u8> &data);