From 4768714472e58a53f7e35808b7c5554ac73ac767 Mon Sep 17 00:00:00 2001 From: Benjamin Sergeant Date: Thu, 5 Nov 2026 14:02:59 -0800 Subject: [PATCH 1/3] Handle send failures on the connection I/O thread --- ixwebsocket/IXWebSocketTransport.cpp | 100 ++++++++---- ixwebsocket/IXWebSocketTransport.h | 13 +- test/CMakeLists.txt | 3 +- test/IXWebSocketSendBufferTest.cpp | 14 ++ test/IXWebSocketSendFailureTest.cpp | 227 +++++++++++++++++++++++++++ 5 files changed, 323 insertions(+), 34 deletions(-) create mode 100644 test/IXWebSocketSendFailureTest.cpp diff --git a/ixwebsocket/IXWebSocketTransport.cpp b/ixwebsocket/IXWebSocketTransport.cpp index 6ca7d576..ae28a87a 100644 --- a/ixwebsocket/IXWebSocketTransport.cpp +++ b/ixwebsocket/IXWebSocketTransport.cpp @@ -228,6 +228,14 @@ namespace ix // No state change, return if (_readyState == readyState) return; + if (readyState == ReadyState::OPEN) + { + initTimePointsAfterConnect(); + _pongReceived = false; + } + + // Terminal callbacks must observe CLOSED, including when they call close() again. + _readyState = readyState; if (readyState == ReadyState::CLOSED) { if (_onCloseCallback) @@ -239,13 +247,6 @@ namespace ix _closeWireSize = 0; _closeRemote = false; } - else if (readyState == ReadyState::OPEN) - { - initTimePointsAfterConnect(); - _pongReceived = false; - } - - _readyState = readyState; } void WebSocketTransport::setOnCloseCallback(const OnCloseCallback& onCloseCallback) @@ -327,6 +328,7 @@ namespace ix WebSocketTransport::PollResult WebSocketTransport::poll() { + if (_readyState == ReadyState::CLOSED) return PollResult::AbnormalClose; if (_readyState == ReadyState::OPEN) { if (pingIntervalExceeded()) @@ -374,7 +376,9 @@ namespace ix } // poll the socket - PollResultType pollResult = _socket->isReadyToRead(lastingTimeoutDelayInMs); + PollResultType pollResult = _sendError != SendError::None + ? PollResultType::SendRequest + : _socket->isReadyToRead(lastingTimeoutDelayInMs); // Make sure we send all the buffered data // there can be a lot of it for large messages. @@ -382,6 +386,7 @@ namespace ix { if (!flushSendBuffer()) { + receiveFromSocket(true); return PollResult::CannotFlushSendBuffer; } } @@ -399,6 +404,12 @@ namespace ix return PollResult::AbnormalClose; } + if (_sendError != SendError::None) + { + receiveFromSocket(true); + return PollResult::CannotFlushSendBuffer; + } + if (_readyState == ReadyState::CLOSING && closingDelayExceeded()) { _rxbuf.clear(); @@ -418,6 +429,7 @@ namespace ix _txbuf.clear(); _txbufPending.clear(); _txbufPendingOffset = 0; + _sendError = SendError::None; _rxbuf.clear(); _rxbufWanted = 0; _chunks.clear(); @@ -822,7 +834,9 @@ namespace ix // if we previously closed the connection (CLOSING state), then set state to CLOSED // (code/reason were set before) - if (_readyState == ReadyState::CLOSING) + const auto sendError = _sendError.load(); + if (_readyState == ReadyState::CLOSING && + (sendError == SendError::None || sendError == SendError::DuringClose)) { closeSocket(); setReadyState(ReadyState::CLOSED); @@ -830,10 +844,13 @@ namespace ix // if we weren't closing, then close using abnormal close code and message else if (_readyState != ReadyState::CLOSED) { - closeSocketAndSwitchToClosedState(WebSocketCloseConstants::kAbnormalCloseCode, - WebSocketCloseConstants::kAbnormalCloseMessage, - 0, - false); + closeSocketAndSwitchToClosedState( + WebSocketCloseConstants::kAbnormalCloseCode, + sendError == SendError::Timeout + ? WebSocketCloseConstants::kSendTimeoutMessage + : WebSocketCloseConstants::kAbnormalCloseMessage, + 0, + false); } } } @@ -901,7 +918,8 @@ namespace ix bool compress, const OnProgressCallback& onProgressCallback) { - if (_readyState != ReadyState::OPEN && _readyState != ReadyState::CLOSING) + if (_sendError != SendError::None || + (_readyState != ReadyState::OPEN && _readyState != ReadyState::CLOSING)) { return WebSocketSendInfo(false); } @@ -1130,9 +1148,30 @@ namespace ix wsheader_type::TEXT_FRAME, message, _enablePerMessageDeflate, onProgressCallback); } + void WebSocketTransport::recordSendFailure(SendError error) + { + _txbuf.clear(); + _txbufPending.clear(); + _txbufPendingOffset = 0; + // Preserve a close request that preceded the failed write, including its code/reason. + if (_readyState == ReadyState::CLOSING) error = SendError::DuringClose; + auto expected = SendError::None; + if (_sendError.compare_exchange_strong(expected, error)) + { + wakeUpFromPoll(SelectInterrupt::kSendRequest); + } + } + bool WebSocketTransport::sendOnSocket() { std::lock_guard lock(_txbufMutex); + if (_sendError != SendError::None || _readyState == ReadyState::CLOSED) + { + _txbuf.clear(); + _txbufPending.clear(); + _txbufPendingOffset = 0; + return false; + } while (!_txbufPending.empty() || !_txbuf.empty()) { @@ -1154,14 +1193,7 @@ namespace ix } else if (ret <= 0) { - _txbuf.clear(); - _txbufPending.clear(); - _txbufPendingOffset = 0; - closeSocket(); - if (_readyState != ReadyState::CLOSING) - { - setReadyState(ReadyState::CLOSED); - } + recordSendFailure(SendError::SocketError); return false; } else @@ -1178,7 +1210,7 @@ namespace ix return true; } - bool WebSocketTransport::receiveFromSocket() + bool WebSocketTransport::receiveFromSocket(bool readOnce) { while (true) { @@ -1194,7 +1226,11 @@ namespace ix // There's also no point in reading more bytes than needed. if (_rxbufWanted > 0 && _rxbuf.size() >= _rxbufWanted) break; - std::ptrdiff_t ret = _socket->recv((char*) &_readbuf[0], _readbuf.size()); + std::ptrdiff_t ret; + { + std::lock_guard lock(_socketMutex); + ret = _socket->recv((char*) &_readbuf[0], _readbuf.size()); + } if (ret < 0 && Socket::isWaitNeeded()) { @@ -1212,6 +1248,8 @@ namespace ix else { _rxbuf.insert(_rxbuf.end(), _readbuf.begin(), _readbuf.begin() + ret); + // Bound the final drain even if the peer continues sending indefinitely. + if (readOnce || _sendError != SendError::None) break; } } @@ -1351,7 +1389,7 @@ namespace ix timeoutSecs = std::chrono::seconds(_pingIntervalSecs); } - while (!isSendBufferEmpty() && !_requestInitCancellation) + while (!isSendBufferEmpty() && !_requestInitCancellation && _sendError == SendError::None) { // Wait with a 10ms timeout until the socket is ready to write. // This way we are not busy looping @@ -1359,8 +1397,8 @@ namespace ix if (result == PollResultType::Error) { - closeSocket(); - setReadyState(ReadyState::CLOSED); + std::lock_guard lock(_txbufMutex); + recordSendFailure(SendError::SocketError); return false; } else if (result == PollResultType::ReadyForWrite) @@ -1377,16 +1415,14 @@ namespace ix // as abnormal close and use "Send Timeout" for the reason. if (now > start + timeoutSecs) { - closeSocketAndSwitchToClosedState(WebSocketCloseConstants::kAbnormalCloseCode, - WebSocketCloseConstants::kSendTimeoutMessage, - 0, - false); + std::lock_guard lock(_txbufMutex); + recordSendFailure(SendError::Timeout); return false; } } } - return true; + return _sendError == SendError::None; } void WebSocketTransport::setCloseReason(const std::string& reason) diff --git a/ixwebsocket/IXWebSocketTransport.h b/ixwebsocket/IXWebSocketTransport.h index 24b13cc9..e4f1a04c 100644 --- a/ixwebsocket/IXWebSocketTransport.h +++ b/ixwebsocket/IXWebSocketTransport.h @@ -179,6 +179,17 @@ namespace ix // _rxbuf has reached this size to avoid unnecessary erase churn. uint64_t _rxbufWanted = 0; + enum class SendError + { + None, + SocketError, + Timeout, + DuringClose + }; + std::atomic _sendError {SendError::None}; + // Called with _txbufMutex held; notification and socket closure belong to the I/O thread. + void recordSendFailure(SendError error); + // Contains all messages that are waiting to be sent std::vector _txbuf; // Retries must keep the same address, contents and length. New frames go into _txbuf. @@ -275,7 +286,7 @@ namespace ix bool flushSendBuffer(); bool sendOnSocket(); - bool receiveFromSocket(); + bool receiveFromSocket(bool readOnce = false); WebSocketSendInfo sendData(wsheader_type::opcode_type type, const IXWebSocketSendData& message, diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 25690fae..63c19c73 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -32,6 +32,7 @@ set (TEST_TARGET_NAMES IXWebSocketIPv6Test IXWebSocketSendTimeoutTest IXWebSocketSendBufferTest + IXWebSocketSendFailureTest ) # Some unittest don't work on windows yet @@ -118,7 +119,7 @@ foreach(TEST_TARGET_NAME ${TEST_TARGET_NAMES}) endforeach() -set_tests_properties(IXWebSocketReconnectTest IXWebSocketDispatchTest PROPERTIES TIMEOUT 30) +set_tests_properties(IXWebSocketReconnectTest IXWebSocketDispatchTest IXWebSocketSendFailureTest PROPERTIES TIMEOUT 30) if (USE_NETWORK_FRAMEWORK OR (UNIX AND USE_TLS AND (USE_OPEN_SSL OR USE_LIBRE_SSL OR USE_MBED_TLS))) # Only the test credential generator needs the openssl command-line tool. diff --git a/test/IXWebSocketSendBufferTest.cpp b/test/IXWebSocketSendBufferTest.cpp index 36c3efdf..152b0e04 100644 --- a/test/IXWebSocketSendBufferTest.cpp +++ b/test/IXWebSocketSendBufferTest.cpp @@ -88,5 +88,19 @@ TEST_CASE("Fatal sends discard both pending and newly queued bytes", "[send_buff socket.fail = true; CHECK_FALSE(ix::WebSocketTransportTestAccess::flush(transport)); CHECK(transport.bufferedAmount() == 0); + // The sender records failure; only the I/O thread delivers the terminal callback. + CHECK(transport.getReadyState() == ix::WebSocketTransport::ReadyState::OPEN); + transport.dispatch( + ix::WebSocketTransport::PollResult::CannotFlushSendBuffer, + [](const std::string&, size_t, bool, ix::WebSocketTransport::MessageKind) { }); CHECK(transport.getReadyState() == ix::WebSocketTransport::ReadyState::CLOSED); + + ix::WebSocketTransportTestAccess::reset(transport); + PartialWriteSocket nextSocket; + nextSocket.writesBeforeBlocking = -1; + installWriteSocket(transport, nextSocket); + const auto fresh = writeTestMessage(41, 20); + ix::WebSocketTransportTestAccess::enqueue(transport, fresh); + flushWriteTest(transport); + CHECK(nextSocket.received == fresh); } diff --git a/test/IXWebSocketSendFailureTest.cpp b/test/IXWebSocketSendFailureTest.cpp new file mode 100644 index 00000000..d8e732cf --- /dev/null +++ b/test/IXWebSocketSendFailureTest.cpp @@ -0,0 +1,227 @@ +/* + * IXWebSocketSendFailureTest.cpp + * Copyright (c) 2026 Machine Zone, Inc. All rights reserved. + */ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace ix +{ + struct WebSocketTransportTestAccess + { + static void install(WebSocketTransport& transport, + std::unique_ptr socket, + bool blocking) + { + transport._socket = std::move(socket); + transport._useMask = false; + transport._blockingSend = blocking; + transport._sendTimeoutSecs = 1; + transport._rxbuf = {0x81, 3, 'o', 'n', 'e', 0x81, 3, 't', 'w', 'o'}; + transport.setReadyState(WebSocketTransport::ReadyState::OPEN); + } + }; +} // namespace ix + +namespace +{ + using Transport = ix::WebSocketTransport; + + class FailedWriteSocket final : public ix::Socket + { + public: + explicit FailedWriteSocket(int mode) + : mode(mode) + { + } + + std::ptrdiff_t send(char*, size_t) override + { + ++writes; + setErrno(mode < 2 ? EIO : EWOULDBLOCK); + return -1; + } + + std::ptrdiff_t recv(void* buffer, size_t) override + { + // An always-readable peer must not keep failed-send shutdown alive forever. + const std::string packet("\x81\x05three", 7); + std::memcpy(buffer, packet.data(), packet.size()); + ++reads; + return static_cast(packet.size()); + } + + ix::PollResultType isReadyToWrite(int) override + { + if (mode == 3) + { + std::this_thread::sleep_for(std::chrono::milliseconds(1100)); + return ix::PollResultType::Timeout; + } + if (mode == 4) + { + ++polls; + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + return ix::PollResultType::Timeout; + } + return ix::PollResultType::Error; + } + + void close() override + { + ++closes; + } + + const int mode; + std::atomic polls {0}; + std::atomic reads {0}; + std::atomic writes {0}; + std::atomic closes {0}; + }; +} // namespace + +TEST_CASE("Send failures finish callbacks on the I/O thread", "[websocket][close][send_failure]") +{ + // Client fatal write, server fatal write, server flush error, server send timeout. + const int mode = GENERATE(0, 1, 2, 3); + CAPTURE(mode); + Transport transport; + auto* socket = new FailedWriteSocket(mode); + ix::WebSocketTransportTestAccess::install( + transport, std::unique_ptr(socket), mode != 0); + std::mutex mutex; + std::condition_variable changed; + bool entered = false; + bool release = false; + bool messageActive = false; + bool overlapped = false; + bool closedInCallback = false; + uint16_t closeCode = 0; + std::string closeReason; + std::vector events; + std::vector callbackThreads; + transport.setOnCloseCallback( + [&](uint16_t code, const std::string& reason, size_t, bool) + { + { + std::lock_guard lock(mutex); + overlapped = messageActive; + closedInCallback = transport.getReadyState() == Transport::ReadyState::CLOSED; + closeCode = code; + closeReason = reason; + events.push_back("close"); + callbackThreads.push_back(std::this_thread::get_id()); + } + // Calling close() from a terminal notification must not deadlock. + if (closedInCallback) transport.close(); + }); + std::thread::id ioThread; + auto owner = std::async( + std::launch::async, + [&] + { + ioThread = std::this_thread::get_id(); + auto message = [&](const std::string& value, size_t, bool, Transport::MessageKind) + { + std::unique_lock lock(mutex); + events.push_back(value); + callbackThreads.push_back(std::this_thread::get_id()); + if (value == "one") + { + messageActive = true; + entered = true; + changed.notify_all(); + changed.wait(lock, [&] { return release; }); + messageActive = false; + } + }; + transport.dispatch(Transport::PollResult::Succeeded, message); + transport.dispatch(transport.poll(), message); + }); + bool received; + { + std::unique_lock lock(mutex); + received = changed.wait_for(lock, std::chrono::seconds(2), [&] { return entered; }); + } + auto sender = + std::async(std::launch::async, + [&] { return transport.sendText(std::string("outgoing"), nullptr).success; }); + const bool returned = sender.wait_for(std::chrono::seconds(3)) == std::future_status::ready; + { + std::lock_guard lock(mutex); + release = true; + } + changed.notify_all(); + const bool success = sender.get(); + owner.get(); + REQUIRE(received); + REQUIRE(returned); + CHECK_FALSE(success); + CHECK_FALSE(overlapped); + CHECK(closedInCallback); + CHECK(closeCode == ix::WebSocketCloseConstants::kAbnormalCloseCode); + CHECK(closeReason == (mode == 3 ? ix::WebSocketCloseConstants::kSendTimeoutMessage + : ix::WebSocketCloseConstants::kAbnormalCloseMessage)); + CHECK(events == std::vector {"one", "two", "three", "close"}); + CHECK(callbackThreads == std::vector(4, ioThread)); + CHECK(socket->closes == 1); + CHECK(socket->reads == 1); + CHECK(transport.bufferedAmount() == 0); +} + +TEST_CASE("Closing cancels a blocked send without waiting for its timeout", + "[websocket][close][send_failure]") +{ + Transport transport; + auto* socket = new FailedWriteSocket(4); + ix::WebSocketTransportTestAccess::install(transport, std::unique_ptr(socket), true); + auto sender = std::async(std::launch::async, + [&] { return transport.sendText(std::string("outgoing"), nullptr); }); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (socket->polls == 0 && std::chrono::steady_clock::now() < deadline) + std::this_thread::yield(); + const bool blocked = socket->polls != 0; + const auto start = std::chrono::steady_clock::now(); + transport.close(4000, "cancelled"); + const bool returned = + sender.wait_for(std::chrono::milliseconds(500)) == std::future_status::ready; + sender.get(); + CHECK(blocked); + CHECK(returned); + CHECK(std::chrono::steady_clock::now() - start < std::chrono::milliseconds(500)); + CHECK(transport.getReadyState() == Transport::ReadyState::CLOSING); +} + +TEST_CASE("A failed write preserves an earlier close request", "[websocket][close][send_failure]") +{ + const bool closeFirst = GENERATE(false, true); + Transport transport; + ix::WebSocketTransportTestAccess::install( + transport, std::unique_ptr(new FailedWriteSocket(0)), false); + uint16_t code = 0; + unsigned notifications = 0; + transport.setOnCloseCallback( + [&](uint16_t value, const std::string&, size_t, bool) + { + code = value; + ++notifications; + CHECK(transport.bufferedAmount() == 0); + CHECK_FALSE(transport.sendText(std::string("after close"), nullptr).success); + }); + if (!closeFirst) CHECK_FALSE(transport.sendText(std::string("outgoing"), nullptr).success); + transport.close(4000, "cancelled"); + auto message = [](const std::string&, size_t, bool, Transport::MessageKind) { }; + transport.dispatch(transport.poll(), message); + transport.dispatch(transport.poll(), message); + CHECK(code == (closeFirst ? 4000 : ix::WebSocketCloseConstants::kAbnormalCloseCode)); + CHECK(notifications == 1); +} From fef6c07d3d7c3aef90e9073fb8eab9a283d4bc67 Mon Sep 17 00:00:00 2001 From: Benjamin Sergeant Date: Thu, 5 Nov 2026 14:12:05 -0800 Subject: [PATCH 2/3] Simplify deferred send-failure shutdown --- ixwebsocket/IXWebSocketTransport.cpp | 49 +++++++++------------------- ixwebsocket/IXWebSocketTransport.h | 5 ++- test/IXWebSocketSendFailureTest.cpp | 20 +++--------- 3 files changed, 21 insertions(+), 53 deletions(-) diff --git a/ixwebsocket/IXWebSocketTransport.cpp b/ixwebsocket/IXWebSocketTransport.cpp index ae28a87a..120dcece 100644 --- a/ixwebsocket/IXWebSocketTransport.cpp +++ b/ixwebsocket/IXWebSocketTransport.cpp @@ -228,13 +228,7 @@ namespace ix // No state change, return if (_readyState == readyState) return; - if (readyState == ReadyState::OPEN) - { - initTimePointsAfterConnect(); - _pongReceived = false; - } - - // Terminal callbacks must observe CLOSED, including when they call close() again. + // Callbacks must observe the new state, including reentrant close() calls. _readyState = readyState; if (readyState == ReadyState::CLOSED) { @@ -247,6 +241,11 @@ namespace ix _closeWireSize = 0; _closeRemote = false; } + else if (readyState == ReadyState::OPEN) + { + initTimePointsAfterConnect(); + _pongReceived = false; + } } void WebSocketTransport::setOnCloseCallback(const OnCloseCallback& onCloseCallback) @@ -328,7 +327,7 @@ namespace ix WebSocketTransport::PollResult WebSocketTransport::poll() { - if (_readyState == ReadyState::CLOSED) return PollResult::AbnormalClose; + if (_sendError != SendError::None) return PollResult::CannotFlushSendBuffer; if (_readyState == ReadyState::OPEN) { if (pingIntervalExceeded()) @@ -376,9 +375,7 @@ namespace ix } // poll the socket - PollResultType pollResult = _sendError != SendError::None - ? PollResultType::SendRequest - : _socket->isReadyToRead(lastingTimeoutDelayInMs); + PollResultType pollResult = _socket->isReadyToRead(lastingTimeoutDelayInMs); // Make sure we send all the buffered data // there can be a lot of it for large messages. @@ -386,7 +383,6 @@ namespace ix { if (!flushSendBuffer()) { - receiveFromSocket(true); return PollResult::CannotFlushSendBuffer; } } @@ -404,11 +400,7 @@ namespace ix return PollResult::AbnormalClose; } - if (_sendError != SendError::None) - { - receiveFromSocket(true); - return PollResult::CannotFlushSendBuffer; - } + if (_sendError != SendError::None) return PollResult::CannotFlushSendBuffer; if (_readyState == ReadyState::CLOSING && closingDelayExceeded()) { @@ -835,8 +827,7 @@ namespace ix // if we previously closed the connection (CLOSING state), then set state to CLOSED // (code/reason were set before) const auto sendError = _sendError.load(); - if (_readyState == ReadyState::CLOSING && - (sendError == SendError::None || sendError == SendError::DuringClose)) + if (_readyState == ReadyState::CLOSING) { closeSocket(); setReadyState(ReadyState::CLOSED); @@ -1153,13 +1144,8 @@ namespace ix _txbuf.clear(); _txbufPending.clear(); _txbufPendingOffset = 0; - // Preserve a close request that preceded the failed write, including its code/reason. - if (_readyState == ReadyState::CLOSING) error = SendError::DuringClose; - auto expected = SendError::None; - if (_sendError.compare_exchange_strong(expected, error)) - { - wakeUpFromPoll(SelectInterrupt::kSendRequest); - } + if (_sendError == SendError::None) _sendError = error; + wakeUpFromPoll(SelectInterrupt::kSendRequest); } bool WebSocketTransport::sendOnSocket() @@ -1210,7 +1196,7 @@ namespace ix return true; } - bool WebSocketTransport::receiveFromSocket(bool readOnce) + bool WebSocketTransport::receiveFromSocket() { while (true) { @@ -1226,11 +1212,7 @@ namespace ix // There's also no point in reading more bytes than needed. if (_rxbufWanted > 0 && _rxbuf.size() >= _rxbufWanted) break; - std::ptrdiff_t ret; - { - std::lock_guard lock(_socketMutex); - ret = _socket->recv((char*) &_readbuf[0], _readbuf.size()); - } + std::ptrdiff_t ret = _socket->recv((char*) &_readbuf[0], _readbuf.size()); if (ret < 0 && Socket::isWaitNeeded()) { @@ -1248,8 +1230,7 @@ namespace ix else { _rxbuf.insert(_rxbuf.end(), _readbuf.begin(), _readbuf.begin() + ret); - // Bound the final drain even if the peer continues sending indefinitely. - if (readOnce || _sendError != SendError::None) break; + if (_sendError != SendError::None) break; } } diff --git a/ixwebsocket/IXWebSocketTransport.h b/ixwebsocket/IXWebSocketTransport.h index e4f1a04c..083c1ecf 100644 --- a/ixwebsocket/IXWebSocketTransport.h +++ b/ixwebsocket/IXWebSocketTransport.h @@ -183,8 +183,7 @@ namespace ix { None, SocketError, - Timeout, - DuringClose + Timeout }; std::atomic _sendError {SendError::None}; // Called with _txbufMutex held; notification and socket closure belong to the I/O thread. @@ -286,7 +285,7 @@ namespace ix bool flushSendBuffer(); bool sendOnSocket(); - bool receiveFromSocket(bool readOnce = false); + bool receiveFromSocket(); WebSocketSendInfo sendData(wsheader_type::opcode_type type, const IXWebSocketSendData& message, diff --git a/test/IXWebSocketSendFailureTest.cpp b/test/IXWebSocketSendFailureTest.cpp index d8e732cf..2f8ed72f 100644 --- a/test/IXWebSocketSendFailureTest.cpp +++ b/test/IXWebSocketSendFailureTest.cpp @@ -6,7 +6,6 @@ #include #include #include -#include #include #include #include @@ -51,15 +50,6 @@ namespace return -1; } - std::ptrdiff_t recv(void* buffer, size_t) override - { - // An always-readable peer must not keep failed-send shutdown alive forever. - const std::string packet("\x81\x05three", 7); - std::memcpy(buffer, packet.data(), packet.size()); - ++reads; - return static_cast(packet.size()); - } - ix::PollResultType isReadyToWrite(int) override { if (mode == 3) @@ -83,7 +73,6 @@ namespace const int mode; std::atomic polls {0}; - std::atomic reads {0}; std::atomic writes {0}; std::atomic closes {0}; }; @@ -171,10 +160,9 @@ TEST_CASE("Send failures finish callbacks on the I/O thread", "[websocket][close CHECK(closeCode == ix::WebSocketCloseConstants::kAbnormalCloseCode); CHECK(closeReason == (mode == 3 ? ix::WebSocketCloseConstants::kSendTimeoutMessage : ix::WebSocketCloseConstants::kAbnormalCloseMessage)); - CHECK(events == std::vector {"one", "two", "three", "close"}); - CHECK(callbackThreads == std::vector(4, ioThread)); + CHECK(events == std::vector {"one", "two", "close"}); + CHECK(callbackThreads == std::vector(3, ioThread)); CHECK(socket->closes == 1); - CHECK(socket->reads == 1); CHECK(transport.bufferedAmount() == 0); } @@ -201,7 +189,7 @@ TEST_CASE("Closing cancels a blocked send without waiting for its timeout", CHECK(transport.getReadyState() == Transport::ReadyState::CLOSING); } -TEST_CASE("A failed write preserves an earlier close request", "[websocket][close][send_failure]") +TEST_CASE("A close request keeps its code when a write fails", "[websocket][close][send_failure]") { const bool closeFirst = GENERATE(false, true); Transport transport; @@ -222,6 +210,6 @@ TEST_CASE("A failed write preserves an earlier close request", "[websocket][clos auto message = [](const std::string&, size_t, bool, Transport::MessageKind) { }; transport.dispatch(transport.poll(), message); transport.dispatch(transport.poll(), message); - CHECK(code == (closeFirst ? 4000 : ix::WebSocketCloseConstants::kAbnormalCloseCode)); + CHECK(code == 4000); CHECK(notifications == 1); } From 3b8c88b64d54522d961323cfeb396b5630aa626c Mon Sep 17 00:00:00 2001 From: Benjamin Sergeant Date: Thu, 5 Nov 2026 14:37:20 -0800 Subject: [PATCH 3/3] Reuse abnormal close state for failed sends --- ixwebsocket/IXWebSocketTransport.cpp | 68 ++++++++++++++++------------ ixwebsocket/IXWebSocketTransport.h | 11 +---- test/IXWebSocketSendBufferTest.cpp | 4 +- test/IXWebSocketSendFailureTest.cpp | 15 ++++-- 4 files changed, 54 insertions(+), 44 deletions(-) diff --git a/ixwebsocket/IXWebSocketTransport.cpp b/ixwebsocket/IXWebSocketTransport.cpp index 120dcece..ac999cc6 100644 --- a/ixwebsocket/IXWebSocketTransport.cpp +++ b/ixwebsocket/IXWebSocketTransport.cpp @@ -327,7 +327,7 @@ namespace ix WebSocketTransport::PollResult WebSocketTransport::poll() { - if (_sendError != SendError::None) return PollResult::CannotFlushSendBuffer; + if (isAbnormalClose()) return PollResult::AbnormalClose; if (_readyState == ReadyState::OPEN) { if (pingIntervalExceeded()) @@ -400,7 +400,7 @@ namespace ix return PollResult::AbnormalClose; } - if (_sendError != SendError::None) return PollResult::CannotFlushSendBuffer; + if (isAbnormalClose()) return PollResult::AbnormalClose; if (_readyState == ReadyState::CLOSING && closingDelayExceeded()) { @@ -421,7 +421,6 @@ namespace ix _txbuf.clear(); _txbufPending.clear(); _txbufPendingOffset = 0; - _sendError = SendError::None; _rxbuf.clear(); _rxbufWanted = 0; _chunks.clear(); @@ -826,7 +825,6 @@ namespace ix // if we previously closed the connection (CLOSING state), then set state to CLOSED // (code/reason were set before) - const auto sendError = _sendError.load(); if (_readyState == ReadyState::CLOSING) { closeSocket(); @@ -835,13 +833,10 @@ namespace ix // if we weren't closing, then close using abnormal close code and message else if (_readyState != ReadyState::CLOSED) { - closeSocketAndSwitchToClosedState( - WebSocketCloseConstants::kAbnormalCloseCode, - sendError == SendError::Timeout - ? WebSocketCloseConstants::kSendTimeoutMessage - : WebSocketCloseConstants::kAbnormalCloseMessage, - 0, - false); + closeSocketAndSwitchToClosedState(WebSocketCloseConstants::kAbnormalCloseCode, + WebSocketCloseConstants::kAbnormalCloseMessage, + 0, + false); } } } @@ -909,7 +904,7 @@ namespace ix bool compress, const OnProgressCallback& onProgressCallback) { - if (_sendError != SendError::None || + if (isAbnormalClose() || (_readyState != ReadyState::OPEN && _readyState != ReadyState::CLOSING)) { return WebSocketSendInfo(false); @@ -1139,19 +1134,16 @@ namespace ix wsheader_type::TEXT_FRAME, message, _enablePerMessageDeflate, onProgressCallback); } - void WebSocketTransport::recordSendFailure(SendError error) + bool WebSocketTransport::isAbnormalClose() const { - _txbuf.clear(); - _txbufPending.clear(); - _txbufPendingOffset = 0; - if (_sendError == SendError::None) _sendError = error; - wakeUpFromPoll(SelectInterrupt::kSendRequest); + return _readyState == ReadyState::CLOSING && + _closeCode == WebSocketCloseConstants::kAbnormalCloseCode; } bool WebSocketTransport::sendOnSocket() { - std::lock_guard lock(_txbufMutex); - if (_sendError != SendError::None || _readyState == ReadyState::CLOSED) + std::unique_lock lock(_txbufMutex); + if (isAbnormalClose() || _readyState == ReadyState::CLOSED) { _txbuf.clear(); _txbufPending.clear(); @@ -1179,7 +1171,10 @@ namespace ix } else if (ret <= 0) { - recordSendFailure(SendError::SocketError); + // close() takes the send-buffer lock when discarding an aborted connection. + lock.unlock(); + close(WebSocketCloseConstants::kAbnormalCloseCode, + WebSocketCloseConstants::kAbnormalCloseMessage); return false; } else @@ -1230,7 +1225,7 @@ namespace ix else { _rxbuf.insert(_rxbuf.end(), _readbuf.begin(), _readbuf.begin() + ret); - if (_sendError != SendError::None) break; + if (isAbnormalClose()) break; } } @@ -1322,6 +1317,23 @@ namespace ix } } + // An abnormal close has no handshake. Discard pending output and let the I/O + // thread close the socket and deliver the terminal callback. Never send 1006 on wire. + if (code == WebSocketCloseConstants::kAbnormalCloseCode) + { + { + std::lock_guard lock(_txbufMutex); + _txbuf.clear(); + _txbufPending.clear(); + _txbufPendingOffset = 0; + } + if (_readyState != ReadyState::CLOSED) + { + wakeUpFromPoll(SelectInterrupt::kCloseRequest); + } + return; + } + if (!startedClose) { // Wake up the socket polling thread, as @@ -1370,7 +1382,7 @@ namespace ix timeoutSecs = std::chrono::seconds(_pingIntervalSecs); } - while (!isSendBufferEmpty() && !_requestInitCancellation && _sendError == SendError::None) + while (!isSendBufferEmpty() && !_requestInitCancellation) { // Wait with a 10ms timeout until the socket is ready to write. // This way we are not busy looping @@ -1378,8 +1390,8 @@ namespace ix if (result == PollResultType::Error) { - std::lock_guard lock(_txbufMutex); - recordSendFailure(SendError::SocketError); + close(WebSocketCloseConstants::kAbnormalCloseCode, + WebSocketCloseConstants::kAbnormalCloseMessage); return false; } else if (result == PollResultType::ReadyForWrite) @@ -1396,14 +1408,14 @@ namespace ix // as abnormal close and use "Send Timeout" for the reason. if (now > start + timeoutSecs) { - std::lock_guard lock(_txbufMutex); - recordSendFailure(SendError::Timeout); + close(WebSocketCloseConstants::kAbnormalCloseCode, + WebSocketCloseConstants::kSendTimeoutMessage); return false; } } } - return _sendError == SendError::None; + return _readyState != ReadyState::CLOSED && !isAbnormalClose(); } void WebSocketTransport::setCloseReason(const std::string& reason) diff --git a/ixwebsocket/IXWebSocketTransport.h b/ixwebsocket/IXWebSocketTransport.h index 083c1ecf..498159b5 100644 --- a/ixwebsocket/IXWebSocketTransport.h +++ b/ixwebsocket/IXWebSocketTransport.h @@ -179,16 +179,6 @@ namespace ix // _rxbuf has reached this size to avoid unnecessary erase churn. uint64_t _rxbufWanted = 0; - enum class SendError - { - None, - SocketError, - Timeout - }; - std::atomic _sendError {SendError::None}; - // Called with _txbufMutex held; notification and socket closure belong to the I/O thread. - void recordSendFailure(SendError error); - // Contains all messages that are waiting to be sent std::vector _txbuf; // Retries must keep the same address, contents and length. New frames go into _txbuf. @@ -283,6 +273,7 @@ namespace ix bool wakeUpFromPoll(uint64_t wakeUpCode); + bool isAbnormalClose() const; bool flushSendBuffer(); bool sendOnSocket(); bool receiveFromSocket(); diff --git a/test/IXWebSocketSendBufferTest.cpp b/test/IXWebSocketSendBufferTest.cpp index 152b0e04..d101858e 100644 --- a/test/IXWebSocketSendBufferTest.cpp +++ b/test/IXWebSocketSendBufferTest.cpp @@ -88,8 +88,8 @@ TEST_CASE("Fatal sends discard both pending and newly queued bytes", "[send_buff socket.fail = true; CHECK_FALSE(ix::WebSocketTransportTestAccess::flush(transport)); CHECK(transport.bufferedAmount() == 0); - // The sender records failure; only the I/O thread delivers the terminal callback. - CHECK(transport.getReadyState() == ix::WebSocketTransport::ReadyState::OPEN); + // The sender starts an abnormal close; only the I/O thread delivers the callback. + CHECK(transport.getReadyState() == ix::WebSocketTransport::ReadyState::CLOSING); transport.dispatch( ix::WebSocketTransport::PollResult::CannotFlushSendBuffer, [](const std::string&, size_t, bool, ix::WebSocketTransport::MessageKind) { }); diff --git a/test/IXWebSocketSendFailureTest.cpp b/test/IXWebSocketSendFailureTest.cpp index 2f8ed72f..77db26ed 100644 --- a/test/IXWebSocketSendFailureTest.cpp +++ b/test/IXWebSocketSendFailureTest.cpp @@ -160,9 +160,10 @@ TEST_CASE("Send failures finish callbacks on the I/O thread", "[websocket][close CHECK(closeCode == ix::WebSocketCloseConstants::kAbnormalCloseCode); CHECK(closeReason == (mode == 3 ? ix::WebSocketCloseConstants::kSendTimeoutMessage : ix::WebSocketCloseConstants::kAbnormalCloseMessage)); - CHECK(events == std::vector {"one", "two", "close"}); - CHECK(callbackThreads == std::vector(3, ioThread)); + CHECK(events == std::vector {"one", "close"}); + CHECK(callbackThreads == std::vector(2, ioThread)); CHECK(socket->closes == 1); + CHECK(socket->writes == 1); // An abnormal close must not attempt a Close frame. CHECK(transport.bufferedAmount() == 0); } @@ -202,7 +203,6 @@ TEST_CASE("A close request keeps its code when a write fails", "[websocket][clos { code = value; ++notifications; - CHECK(transport.bufferedAmount() == 0); CHECK_FALSE(transport.sendText(std::string("after close"), nullptr).success); }); if (!closeFirst) CHECK_FALSE(transport.sendText(std::string("outgoing"), nullptr).success); @@ -210,6 +210,13 @@ TEST_CASE("A close request keeps its code when a write fails", "[websocket][clos auto message = [](const std::string&, size_t, bool, Transport::MessageKind) { }; transport.dispatch(transport.poll(), message); transport.dispatch(transport.poll(), message); - CHECK(code == 4000); + CHECK(code == (closeFirst ? 4000 : ix::WebSocketCloseConstants::kAbnormalCloseCode)); CHECK(notifications == 1); } + +TEST_CASE("Abnormal close before connecting is harmless", "[websocket][close][send_failure]") +{ + Transport transport; + transport.close(ix::WebSocketCloseConstants::kAbnormalCloseCode); + CHECK(transport.getReadyState() == Transport::ReadyState::CLOSED); +}