Fix joining / unmuting.

Telegram/SourceFiles/calls/calls_group_call.cpp

@@ -471,6 +471,7 @@ void GroupCall::rejoin(not_null<PeerData*> as) {
 				MTP_dataJSON(MTP_bytes(json))
 			)).done([=](const MTPUpdates &updates) {
 				_mySsrc = ssrc;
+				_mySsrcs.emplace(ssrc);
 				setState((_instanceState.current()
 					== InstanceState::Disconnected)
 					? State::Connecting
@@ -732,9 +733,7 @@ void GroupCall::handlePossibleCreateOrJoinResponse(
 			join(MTP_inputGroupCall(data.vid(), data.vaccess_hash()));
 		}
 		return;
-	} else if (_id != data.vid().v
-		|| _accessHash != data.vaccess_hash().v
-		|| !_instance) {
+	} else if (_id != data.vid().v || !_instance) {
 		return;
 	}
 	const auto streamDcId = MTP::BareDcId(
@@ -923,18 +922,27 @@ void GroupCall::handleUpdate(const MTPDupdateGroupCallParticipants &data) {
 			if (data.is_left()) {
 				if (data.vsource().v == _mySsrc) {
 					// I was removed from the call, rejoin.
-					LOG(("Call Info: Rejoin after got 'left' with my ssrc."));
+					LOG(("Call Info: "
+						"Rejoin after got 'left' with my ssrc."));
 					setState(State::Joining);
 					rejoin();
 				}
 				return;
 			} else if (data.vsource().v != _mySsrc) {
-				// I joined from another device, hangup.
-				LOG(("Call Info: Hangup after '!left' with ssrc %1, my %2."
-					).arg(data.vsource().v
-					).arg(_mySsrc));
-				_mySsrc = 0;
-				hangup();
+				if (!_mySsrcs.contains(data.vsource().v)) {
+					// I joined from another device, hangup.
+					LOG(("Call Info: "
+						"Hangup after '!left' with ssrc %1, my %2."
+						).arg(data.vsource().v
+						).arg(_mySsrc));
+					_mySsrc = 0;
+					hangup();
+				} else {
+					LOG(("Call Info: "
+						"Some old 'self' with '!left' and ssrc %1, my %2."
+						).arg(data.vsource().v
+						).arg(_mySsrc));
+				}
 				return;
 			}
 			if (data.is_muted() && !data.is_can_self_unmute()) {

Telegram/SourceFiles/calls/calls_group_call.h

@@ -309,6 +309,7 @@ private:
 	uint64 _id = 0;
 	uint64 _accessHash = 0;
 	uint32 _mySsrc = 0;
+	base::flat_set<uint32> _mySsrcs;
 	mtpRequestId _createRequestId = 0;
 	mtpRequestId _updateMuteRequestId = 0;
 

Telegram/SourceFiles/data/data_group_call.cpp

@@ -202,21 +202,23 @@ void GroupCall::enqueueUpdate(const MTPUpdate &update) {
 	update.match([&](const MTPDupdateGroupCall &updateData) {
 		updateData.vcall().match([&](const MTPDgroupCall &data) {
 			const auto version = data.vversion().v;
-			if (!_version || _version == version) {
+			if (!_applyingQueuedUpdates
+				&& (!_version || _version == version)) {
 				DEBUG_LOG(("Group Call Participants: "
 					"Apply updateGroupCall %1 -> %2"
 					).arg(_version
 					).arg(version));
-				applyUpdate(update);
-			} else if (_version < version) {
+				applyEnqueuedUpdate(update);
+			} else if (!_version || _version <= version) {
 				DEBUG_LOG(("Group Call Participants: "
 					"Queue updateGroupCall %1 -> %2"
 					).arg(_version
 					).arg(version));
-				_queuedUpdates.emplace(std::pair{ version, false }, update);
+				const auto type = QueuedType::Call;
+				_queuedUpdates.emplace(std::pair{ version, type }, update);
 			}
 		}, [&](const MTPDgroupCallDiscarded &data) {
-			applyUpdate(update);
+			discard();
 		});
 	}, [&](const MTPDupdateGroupCallParticipants &updateData) {
 		const auto version = updateData.vversion().v;
@@ -230,19 +232,22 @@ void GroupCall::enqueueUpdate(const MTPUpdate &update) {
 			true,
 			proj);
 		const auto required = increment ? (version - 1) : version;
-		if (_version == required) {
+		if (!_applyingQueuedUpdates && (_version == required)) {
 			DEBUG_LOG(("Group Call Participants: "
 				"Apply updateGroupCallParticipant %1 (%2)"
 				).arg(_version
 				).arg(Logs::b(increment)));
-			applyUpdate(update);
-		} else if (_version < required) {
+			applyEnqueuedUpdate(update);
+		} else if (_version <= required) {
 			DEBUG_LOG(("Group Call Participants: "
 				"Queue updateGroupCallParticipant %1 -> %2 (%3)"
 				).arg(_version
 				).arg(version
 				).arg(Logs::b(increment)));
-			_queuedUpdates.emplace(std::pair{ version, increment }, update);
+			const auto type = increment
+				? QueuedType::VersionedParticipant
+				: QueuedType::Participant;
+			_queuedUpdates.emplace(std::pair{ version, type }, update);
 		}
 	}, [](const auto &) {
 		Unexpected("Type in GroupCall::enqueueUpdate.");
@@ -327,12 +332,16 @@ void GroupCall::applyLocalUpdate(
 		ApplySliceSource::UpdateReceived);
 }
 
-void GroupCall::applyUpdate(const MTPUpdate &update) {
+void GroupCall::applyEnqueuedUpdate(const MTPUpdate &update) {
+	Expects(!_applyingQueuedUpdates);
+
+	_applyingQueuedUpdates = true;
+	const auto guard = gsl::finally([&] { _applyingQueuedUpdates = false; });
+
 	update.match([&](const MTPDupdateGroupCall &data) {
 		data.vcall().match([&](const MTPDgroupCall &data) {
 			applyCallFields(data);
 			computeParticipantsCount();
-			processQueuedUpdates();
 		}, [&](const MTPDgroupCallDiscarded &data) {
 			discard();
 		});
@@ -351,7 +360,7 @@ void GroupCall::applyUpdate(const MTPUpdate &update) {
 			data.vparticipants().v,
 			ApplySliceSource::UpdateReceived);
 	}, [](const auto &) {
-		Unexpected("Type in GroupCall::processQueuedUpdates.");
+		Unexpected("Type in GroupCall::applyEnqueuedUpdate.");
 	});
 	Core::App().calls().applyGroupCallUpdateChecked(
 		&_peer->session(),
@@ -359,7 +368,7 @@ void GroupCall::applyUpdate(const MTPUpdate &update) {
 }
 
 void GroupCall::processQueuedUpdates() {
-	if (!_version) {
+	if (!_version || _applyingQueuedUpdates) {
 		return;
 	}
 
@@ -367,15 +376,16 @@ void GroupCall::processQueuedUpdates() {
 	while (!_queuedUpdates.empty()) {
 		const auto &entry = _queuedUpdates.front();
 		const auto version = entry.first.first;
-		const auto versionIncremented = entry.first.second;
+		const auto type = entry.first.second;
+		const auto incremented = (type == QueuedType::VersionedParticipant);
 		if ((version < _version)
-			|| (version == _version && versionIncremented)) {
+			|| (version == _version && incremented)) {
 			_queuedUpdates.erase(_queuedUpdates.begin());
 		} else if (version == _version
-			|| (version == _version + 1 && versionIncremented)) {
+			|| (version == _version + 1 && incremented)) {
 			const auto update = entry.second;
 			_queuedUpdates.erase(_queuedUpdates.begin());
-			applyUpdate(update);
+			applyEnqueuedUpdate(update);
 		} else {
 			break;
 		}
@@ -395,7 +405,7 @@ void GroupCall::computeParticipantsCount() {
 }
 
 void GroupCall::reload() {
-	if (_reloadRequestId) {
+	if (_reloadRequestId || _applyingQueuedUpdates) {
 		return;
 	} else if (_participantsRequestId) {
 		api().request(_participantsRequestId).cancel();
@@ -410,7 +420,7 @@ void GroupCall::reload() {
 		const auto &entry = _queuedUpdates.front();
 		const auto update = entry.second;
 		_queuedUpdates.erase(_queuedUpdates.begin());
-		applyUpdate(update);
+		applyEnqueuedUpdate(update);
 	}
 	_reloadByQueuedUpdatesTimer.cancel();
 

Telegram/SourceFiles/data/data_group_call.h

@@ -113,6 +113,11 @@ private:
 		UnknownLoaded,
 		UpdateReceived,
 	};
+	enum class QueuedType : uint8 {
+		VersionedParticipant,
+		Participant,
+		Call,
+	};
 	[[nodiscard]] ApiWrap &api() const;
 
 	void discard();
@@ -124,7 +129,7 @@ private:
 	void changePeerEmptyCallFlag();
 	void checkFinishSpeakingByActive();
 	void applyCallFields(const MTPDgroupCall &data);
-	void applyUpdate(const MTPUpdate &update);
+	void applyEnqueuedUpdate(const MTPUpdate &update);
 	void setServerParticipantsCount(int count);
 	void computeParticipantsCount();
 	void processQueuedUpdates();
@@ -144,7 +149,9 @@ private:
 	mtpRequestId _reloadRequestId = 0;
 	rpl::variable<QString> _title;
 
-	base::flat_multi_map<std::pair<int,bool>, MTPUpdate> _queuedUpdates;
+	base::flat_multi_map<
+		std::pair<int, QueuedType>,
+		MTPUpdate> _queuedUpdates;
 	base::Timer _reloadByQueuedUpdatesTimer;
 	std::optional<MTPphone_GroupCall> _savedFull;
 
@@ -168,6 +175,7 @@ private:
 	bool _canChangeJoinMuted = true;
 	bool _allParticipantsLoaded = false;
 	bool _joinedToTop = false;
+	bool _applyingQueuedUpdates = false;
 
 };