From bcc12faa6f6d379fee0e00d26df939dad8459083 Mon Sep 17 00:00:00 2001 From: Benjamin Sergeant Date: Wed, 14 Oct 2026 17:13:04 -0700 Subject: [PATCH 1/2] Reset WebSocket session state before reconnecting Discard queued wire frames, partial receive data, receive targets and fragment assembly state before a new client or server handshake. Old connection data must not be replayed or joined with a new connection's messages. Serialize handshakes with active application sends, releasing the write lock before Open callbacks so callbacks can still send normally. This prevents an interrupted fragmented send from appending old continuations after the new session's reset. Add loopback regressions for automatic reconnect, partial frames, queued output, failed reconnect attempts and a send paused between fragments. The tests fail without the reset; removing send coordination independently reproduces cross-connection continuation writes. Fixes #610 Fixes #611 --- ixwebsocket/IXWebSocket.cpp | 5 + ixwebsocket/IXWebSocketTransport.cpp | 15 ++ ixwebsocket/IXWebSocketTransport.h | 1 + test/CMakeLists.txt | 3 + test/IXWebSocketReconnectTest.cpp | 358 +++++++++++++++++++++++++++ 5 files changed, 382 insertions(+) create mode 100644 test/IXWebSocketReconnectTest.cpp diff --git a/ixwebsocket/IXWebSocket.cpp b/ixwebsocket/IXWebSocket.cpp index a92bd6d2..71022c7f 100644 --- a/ixwebsocket/IXWebSocket.cpp +++ b/ixwebsocket/IXWebSocket.cpp @@ -210,6 +210,8 @@ namespace ix WebSocketInitResult WebSocket::connect(int timeoutSecs) { + // Finish any in-flight send before discarding the previous connection's state. + std::unique_lock writeLock(_writeMutex); { std::lock_guard lock(_configMutex); _ws.configure( @@ -240,6 +242,7 @@ namespace ix } WebSocketInitResult status = _ws.connectToUrl(_url, headers, timeoutSecs); + writeLock.unlock(); if (!status.success) { return status; @@ -268,6 +271,7 @@ namespace ix HttpRequestPtr request, int sendTimeoutSecs) { + std::unique_lock writeLock(_writeMutex); { std::lock_guard lock(_configMutex); _ws.configure( @@ -276,6 +280,7 @@ namespace ix WebSocketInitResult status = _ws.connectToSocket( std::move(socket), timeoutSecs, enablePerMessageDeflate, request, sendTimeoutSecs); + writeLock.unlock(); if (!status.success) { return status; diff --git a/ixwebsocket/IXWebSocketTransport.cpp b/ixwebsocket/IXWebSocketTransport.cpp index 6b832fda..a56aa708 100644 --- a/ixwebsocket/IXWebSocketTransport.cpp +++ b/ixwebsocket/IXWebSocketTransport.cpp @@ -106,6 +106,7 @@ namespace ix const WebSocketHttpHeaders& headers, int timeoutSecs) { + resetConnectionState(); std::lock_guard lock(_socketMutex); std::string protocol, host, path, query; @@ -176,6 +177,7 @@ namespace ix HttpRequestPtr request, int sendTimeoutSecs) { + resetConnectionState(); std::lock_guard lock(_socketMutex); // Server should not mask the data it sends to the client @@ -401,6 +403,19 @@ namespace ix return PollResult::Succeeded; } + void WebSocketTransport::resetConnectionState() + { + // Frames and fragments belong to one connection and cannot be replayed on another. + // Take this lock before _socketMutex, matching sendOnSocket()'s lock order. + std::lock_guard lock(_txbufMutex); + _txbuf.clear(); + _rxbuf.clear(); + _rxbufWanted = 0; + _chunks.clear(); + _fragmentedMessageKind = MessageKind::MSG_TEXT; + _receivedMessageCompressed = false; + } + bool WebSocketTransport::isSendBufferEmpty() const { std::lock_guard lock(_txbufMutex); diff --git a/ixwebsocket/IXWebSocketTransport.h b/ixwebsocket/IXWebSocketTransport.h index f2391e15..5f8d6ebb 100644 --- a/ixwebsocket/IXWebSocketTransport.h +++ b/ixwebsocket/IXWebSocketTransport.h @@ -279,6 +279,7 @@ namespace ix const OnMessageCallback& onMessageCallback); bool isSendBufferEmpty() const; + void resetConnectionState(); template void appendToSendBuffer(const std::vector& header, diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index d0fc879a..c830d91d 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -15,6 +15,7 @@ set (TEST_TARGET_NAMES IXSocketConnectTest IXWebSocketServerTest IXWebSocketTestConnectionDisconnection + IXWebSocketReconnectTest IXUrlParserTest # IXHttpClientTest ## FIXME httpbin.org does not seem normal IXUnityBuildsTest @@ -106,6 +107,8 @@ foreach(TEST_TARGET_NAME ${TEST_TARGET_NAMES}) endforeach() +set_tests_properties(IXWebSocketReconnectTest PROPERTIES TIMEOUT 30) + if (USE_NETWORK_FRAMEWORK) # Only the test credential generator needs the openssl command-line tool. find_program(NETWORK_TEST_OPENSSL openssl) diff --git a/test/IXWebSocketReconnectTest.cpp b/test/IXWebSocketReconnectTest.cpp new file mode 100644 index 00000000..fb6a751b --- /dev/null +++ b/test/IXWebSocketReconnectTest.cpp @@ -0,0 +1,358 @@ +/* + * IXWebSocketReconnectTest.cpp + * Copyright (c) 2026 Machine Zone, Inc. All rights reserved. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace +{ + using Clock = std::chrono::steady_clock; + using Transport = ix::WebSocketTransport; + + class ReconnectTestServer final : public ix::SocketServer + { + public: + using Script = std::function; + + explicit ReconnectTestServer(const Script& script) + : SocketServer(ix::getFreePort(), "127.0.0.1") + , _script(script) + { + } + + ~ReconnectTestServer() override + { + stop(); + } + + void stop() override + { + _stopping = true; + SocketServer::stop(); + } + + std::string url() + { + return "ws://127.0.0.1:" + std::to_string(getPort()) + "/"; + } + + unsigned errors() const + { + return _errors; + } + + private: + void handleConnection(std::unique_ptr socket, + std::shared_ptr state) override + { + unsigned index = _connections++; + { + auto* rawSocket = socket.get(); + Transport transport; + auto result = transport.connectToSocket(std::move(socket), 2, false); + const auto deadline = Clock::now() + std::chrono::seconds(5); + auto cancelled = [this, deadline] { return _stopping || Clock::now() >= deadline; }; + if (!result.success || !_script(*rawSocket, index, cancelled)) ++_errors; + } + state->setTerminated(); + } + + size_t getConnectedClientsCount() override + { + return 0; + } + + Script _script; + std::atomic _stopping {false}; + std::atomic _connections {0}; + std::atomic _errors {0}; + }; + + void configureReconnectTestTransport(Transport& transport) + { + // Poll without blocking so that test deadlines also cover incomplete frames. + transport.configure(ix::WebSocketPerMessageDeflateOptions(false), {}, false, 0); + } +} // namespace + +TEST_CASE("Automatic reconnect discards old message fragments", "[websocket][reconnect]") +{ + ReconnectTestServer server( + [](ix::Socket& socket, unsigned index, const ix::CancellationRequest& cancelled) + { + if (index == 0) return socket.writeBytes(std::string("\x01\x03OLD", 5), cancelled); + if (!socket.writeBytes(std::string("\x81\x03NEW", 5), cancelled)) return false; + char byte; + socket.readByte(&byte, cancelled); // Keep this connection open until the client stops. + return true; + }); + REQUIRE(server.listen().first); + server.start(); + + std::mutex mutex; + std::condition_variable changed; + std::vector messages; + unsigned fragments = 0; + ix::WebSocket client; + client.setUrl(server.url()); + client.disablePerMessageDeflate(); + client.setMinWaitBetweenReconnectionRetries(10); + client.setMaxWaitBetweenReconnectionRetries(10); + client.setOnMessageCallback( + [&](const ix::WebSocketMessagePtr& message) + { + std::lock_guard lock(mutex); + if (message->type == ix::WebSocketMessageType::Fragment) ++fragments; + if (message->type == ix::WebSocketMessageType::Message) + { + messages.push_back(message->str); + changed.notify_all(); + } + }); + client.start(); + bool received; + { + std::unique_lock lock(mutex); + received = + changed.wait_for(lock, std::chrono::seconds(5), [&] { return !messages.empty(); }); + } + client.stop(); + server.stop(); + + REQUIRE(received); + REQUIRE(messages.size() == 1); + CHECK(messages.front() == "NEW"); + CHECK(fragments == 1); + CHECK(server.errors() == 0); +} + +TEST_CASE("Reconnect discards an incomplete received frame", "[websocket][reconnect]") +{ + std::atomic releaseFirst {false}; + ReconnectTestServer server( + [&](ix::Socket& socket, unsigned index, const ix::CancellationRequest& cancelled) + { + if (index == 0) + { + if (!socket.writeBytes(std::string("\x81\x05OL", 4), cancelled)) return false; + while (!releaseFirst && !cancelled()) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + return true; + } + return socket.writeBytes(std::string("\x81\x03NEW", 5), cancelled); + }); + REQUIRE(server.listen().first); + server.start(); + + Transport client; + configureReconnectTestTransport(client); + REQUIRE(client.connectToUrl(server.url(), {}, 2).success); + std::vector messages; + auto onMessage = [&](const std::string& message, size_t, bool, Transport::MessageKind kind) + { + if (kind == Transport::MessageKind::MSG_TEXT) messages.push_back(message); + }; + const auto deadline = Clock::now() + std::chrono::milliseconds(100); + while (Clock::now() < deadline) + { + client.dispatch(client.poll(), onMessage); + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + CHECK(messages.empty()); + CHECK(client.getReadyState() == Transport::ReadyState::OPEN); + + // A send-side failure can close the socket before dispatch consumes its receive buffer. + client.closeSocket(); + client.setReadyState(Transport::ReadyState::CLOSED); + releaseFirst = true; + REQUIRE(client.connectToUrl(server.url(), {}, 2).success); + const auto receiveDeadline = Clock::now() + std::chrono::seconds(3); + while (messages.empty() && Clock::now() < receiveDeadline && + client.getReadyState() == Transport::ReadyState::OPEN) + { + client.dispatch(client.poll(), onMessage); + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + client.closeSocket(); + server.stop(); + REQUIRE(messages.size() == 1); + CHECK(messages.front() == "NEW"); + CHECK(server.errors() == 0); +} + +TEST_CASE("Reconnect discards queued output", "[websocket][reconnect]") +{ + const bool invalidUrl = GENERATE(false, true); + CAPTURE(invalidUrl); + std::atomic releaseFirst {false}; + std::promise received; + auto wire = received.get_future(); + ReconnectTestServer server( + [&](ix::Socket& socket, unsigned index, const ix::CancellationRequest& cancelled) + { + if (index == 0) + { + // Deliberately do not read, so the client's nonblocking send queues output. + while (!releaseFirst && !cancelled()) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + return true; + } + if (index != 1) return false; + auto result = socket.readBytes(9, nullptr, nullptr, cancelled); + received.set_value(result.second); + return result.first; + }); + REQUIRE(server.listen().first); + server.start(); + + Transport client; + configureReconnectTestTransport(client); + REQUIRE(client.connectToUrl(server.url(), {}, 2).success); + const std::string payload(4 * 1024 * 1024, 'x'); + for (int i = 0; i < 4 && client.bufferedAmount() == 0; ++i) + REQUIRE(client.sendBinary(payload, nullptr).success); + REQUIRE(client.bufferedAmount() > 0); + client.closeSocket(); + client.setReadyState(Transport::ReadyState::CLOSED); + releaseFirst = true; + + auto result = client.connectToUrl(invalidUrl ? "invalid-url" : server.url(), {}, 2); + CHECK(result.success == !invalidUrl); + CHECK(client.bufferedAmount() == 0); + if (!invalidUrl) + { + REQUIRE(result.success); + CHECK(client.sendText(std::string("NEW"), nullptr).success); + REQUIRE(wire.wait_for(std::chrono::seconds(3)) == std::future_status::ready); + auto frame = wire.get(); + REQUIRE(frame.size() == 9); + CHECK(static_cast(frame[0]) == 0x81); + CHECK(static_cast(frame[1]) == 0x83); + std::string message; + for (size_t i = 0; i < 3; ++i) + message += frame[6 + i] ^ frame[2 + i]; + CHECK(message == "NEW"); + } + client.closeSocket(); + server.stop(); + CHECK(server.errors() == 0); +} + +TEST_CASE("Reconnect waits for an interrupted send", "[websocket][reconnect]") +{ + std::promise received; + auto wire = received.get_future(); + ReconnectTestServer server( + [&](ix::Socket& socket, unsigned index, const ix::CancellationRequest& cancelled) + { + if (index == 0) return socket.readBytes(32768 + 8, nullptr, nullptr, cancelled).first; + if (index != 1) return false; + auto result = socket.readBytes(9, nullptr, nullptr, cancelled); + received.set_value(result.second); + char byte; + socket.readByte(&byte, cancelled); + return result.first; + }); + REQUIRE(server.listen().first); + server.start(); + + std::mutex mutex; + std::condition_variable changed; + unsigned opens = 0; + bool closed = false; + bool paused = false; + bool release = false; + ix::WebSocket client; + client.setUrl(server.url()); + client.disablePerMessageDeflate(); + client.setOnMessageCallback( + [&](const ix::WebSocketMessagePtr& message) + { + std::lock_guard lock(mutex); + if (message->type == ix::WebSocketMessageType::Open) ++opens; + if (message->type == ix::WebSocketMessageType::Close) closed = true; + changed.notify_all(); + }); + client.start(); + bool connected; + { + std::unique_lock lock(mutex); + connected = changed.wait_for(lock, std::chrono::seconds(5), [&] { return opens == 1; }); + } + if (!connected) + { + client.stop(); + FAIL("Initial connection failed"); + } + + bool sendSucceeded = true; + std::thread sender( + [&] + { + sendSucceeded = client + .sendBinary(std::string(65536, 'x'), + [&](int step, int) + { + if (step == 0) + { + std::unique_lock lock(mutex); + paused = true; + changed.notify_all(); + changed.wait_for(lock, + std::chrono::seconds(5), + [&] { return release; }); + } + return true; + }) + .success; + }); + bool interrupted; + bool reconnectedWhilePaused; + { + std::unique_lock lock(mutex); + interrupted = + changed.wait_for(lock, std::chrono::seconds(3), [&] { return closed && paused; }); + reconnectedWhilePaused = + changed.wait_for(lock, std::chrono::milliseconds(100), [&] { return opens > 1; }); + release = true; + changed.notify_all(); + } + sender.join(); + bool reconnected; + { + std::unique_lock lock(mutex); + reconnected = changed.wait_for(lock, std::chrono::seconds(3), [&] { return opens >= 2; }); + } + bool freshSent = reconnected && client.sendText("NEW").success; + bool receivedFresh = wire.wait_for(std::chrono::seconds(3)) == std::future_status::ready; + client.stop(); + server.stop(); + + CHECK(interrupted); + CHECK_FALSE(reconnectedWhilePaused); + CHECK_FALSE(sendSucceeded); + CHECK(reconnected); + CHECK(freshSent); + REQUIRE(receivedFresh); + auto frame = wire.get(); + REQUIRE(frame.size() == 9); + CHECK(static_cast(frame[0]) == 0x81); + CHECK(static_cast(frame[1]) == 0x83); + std::string message; + for (size_t i = 0; i < 3; ++i) + message += frame[6 + i] ^ frame[2 + i]; + CHECK(message == "NEW"); + CHECK(server.errors() == 0); +} From 5f52adb2164c7b11b1703b6e7512905ad89a9ffb Mon Sep 17 00:00:00 2001 From: Benjamin Sergeant Date: Wed, 14 Oct 2026 17:15:50 -0700 Subject: [PATCH 2/2] Keep reconnect waiting cancellable from send progress callbacks --- ixwebsocket/IXWebSocket.cpp | 28 +++++++++++++++++++++-- ixwebsocket/IXWebSocket.h | 1 + test/IXWebSocketReconnectTest.cpp | 37 +++++++++++++++---------------- 3 files changed, 45 insertions(+), 21 deletions(-) diff --git a/ixwebsocket/IXWebSocket.cpp b/ixwebsocket/IXWebSocket.cpp index 71022c7f..14d0b16f 100644 --- a/ixwebsocket/IXWebSocket.cpp +++ b/ixwebsocket/IXWebSocket.cpp @@ -211,7 +211,11 @@ namespace ix WebSocketInitResult WebSocket::connect(int timeoutSecs) { // Finish any in-flight send before discarding the previous connection's state. - std::unique_lock writeLock(_writeMutex); + std::unique_lock writeLock(_writeMutex, std::defer_lock); + if (!waitForWriteLock(writeLock)) + { + return WebSocketInitResult(false, 0, "Connection cancelled while waiting for a send"); + } { std::lock_guard lock(_configMutex); _ws.configure( @@ -271,7 +275,11 @@ namespace ix HttpRequestPtr request, int sendTimeoutSecs) { - std::unique_lock writeLock(_writeMutex); + std::unique_lock writeLock(_writeMutex, std::defer_lock); + if (!waitForWriteLock(writeLock)) + { + return WebSocketInitResult(false, 0, "Connection cancelled while waiting for a send"); + } { std::lock_guard lock(_configMutex); _ws.configure( @@ -303,6 +311,22 @@ namespace ix return status; } + bool WebSocket::waitForWriteLock(std::unique_lock& lock) + { + while (!lock.try_lock()) + { + // A progress callback can call stop() while holding _writeMutex. Let the + // background thread exit so stop() can join it without waiting for that send. + std::unique_lock sleepLock(_sleepMutex); + if (_sleepCondition.wait_for( + sleepLock, std::chrono::milliseconds(10), [this] { return _stop.load(); })) + { + return false; + } + } + return true; + } + bool WebSocket::isConnected() const { return getReadyState() == ReadyState::Open; diff --git a/ixwebsocket/IXWebSocket.h b/ixwebsocket/IXWebSocket.h index 1070836f..139de2a7 100644 --- a/ixwebsocket/IXWebSocket.h +++ b/ixwebsocket/IXWebSocket.h @@ -129,6 +129,7 @@ namespace ix bool isConnected() const; bool isClosing() const; void checkConnection(bool firstConnectionAttempt); + bool waitForWriteLock(std::unique_lock& lock); static void invokeTrafficTrackerCallback(size_t size, bool incoming); // Server diff --git a/test/IXWebSocketReconnectTest.cpp b/test/IXWebSocketReconnectTest.cpp index fb6a751b..3ca6daca 100644 --- a/test/IXWebSocketReconnectTest.cpp +++ b/test/IXWebSocketReconnectTest.cpp @@ -252,6 +252,8 @@ TEST_CASE("Reconnect discards queued output", "[websocket][reconnect]") TEST_CASE("Reconnect waits for an interrupted send", "[websocket][reconnect]") { + const bool stopFromProgressCallback = GENERATE(false, true); + CAPTURE(stopFromProgressCallback); std::promise received; auto wire = received.get_future(); ReconnectTestServer server( @@ -297,27 +299,23 @@ TEST_CASE("Reconnect waits for an interrupted send", "[websocket][reconnect]") FAIL("Initial connection failed"); } + auto onProgress = [&](int step, int) + { + if (step == 0) + { + { + std::unique_lock lock(mutex); + paused = true; + changed.notify_all(); + changed.wait_for(lock, std::chrono::seconds(5), [&] { return release; }); + } + if (stopFromProgressCallback) client.stop(); + } + return true; + }; bool sendSucceeded = true; std::thread sender( - [&] - { - sendSucceeded = client - .sendBinary(std::string(65536, 'x'), - [&](int step, int) - { - if (step == 0) - { - std::unique_lock lock(mutex); - paused = true; - changed.notify_all(); - changed.wait_for(lock, - std::chrono::seconds(5), - [&] { return release; }); - } - return true; - }) - .success; - }); + [&] { sendSucceeded = client.sendBinary(std::string(65536, 'x'), onProgress).success; }); bool interrupted; bool reconnectedWhilePaused; { @@ -330,6 +328,7 @@ TEST_CASE("Reconnect waits for an interrupted send", "[websocket][reconnect]") changed.notify_all(); } sender.join(); + if (stopFromProgressCallback) client.start(); bool reconnected; { std::unique_lock lock(mutex);