Wake a WebSocket read on a pong timeout on every platform

When max_missed_pongs ran out, the heartbeat thread shut down the read
side of the socket to end a read() blocked on the silent peer. On
Windows that does not wake the wait, so read() stayed blocked until its
read timeout and then returned Timeout. The heartbeat test never ran on
Windows, so CI did not catch it.

The heartbeat now sets a flag that the stream's read interrupt checks,
the mechanism SSEClient::stop() uses, and WebSocketSSLStream takes the
interrupt too. It is set only when a pong timeout can happen, so other
connections wait as before. read() reports a timeout on a connection
that has closed as Fail, since Timeout means it is still usable.

Build and run the heartbeat test in the Windows jobs, and add a TLS
pong timeout test.

.github/workflows/test.yaml

@@ -623,6 +623,10 @@ jobs:
         }
         if ($failed) { exit 1 }
         Write-Host "All shards passed."
+    - name: Run WebSocket heartbeat test ${{ matrix.config.name }}
+      if: ${{ matrix.config.run_tests }}
+      working-directory: build/test
+      run: ./Release/httplib-heartbeat-test.exe --gtest_color=yes
 
     env:
       VCPKG_ROOT: "C:/vcpkg"

httplib.h

@@ -4645,6 +4645,8 @@ private:
   int max_missed_pongs_;
   int unacked_pings_ = 0;
   std::atomic<bool> closed_{false};
+  // Set when the peer stopped answering pings, to end a read() waiting on it
+  std::atomic<bool> pong_timed_out_{false};
   // Set once the caller has bounded read() through set_read_timeout(). Until
   // then the timeout in effect is the compile-time default, and elapsing it
   // is a failure that closes the connection, not a Timeout.
@@ -10766,6 +10768,7 @@ public:
   socket_t socket() const override;
   time_t duration() const override;
   void set_read_timeout(time_t sec, time_t usec = 0) override;
+  void set_read_interrupt(std::function<bool()> interrupted) override;
 
 private:
   mutable std::mutex session_mutex_;
@@ -10780,6 +10783,7 @@ private:
   time_t write_timeout_sec_;
   time_t write_timeout_usec_;
   const std::chrono::time_point<std::chrono::steady_clock> start_time_;
+  std::function<bool()> read_interrupted_;
 };
 
 #ifdef CPPHTTPLIB_OPENSSL_SUPPORT
@@ -13386,7 +13390,8 @@ inline bool WebSocketSSLStream::is_readable() const {
 }
 
 inline bool WebSocketSSLStream::wait_readable() const {
-  return select_read(sock_, read_timeout_sec_, read_timeout_usec_) > 0;
+  return select_read_interruptible(sock_, read_timeout_sec_, read_timeout_usec_,
+                                   read_interrupted_) > 0;
 }
 
 inline bool WebSocketSSLStream::wait_writable() const {
@@ -13486,6 +13491,11 @@ inline void WebSocketSSLStream::set_read_timeout(time_t sec, time_t usec) {
   read_timeout_usec_ = usec;
 }
 
+inline void
+WebSocketSSLStream::set_read_interrupt(std::function<bool()> interrupted) {
+  read_interrupted_ = std::move(interrupted);
+}
+
 } // namespace detail
 #endif // CPPHTTPLIB_SSL_ENABLED
 
