From 31706db5bb349b6d64ecfe3ac176b095c3512189 Mon Sep 17 00:00:00 2001
From: Michael Mavroudis <michael@mavroudis.com>
Date: Thu, 25 Jun 2026 11:53:01 -0400
Subject: [PATCH] Make WebSocketClient::_ctx access thread-safe

isConnected()/sendBinary()/sendMessage() (callable from the media thread) read
_ctx while connect()/disconnect() write it from another thread, with no
synchronization -- a data race (ThreadSanitizer-confirmed) on the shared_ptr.
Use C++11 atomic shared_ptr ops for every _ctx access; disconnect() atomically
swaps _ctx out and calls stop() (the event-thread join) on the local copy,
outside the swap, so it cannot deadlock a callback thread.
---
 src/WebSocketClient.cpp | 48 +++++++++++++++++++++++++++--------------
 1 file changed, 32 insertions(+), 16 deletions(-)

diff --git a/src/WebSocketClient.cpp b/src/WebSocketClient.cpp
index 9cd4266..f778526 100644
--- a/src/WebSocketClient.cpp
+++ b/src/WebSocketClient.cpp
@@ -17,8 +17,12 @@ WebSocketClient::~WebSocketClient() {
 }
 
 bool WebSocketClient::isConnected() {
-    if (_ctx) {
-        return _ctx->isConnected();
+    // _ctx is published/cleared from connect()/disconnect() on a different thread
+    // than the senders/isConnected callers (media thread). Access it atomically so
+    // the shared_ptr read never races the reset()/assign (TSan-verified).
+    auto ctx = std::atomic_load(&_ctx);
+    if (ctx) {
+        return ctx->isConnected();
     }
     return false;
 }
@@ -106,43 +110,51 @@ void WebSocketClient::enableCompression(bool enable) {
 
 void WebSocketClient::setOpenCallback(OpenCallback callback) {
     open_callback = std::move(callback);
-    if (_ctx) _ctx->setOpenCallback(open_callback);
+    auto ctx = std::atomic_load(&_ctx);
+    if (ctx) ctx->setOpenCallback(open_callback);
 }
 
 void WebSocketClient::setCloseCallback(CloseCallback callback) {
     close_callback = std::move(callback);
-    if (_ctx) _ctx->setCloseCallback(close_callback);
+    auto ctx = std::atomic_load(&_ctx);
+    if (ctx) ctx->setCloseCallback(close_callback);
 }
 
 void WebSocketClient::setErrorCallback(ErrorCallback callback) {
     error_callback = std::move(callback);
-    if (_ctx) _ctx->setErrorCallback(error_callback);
+    auto ctx = std::atomic_load(&_ctx);
+    if (ctx) ctx->setErrorCallback(error_callback);
 }
 
 void WebSocketClient::setMessageCallback(MessageCallback callback) {
     message_callback = std::move(callback);
-    if (_ctx) _ctx->setMessageCallback(message_callback);
+    auto ctx = std::atomic_load(&_ctx);
+    if (ctx) ctx->setMessageCallback(message_callback);
 }
 
 void WebSocketClient::setBinaryCallback(BinaryCallback callback) {
     binary_callback = std::move(callback);
-    if (_ctx) _ctx->setBinaryCallback(binary_callback);
+    auto ctx = std::atomic_load(&_ctx);
+    if (ctx) ctx->setBinaryCallback(binary_callback);
 }
 
 bool WebSocketClient::sendMessage(const std::string& message) {
-    return _ctx && _ctx->sendData(message.data(), message.size(), MessageType::TEXT);
+    auto ctx = std::atomic_load(&_ctx);
+    return ctx && ctx->sendData(message.data(), message.size(), MessageType::TEXT);
 }
 
 bool WebSocketClient::sendMessage(const char* msg, size_t len) {
-    return _ctx && _ctx->sendData(msg, len, MessageType::TEXT);
+    auto ctx = std::atomic_load(&_ctx);
+    return ctx && ctx->sendData(msg, len, MessageType::TEXT);
 }
 
 bool WebSocketClient::sendBinary(const void* data, size_t length) {
-    return _ctx && _ctx->sendData(data, length, MessageType::BINARY);
+    auto ctx = std::atomic_load(&_ctx);
+    return ctx && ctx->sendData(data, length, MessageType::BINARY);
 }
 
 void WebSocketClient::connect() {
-    if (_ctx) {
+    if (std::atomic_load(&_ctx)) {
         return;
     }
 
@@ -166,8 +178,8 @@ void WebSocketClient::connect() {
         if (message_callback) ctx->setMessageCallback(message_callback);
         if (binary_callback) ctx->setBinaryCallback(binary_callback);
 
-        _ctx = ctx;
-        _ctx->start();
+        std::atomic_store(&_ctx, ctx);
+        ctx->start();
 
     } catch (...) {
         // Failed to create or start context.
@@ -176,8 +188,12 @@ void WebSocketClient::connect() {
 }
 
 void WebSocketClient::disconnect() {
-    if (_ctx) {
-        _ctx->stop();
-        _ctx.reset();
+    // Atomically take and clear _ctx so a concurrent sendBinary/isConnected
+    // either sees the old context (and keeps it alive via its own shared_ptr
+    // copy) or sees null — never a torn read. stop() (which joins the event
+    // thread) runs OUTSIDE the swap, so it can't deadlock a callback thread.
+    auto ctx = std::atomic_exchange(&_ctx, std::shared_ptr<WebSocketContext>{});
+    if (ctx) {
+        ctx->stop();
     }
 }