Merge pull request #1554 from arvidn/message-race

Fix race between DTLS ClientHello and registerIncoming()

Author: paullouisageneau

src/impl/transport.cpp

@@ -38,7 +38,24 @@ void Transport::unregisterIncoming() {
 
 Transport::State Transport::state() const { return mState; }
 
-void Transport::onRecv(message_callback callback) { mRecvCallback = std::move(callback); }
+void Transport::onRecv(message_callback callback) {
+	std::vector<message_ptr> pending;
+	{
+		std::lock_guard lock(mPendingMutex);
+		mRecvCallback = std::move(callback);
+		if (mRecvCallback)
+			pending = std::move(mPendingRecv);
+		else
+			mPendingRecv.clear();
+	}
+	for (auto &msg : pending) {
+		try {
+			mRecvCallback(msg);
+		} catch (const std::exception &e) {
+			PLOG_WARNING << e.what();
+		}
+	}
+}
 
 void Transport::onStateChange(state_callback callback) {
 	mStateChangeCallback = std::move(callback);
@@ -52,6 +69,17 @@ bool Transport::send(message_ptr message) { return outgoing(message); }
 
 void Transport::recv(message_ptr message) {
 	try {
+		std::unique_lock lock(mPendingMutex);
+		if (!mRecvCallback) {
+			// No callback registered yet; buffer the message for replay when
+			// onRecv() is called.  Bounded to avoid unbounded growth.
+			if (mPendingRecv.size() < 8)
+				mPendingRecv.push_back(std::move(message));
+			else
+				PLOG_WARNING << "dropping incoming message, no receive callback";
+			return;
+		}
+		lock.unlock();
 		mRecvCallback(message);
 	} catch (const std::exception &e) {
 		PLOG_WARNING << e.what();

src/impl/transport.hpp

@@ -17,6 +17,8 @@
 #include <atomic>
 #include <functional>
 #include <memory>
+#include <mutex>
+#include <vector>
 
 namespace rtc::impl {
 
@@ -53,6 +55,13 @@ class Transport {
 	synchronized_callback<message_ptr> mRecvCallback;
 
 	std::atomic<State> mState = State::Disconnected;
+
+	// Packets received before a callback is registered are held here and
+	// replayed when onRecv() is first called with a non-null callback.
+	// This prevents the ICE->DTLS and DTLS->SCTP races where one side
+	// sends its first packet before the other side has called registerIncoming().
+	std::mutex mPendingMutex;
+	std::vector<message_ptr> mPendingRecv;
 };
 
 } // namespace rtc::impl