@@ -23331,12 +23341,15 @@ inline ReadResult WebSocket::read(std::string &msg) {
     impl::FrameRead r =
         impl::read_websocket_frame(strm_, opcode, payload, fin, is_server_,
                                    CPPHTTPLIB_WEBSOCKET_MAX_PAYLOAD_LENGTH);
-    // A timeout landed on a frame boundary: the connection is untouched and
+    // A timeout landed on a frame boundary of an open connection: it is
-    // still usable, so hand control back without closing it. That is only
+    // untouched and still usable, so hand control back without closing it.
-    // useful to a caller who asked for the timeout; the compile-time default
+    // That is only useful to a caller who asked for the timeout; the
-    // is a backstop against a peer gone quiet, and elapsing it closes the
+    // compile-time default is a backstop against a peer gone quiet, and
-    // connection so a plain `while (ws.read(msg))` loop ends.
+    // elapsing it closes the connection so a plain `while (ws.read(msg))` loop
-    if (r == impl::FrameRead::Timeout && read_timeout_set_) { return Timeout; }
+    // ends. A wait that a pong timeout cut short finds the connection closed.
+    if (r == impl::FrameRead::Timeout && read_timeout_set_ && !closed_) {
+      return Timeout;
+    }
     if (r != impl::FrameRead::Ok) {
       closed_ = true;
       return Fail;
@@ -23465,10 +23478,16 @@ inline WebSocket::~WebSocket() {
   }
   ping_cv_.notify_all();
   if (ping_thread_.joinable()) { ping_thread_.join(); }
+  // A server's stream outlives this object
+  strm_.set_read_interrupt(nullptr);
 }
 
 inline void WebSocket::start_heartbeat() {
   if (ping_interval_sec_ <= 0) { return; }
+  // A pong timeout ends a read() blocked on the silent peer
+  if (max_missed_pongs_ > 0) {
+    strm_.set_read_interrupt([this] { return pong_timed_out_.load(); });
+  }
   ping_thread_ = std::thread([this]() {
     std::unique_lock<std::mutex> lock(ping_mutex_);
     // The predicate keeps a spurious wakeup from sending a ping early
@@ -23480,14 +23499,7 @@ inline void WebSocket::start_heartbeat() {
       if (max_missed_pongs_ > 0 && unacked_pings_ >= max_missed_pongs_) {
         lock.unlock();
         close(CloseStatus::GoingAway, "pong timeout");
-        // Wake a read() blocked on the unresponsive peer. Only the read side
+        pong_timed_out_ = true;
-        // is shut down: a TLS backend answers the EOF with an alert, and
-        // writing it to a socket closed for writing raises SIGPIPE.
-#ifdef _WIN32
-        shutdown(strm_.socket(), SD_RECEIVE);
-#else
-        shutdown(strm_.socket(), SHUT_RD);
-#endif
         return;
       }
       lock.unlock();

test/CMakeLists.txt

@@ -31,6 +31,14 @@ target_compile_options(httplib-test PRIVATE "$<$<CXX_COMPILER_ID:MSVC>:/utf-8;/b
 target_link_libraries(httplib-test PRIVATE httplib GTest::gtest_main CURL::libcurl)
 gtest_discover_tests(httplib-test)
 
+# It overrides the heartbeat macros, which a compiled library has fixed
+if(NOT HTTPLIB_COMPILE)
+    add_executable(httplib-heartbeat-test test_websocket_heartbeat.cc)
+    target_compile_options(httplib-heartbeat-test PRIVATE "$<$<CXX_COMPILER_ID:MSVC>:/utf-8;/bigobj>")
+    target_link_libraries(httplib-heartbeat-test PRIVATE httplib GTest::gtest_main)
+    gtest_discover_tests(httplib-heartbeat-test)
+endif()
+
 file(
     COPY www www2 www3 ca-bundle.crt image.jpg
     DESTINATION ${CMAKE_CURRENT_BINARY_DIR}

test/test_websocket_heartbeat.cc

@@ -146,16 +146,21 @@ TEST_F(WebSocketServerPingIntervalTest, ServerRuntimeInterval) {
 // ping count reaches max_missed_pongs.
 class WebSocketPongTimeoutTest : public ::testing::Test {
 protected:
+  virtual std::unique_ptr<Server> make_server() {
+    return std::unique_ptr<Server>(new Server());
+  }
+
   void SetUp() override {
-    svr_.set_websocket_ping_interval(0);
+    svr_ = make_server();
-    svr_.WebSocket("/ws", [this](const Request &, ws::WebSocket &) {
+    svr_->set_websocket_ping_interval(0);
+    svr_->WebSocket("/ws", [this](const Request &, ws::WebSocket &) {
       std::unique_lock<std::mutex> lock(handler_mutex_);
       handler_cv_.wait(lock, [this]() { return release_; });
     });
 
-    port_ = svr_.bind_to_any_port("localhost");
+    port_ = svr_->bind_to_any_port("localhost");
-    thread_ = std::thread([this]() { svr_.listen_after_bind(); });
+    thread_ = std::thread([this]() { svr_->listen_after_bind(); });
-    svr_.wait_until_ready();
+    svr_->wait_until_ready();
   }
 
   void TearDown() override {
@@ -164,11 +169,11 @@ protected:
       release_ = true;
     }
     handler_cv_.notify_all();
-    svr_.stop();
+    svr_->stop();
     thread_.join();
   }
 
-  Server svr_;
+  std::unique_ptr<Server> svr_;
   int port_;
   std::thread thread_;
   std::mutex handler_mutex_;
@@ -176,8 +181,10 @@ protected:
   bool release_ = false;
 };
 
-TEST_F(WebSocketPongTimeoutTest, ClientDetectsNonResponsivePeer) {
+// Client pings every 1s (compile-time default in this test file). With
-  ws::WebSocketClient client("ws://localhost:" + std::to_string(port_) + "/ws");
+// max_missed_pongs = 2, the heartbeat thread should self-close within roughly
+// 3s, and that has to end a read() already waiting on the peer.
+static void expect_pong_timeout_ends_read(ws::WebSocketClient &client) {
   client.set_websocket_max_missed_pongs(2);
   // A read timeout asked for at runtime is reported as Timeout, so it cannot
   // be mistaken for the Fail a pong timeout produces.
@@ -185,9 +192,6 @@ TEST_F(WebSocketPongTimeoutTest, ClientDetectsNonResponsivePeer) {
   ASSERT_TRUE(client.connect());
   ASSERT_TRUE(client.is_open());
 
-  // Client pings every 1s (compile-time default in this test file).
-  // With max_missed_pongs = 2, the heartbeat thread should self-close within
-  // roughly 3s, and that has to end a read() already waiting on the peer.
   auto start = std::chrono::steady_clock::now();
   std::string msg;
   EXPECT_EQ(client.read(msg), ws::Fail);
@@ -196,6 +200,27 @@ TEST_F(WebSocketPongTimeoutTest, ClientDetectsNonResponsivePeer) {
   EXPECT_FALSE(client.is_open());
 }
 
+TEST_F(WebSocketPongTimeoutTest, ClientDetectsNonResponsivePeer) {
+  ws::WebSocketClient client("ws://localhost:" + std::to_string(port_) + "/ws");
+  expect_pong_timeout_ends_read(client);
+}
+
+#ifdef CPPHTTPLIB_SSL_ENABLED
+class WebSocketPongTimeoutSSLTest : public WebSocketPongTimeoutTest {
+protected:
+  std::unique_ptr<Server> make_server() override {
+    return std::unique_ptr<Server>(new SSLServer("./cert.pem", "./key.pem"));
+  }
+};
+
+TEST_F(WebSocketPongTimeoutSSLTest, ClientDetectsNonResponsivePeer) {
+  ws::WebSocketClient client("wss://localhost:" + std::to_string(port_) +
+                             "/ws");
+  client.enable_server_certificate_verification(false);
+  expect_pong_timeout_ends_read(client);
+}
+#endif
+
 // The compile-time client read timeout (3s here) was never asked for through
 // set_read_timeout(), so when it elapses read() reports Fail and closes the
 // connection rather than handing back a Timeout on a still-open one.