Consumer: Fix sequence number gap (#1494)

versatica · Feb 24, 2025 · 13a7804 · 13a7804
1 parent 8327893
commit 13a7804
Show file tree

Hide file tree

Showing 8 changed files with 330 additions and 9 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,8 @@
 
 ### NEXT
 
+- `Consumer`: Fix sequence number gap ([PR #1494](https://github.com/versatica/mediasoup/pull/1494)).
+
 ### 3.15.4
 
 - `Worker`: Drop VP8 packets with a higher temporal layer than the current one ([PR #1009](https://github.com/versatica/mediasoup/pull/1009)).

diff --git a/worker/include/Channel/ChannelSocket.hpp b/worker/include/Channel/ChannelSocket.hpp
@@ -82,6 +82,9 @@ namespace Channel
 		};
 
 	public:
+#ifdef MS_TEST
+		explicit ChannelSocket();
+#endif
 		explicit ChannelSocket(int consumerFd, int producerFd);
 		explicit ChannelSocket(
 		  ChannelReadFn channelReadFn,

diff --git a/worker/meson.build b/worker/meson.build
@@ -333,6 +333,7 @@ test_sources = [
   'test/src/RTC/TestRtpStreamSend.cpp',
   'test/src/RTC/TestRtpStreamRecv.cpp',
   'test/src/RTC/TestSeqManager.cpp',
+  'test/src/RTC/TestSimpleConsumer.cpp',
   'test/src/RTC/TestTrendCalculator.cpp',
   'test/src/RTC/TestRtpEncodingParameters.cpp',
   'test/src/RTC/TestTransportCongestionControlServer.cpp',

diff --git a/worker/src/Channel/ChannelSocket.cpp b/worker/src/Channel/ChannelSocket.cpp
@@ -30,6 +30,13 @@ namespace Channel
 
 	/* Instance methods. */
 
+#ifdef MS_TEST
+	ChannelSocket::ChannelSocket()
+	{
+		MS_TRACE_STD();
+	}
+#endif
+
 	ChannelSocket::ChannelSocket(int consumerFd, int producerFd)
 	  : consumerSocket(new ConsumerSocket(consumerFd, MessageMaxLen, this)),
 	    producerSocket(new ProducerSocket(producerFd, MessageMaxLen))
@@ -256,7 +263,7 @@ namespace Channel
 		{
 			this->channelWriteFn(payload, payloadLen, this->channelWriteCtx);
 		}
-		else
+		else if (this->producerSocket)
 		{
 			this->producerSocket->Write(payload, payloadLen);
 		}

diff --git a/worker/src/RTC/SimpleConsumer.cpp b/worker/src/RTC/SimpleConsumer.cpp
@@ -320,6 +320,8 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::CONSUMER_INACTIVE);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
@@ -335,6 +337,8 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::UNSUPPORTED_PAYLOAD_TYPE);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
@@ -355,6 +359,8 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::DROPPED_BY_CODEC);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
@@ -366,18 +372,20 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::NOT_A_KEYFRAME);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
 		// Packets with only padding are not forwarded.
 		if (packet->GetPayloadLength() == 0)
 		{
-			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
-
 #ifdef MS_RTC_LOGGER_RTP
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::EMPTY_PAYLOAD);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 

diff --git a/worker/src/RTC/SimulcastConsumer.cpp b/worker/src/RTC/SimulcastConsumer.cpp
@@ -736,6 +736,8 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::CONSUMER_INACTIVE);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
@@ -745,6 +747,8 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::INVALID_TARGET_LAYER);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
@@ -777,6 +781,8 @@ namespace RTC
 				packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::NOT_A_KEYFRAME);
 #endif
 
+				this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 				return;
 			}
 
@@ -804,19 +810,20 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::NOT_A_KEYFRAME);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
 			return;
 		}
 
 		// If the packet belongs to current spatial layer being sent and packet does
 		// not have payload other than padding, then drop it.
 		if (spatialLayer == this->currentSpatialLayer && packet->GetPayloadLength() == 0)
 		{
-			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
-
 #ifdef MS_RTC_LOGGER_RTP
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::EMPTY_PAYLOAD);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
@@ -935,6 +942,8 @@ namespace RTC
 					packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::TOO_HIGH_TIMESTAMP_EXTRA_NEEDED);
 #endif
 
+					this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 					return;
 				}
 
@@ -980,6 +989,8 @@ namespace RTC
 				  RtcLogger::RtpPacket::DropReason::PACKET_PREVIOUS_TO_SPATIAL_LAYER_SWITCH);
 #endif
 
+				this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 				return;
 			}
 			else if (SeqManager<uint16_t>::IsSeqHigherThan(
@@ -1022,12 +1033,12 @@ namespace RTC
 			// Rewrite payload if needed. Drop packet if necessary.
 			if (!packet->ProcessPayload(this->encodingContext.get(), marker))
 			{
-				this->rtpSeqManager->Drop(packet->GetSequenceNumber());
-
 #ifdef MS_RTC_LOGGER_RTP
 				packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::DROPPED_BY_CODEC);
 #endif
 
+				this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 				return;
 			}
 

diff --git a/worker/src/RTC/SvcConsumer.cpp b/worker/src/RTC/SvcConsumer.cpp
@@ -645,6 +645,8 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::CONSUMER_INACTIVE);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
@@ -659,6 +661,8 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::INVALID_TARGET_LAYER);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
@@ -674,6 +678,8 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::UNSUPPORTED_PAYLOAD_TYPE);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
@@ -684,18 +690,20 @@ namespace RTC
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::NOT_A_KEYFRAME);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}
 
 		// Packets with only padding are not forwarded.
 		if (packet->GetPayloadLength() == 0)
 		{
-			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
-
 #ifdef MS_RTC_LOGGER_RTP
 			packet->logger.Dropped(RtcLogger::RtpPacket::DropReason::EMPTY_PAYLOAD);
 #endif
 
+			this->rtpSeqManager->Drop(packet->GetSequenceNumber());
+
 			return;
 		}