Enhance WebSocket support with unresponsive-peer detection and documentation updates

- Added `set_websocket_max_missed_pongs` method to configure unresponsive-peer detection. - Updated README and documentation to clarify WebSocket limitations and features. - Introduced tests for detecting non-responsive peers and ensuring responsive peers do not trigger timeouts.
2026-06-08 18:30:49 +00:00 · 2026-04-11 22:17:38 -04:00
parent b4eec3ee77
commit 6bdd657713
6 changed files with 199 additions and 12 deletions
@@ -3,15 +3,19 @@
 A simple, blocking WebSocket implementation for C++11.

 > [!IMPORTANT]
-> This is a blocking I/O WebSocket implementation using a thread-per-connection model. If you need high-concurrency WebSocket support with non-blocking/async I/O (e.g., thousands of simultaneous connections), this is not the one that you want.
+> This is a blocking I/O WebSocket implementation using a thread-per-connection model (plus one heartbeat thread per connection). It is intended for small- to mid-scale workloads; handling large numbers of simultaneous WebSocket connections is outside the design target of this library. If you need high-concurrency WebSocket support with non-blocking/async I/O (e.g., thousands of simultaneous connections), this is not the one that you want.
+
+> [!NOTE]
+> WebSocket extensions (`permessage-deflate` and others defined by RFC 6455) are **not supported**. If a client proposes an extension via `Sec-WebSocket-Extensions`, the server silently declines it — the negotiated connection always runs without extensions.

 ## Features

- **RFC 6455 compliant**: Full WebSocket protocol support
+- **RFC 6455 compliant**: Full WebSocket protocol support (extensions are not implemented)
 - **Server and Client**: Both sides included
 - **SSL/TLS support**: `wss://` scheme for secure connections
 - **Text and Binary**: Both message types supported
 - **Automatic heartbeat**: Periodic Ping/Pong keeps connections alive
+- **Unresponsive-peer detection**: Opt-in liveness check via `set_websocket_max_missed_pongs()`
 - **Subprotocol negotiation**: `Sec-WebSocket-Protocol` support for GraphQL, MQTT, etc.

 ## Quick Start
@@ -352,6 +356,7 @@ if (ws.connect()) {
 | `CPPHTTPLIB_WEBSOCKET_READ_TIMEOUT_SECOND`  | `300`             | Read timeout for WebSocket connections (seconds)         |
 | `CPPHTTPLIB_WEBSOCKET_CLOSE_TIMEOUT_SECOND` | `5`               | Timeout for waiting peer's Close response (seconds)      |
 | `CPPHTTPLIB_WEBSOCKET_PING_INTERVAL_SECOND` | `30`              | Automatic Ping interval for heartbeat (seconds)          |
+| `CPPHTTPLIB_WEBSOCKET_MAX_MISSED_PONGS`     | `0` (disabled)    | Close the connection after N consecutive unacked pings   |

 ### Runtime Ping Interval

@@ -373,6 +378,20 @@ ws.set_websocket_ping_interval(10);  // 10 seconds
 ws.set_websocket_ping_interval(0);
 ```

+### Unresponsive-Peer Detection (Pong Timeout)
+
+By default the heartbeat only sends pings — it does not enforce that pongs come back. To detect a silently dropped connection faster, enable the max-missed-pongs check. Once `max_missed_pongs` consecutive pings go unanswered, the heartbeat thread closes the connection with `CloseStatus::GoingAway` and the reason `"pong timeout"`.
+
+```cpp
+ws.set_websocket_max_missed_pongs(2); // close after 2 consecutive unacked pings
+```
+
+The server side has the same `set_websocket_max_missed_pongs()`.
+
+With the default ping interval of 30 seconds, `max_missed_pongs = 2` detects a dead peer within ~60 seconds. The counter is reset every time a Pong frame is received, so the mechanism only works when your code is actively calling `read()` — exactly the pattern a normal WebSocket client already uses.
+
+**The default is `0`**, which means "never close the connection because of missing pongs." Pings are still sent on the heartbeat interval, but their responses are not checked. Even so, a dead connection does not linger forever: while your code is inside `read()`, `CPPHTTPLIB_WEBSOCKET_READ_TIMEOUT_SECOND` (default **300 seconds = 5 minutes**) acts as a backstop and `read()` fails if no frame arrives in time. `max_missed_pongs` is the knob for detecting an unresponsive peer faster than that 5-minute fallback.
+
 ## Threading Model

 WebSocket connections share the same thread pool as HTTP requests. Each WebSocket connection occupies one thread for its entire lifetime.
@@ -1462,7 +1462,11 @@ if (ws.connect()) {

 SSL is also supported via `wss://` scheme (e.g. `WebSocketClient("wss://example.com/ws")`). Subprotocol negotiation (`Sec-WebSocket-Protocol`) is supported via `SubProtocolSelector` callback.

-> **Note:** WebSocket connections occupy a thread for their entire lifetime. If you plan to handle many simultaneous WebSocket connections, consider using a dynamic thread pool: `svr.new_task_queue = [] { return new ThreadPool(8, 64); };`
+> **Note:** WebSocket connections occupy a thread for their entire lifetime (plus an additional thread per connection for heartbeat pings). This thread-per-connection model is intended for small- to mid-scale workloads; large numbers of simultaneous WebSocket connections are outside the design target of this library. If you expect many concurrent WebSocket clients, configure a dynamic thread pool (`svr.new_task_queue = [] { return new ThreadPool(8, 64); };`) and measure carefully.
+
+> **WebSocket extensions are not supported.** `permessage-deflate` and other RFC 6455 extensions are not implemented. If a client proposes them via `Sec-WebSocket-Extensions`, the server silently declines them in its handshake response.
+
+> **Unresponsive-peer detection.** Heartbeat pings also serve as a liveness probe when `set_websocket_max_missed_pongs(n)` is set: if the client sends `n` consecutive pings without receiving a pong, it will close the connection. Disabled by default (`0`).

 See [README-websocket.md](README-websocket.md) for more details.

@@ -57,4 +57,24 @@ Too short wastes bandwidth; too long and connections get dropped. As a rule of t

 > **Warning:** A very short ping interval spawns background work per connection and increases CPU usage. For servers with many connections, keep the interval modest.

+## Detecting an unresponsive peer
+
+Sending pings alone doesn't tell you anything if the peer just silently dies — the TCP socket might still look open while the process on the other end is long gone. To catch that, enable the max-missed-pongs check: if N consecutive pings go unanswered, the connection is closed.
+
+```cpp
+cli.set_websocket_max_missed_pongs(2); // close after 2 consecutive unacked pings
+```
+
+The server side has the same `set_websocket_max_missed_pongs()`.
+
+With a 30-second ping interval and `max_missed_pongs = 2`, a dead peer is detected within roughly 60 seconds and the connection is closed with `CloseStatus::GoingAway` and the reason `"pong timeout"`.
+
+The counter is reset whenever `read()` consumes an incoming Pong frame, so this only works if your code is actively calling `read()` in a loop — which is what a normal WebSocket client does anyway.
+
+### Why the default is 0
+
+`max_missed_pongs` defaults to `0`, which means "never close the connection because of missing pongs." Pings are still sent on the heartbeat interval, but their responses aren't checked. If you want unresponsive-peer detection, set it explicitly to `1` or higher.
+
+Even with `0`, a dead connection won't linger forever: while your code is inside `read()`, `CPPHTTPLIB_WEBSOCKET_READ_TIMEOUT_SECOND` (default **300 seconds = 5 minutes**) acts as a backstop and `read()` fails if no frame arrives in time. Think of `max_missed_pongs` as the knob for detecting an unresponsive peer **faster** than that.
+
 > For handling a closed connection, see [W03. Handle connection close](w03-websocket-close).
@@ -57,4 +57,24 @@ WebSocketプロトコルでは、PingフレームにはPongフレームで応答

 > **Warning:** Ping間隔を極端に短くすると、WebSocket接続ごとにバックグラウンドでスレッドが走るので、CPU負荷が上がります。接続数が多いサーバーでは控えめな値に設定しましょう。

+## 無応答のピアを検出する
+
+Pingを送るだけでは、相手が「黙って落ちた」場合に気付けません。TCPの接続自体は生きているように見えるのに、相手のプロセスはもう応答しない、というケースです。これを検出するには、送ったPingに対してPongがN回連続で返ってこなかったら接続を切る、というオプションを有効にします。
+
+```cpp
+cli.set_websocket_max_missed_pongs(2); // 2回連続でPongが返ってこなければ切断
+```
+
+サーバー側にも同じ`set_websocket_max_missed_pongs()`があります。
+
+たとえばPing間隔が30秒で`max_missed_pongs = 2`なら、無応答のピアは約60秒で検出され、`CloseStatus::GoingAway`（理由は`"pong timeout"`）で接続が閉じられます。
+
+この仕組みは`read()`を呼んでPongフレームを消費したタイミングでカウンタがリセットされます。つまり通常のWebSocketクライアントのように`read()`をループで回していれば、特に意識することなく動きます。
+
+### デフォルトは無効
+
+`max_missed_pongs`のデフォルトは`0`で、これは「Pongが何回返ってこなくてもこの仕組みでは切断しない」という意味です。Ping自体は送られ続けますが、応答の有無はチェックされません。無応答ピアを検出したい場合は明示的に`1`以上を設定してください。
+
+ただし`0`のままでも最終的に接続が残り続けることはありません。`read()`を呼んでいる間は`CPPHTTPLIB_WEBSOCKET_READ_TIMEOUT_SECOND`（デフォルト**300秒 = 5分**）が保険として働き、フレームが一定時間来なければ`read()`が失敗します。つまり`max_missed_pongs`は「**もっと速く**無応答を検出したい」ときに使うオプションだと考えてください。
+
 > 接続が閉じたときの処理は[W03. 接続クローズをハンドリングする](w03-websocket-close)を参照してください。
@@ -205,6 +205,10 @@
 #define CPPHTTPLIB_WEBSOCKET_PING_INTERVAL_SECOND 30
 #endif

+#ifndef CPPHTTPLIB_WEBSOCKET_MAX_MISSED_PONGS
+#define CPPHTTPLIB_WEBSOCKET_MAX_MISSED_PONGS 0
+#endif
+
 /*
 * Headers
 */
@@ -1720,6 +1724,8 @@ public:
  Server &set_websocket_ping_interval(
      const std::chrono::duration<Rep, Period> &duration);

+  Server &set_websocket_max_missed_pongs(int count);
+
  bool bind_to_port(const std::string &host, int port, int socket_flags = 0);
  int bind_to_any_port(const std::string &host, int socket_flags = 0);
  bool listen_after_bind();
@@ -1756,6 +1762,7 @@ protected:
  size_t payload_max_length_ = CPPHTTPLIB_PAYLOAD_MAX_LENGTH;
  time_t websocket_ping_interval_sec_ =
      CPPHTTPLIB_WEBSOCKET_PING_INTERVAL_SECOND;
+  int websocket_max_missed_pongs_ = CPPHTTPLIB_WEBSOCKET_MAX_MISSED_PONGS;

 private:
  using Handlers =
@@ -3728,17 +3735,21 @@ private:

  WebSocket(
      Stream &strm, const Request &req, bool is_server,
-      time_t ping_interval_sec = CPPHTTPLIB_WEBSOCKET_PING_INTERVAL_SECOND)
+      time_t ping_interval_sec = CPPHTTPLIB_WEBSOCKET_PING_INTERVAL_SECOND,
+      int max_missed_pongs = CPPHTTPLIB_WEBSOCKET_MAX_MISSED_PONGS)
      : strm_(strm), req_(req), is_server_(is_server),
-        ping_interval_sec_(ping_interval_sec) {
+        ping_interval_sec_(ping_interval_sec),
+        max_missed_pongs_(max_missed_pongs) {
    start_heartbeat();
  }

  WebSocket(
      std::unique_ptr<Stream> &&owned_strm, const Request &req, bool is_server,
-      time_t ping_interval_sec = CPPHTTPLIB_WEBSOCKET_PING_INTERVAL_SECOND)
+      time_t ping_interval_sec = CPPHTTPLIB_WEBSOCKET_PING_INTERVAL_SECOND,
+      int max_missed_pongs = CPPHTTPLIB_WEBSOCKET_MAX_MISSED_PONGS)
      : strm_(*owned_strm), owned_strm_(std::move(owned_strm)), req_(req),
-        is_server_(is_server), ping_interval_sec_(ping_interval_sec) {
+        is_server_(is_server), ping_interval_sec_(ping_interval_sec),
+        max_missed_pongs_(max_missed_pongs) {
    start_heartbeat();
  }

@@ -3750,6 +3761,8 @@ private:
  Request req_;
  bool is_server_;
  time_t ping_interval_sec_;
+  int max_missed_pongs_;
+  int unacked_pings_ = 0;
  std::atomic<bool> closed_{false};
  std::mutex write_mutex_;
  std::thread ping_thread_;
@@ -3779,6 +3792,7 @@ public:
  void set_read_timeout(time_t sec, time_t usec = 0);
  void set_write_timeout(time_t sec, time_t usec = 0);
  void set_websocket_ping_interval(time_t sec);
+  void set_websocket_max_missed_pongs(int count);
  void set_tcp_nodelay(bool on);
  void set_address_family(int family);
  void set_ipv6_v6only(bool on);
@@ -3810,6 +3824,7 @@ private:
  time_t write_timeout_usec_ = CPPHTTPLIB_CLIENT_WRITE_TIMEOUT_USECOND;
  time_t websocket_ping_interval_sec_ =
      CPPHTTPLIB_WEBSOCKET_PING_INTERVAL_SECOND;
+  int websocket_max_missed_pongs_ = CPPHTTPLIB_WEBSOCKET_MAX_MISSED_PONGS;
  int address_family_ = AF_UNSPEC;
  bool tcp_nodelay_ = CPPHTTPLIB_TCP_NODELAY;
  bool ipv6_v6only_ = CPPHTTPLIB_IPV6_V6ONLY;
@@ -10912,6 +10927,11 @@ inline Server &Server::set_payload_max_length(size_t length) {
  return *this;
 }

+inline Server &Server::set_websocket_max_missed_pongs(int count) {
+  websocket_max_missed_pongs_ = count;
+  return *this;
+}
+
 inline Server &Server::set_websocket_ping_interval(time_t sec) {
  websocket_ping_interval_sec_ = sec;
  return *this;
@@ -12050,7 +12070,8 @@ Server::process_request(Stream &strm, const std::string &remote_addr,
        {
          // Use WebSocket-specific read timeout instead of HTTP timeout
          strm.set_read_timeout(CPPHTTPLIB_WEBSOCKET_READ_TIMEOUT_SECOND, 0);
-          ws::WebSocket ws(strm, req, true, websocket_ping_interval_sec_);
+          ws::WebSocket ws(strm, req, true, websocket_ping_interval_sec_,
+                           websocket_max_missed_pongs_);
          entry.handler(req, ws);
        }
        return true;
@@ -19700,7 +19721,11 @@ inline ReadResult WebSocket::read(std::string &msg) {
                                    payload.size(), true, !is_server_);
      continue;
    }
-    case Opcode::Pong: continue;
+    case Opcode::Pong: {
+      std::lock_guard<std::mutex> lock(ping_mutex_);
+      unacked_pings_ = 0;
+      continue;
+    }
    case Opcode::Close: {
      if (!closed_.exchange(true)) {
        // Echo close frame back
@@ -19734,7 +19759,11 @@ inline ReadResult WebSocket::read(std::string &msg) {
                true, !is_server_);
            continue;
          }
-          if (cont_opcode == Opcode::Pong) { continue; }
+          if (cont_opcode == Opcode::Pong) {
+            std::lock_guard<std::mutex> lock(ping_mutex_);
+            unacked_pings_ = 0;
+            continue;
+          }
          if (cont_opcode == Opcode::Close) {
            if (!closed_.exchange(true)) {
              std::lock_guard<std::mutex> lock(write_mutex_);
@@ -19822,12 +19851,22 @@ inline void WebSocket::start_heartbeat() {
    while (!closed_) {
      ping_cv_.wait_for(lock, std::chrono::seconds(ping_interval_sec_));
      if (closed_) { break; }
+      // If the peer has failed to respond to the previous pings, give up.
+      // RFC 6455 does not define a pong-timeout mechanism; this is an
+      // opt-in liveness check controlled by max_missed_pongs_.
+      if (max_missed_pongs_ > 0 && unacked_pings_ >= max_missed_pongs_) {
+        lock.unlock();
+        close(CloseStatus::GoingAway, "pong timeout");
+        return;
+      }
      lock.unlock();
      if (!send_frame(Opcode::Ping, nullptr, 0)) {
+        lock.lock();
        closed_ = true;
        break;
      }
      lock.lock();
+      unacked_pings_++;
    }
  });
 }
@@ -19955,8 +19994,9 @@ inline bool WebSocketClient::connect() {
  Request req;
  req.method = "GET";
  req.path = path_;
-  ws_ = std::unique_ptr<WebSocket>(
-      new WebSocket(std::move(strm), req, false, websocket_ping_interval_sec_));
+  ws_ = std::unique_ptr<WebSocket>(new WebSocket(std::move(strm), req, false,
+                                                 websocket_ping_interval_sec_,
+                                                 websocket_max_missed_pongs_));
  return true;
 }

@@ -20000,6 +20040,10 @@ inline void WebSocketClient::set_websocket_ping_interval(time_t sec) {
  websocket_ping_interval_sec_ = sec;
 }

+inline void WebSocketClient::set_websocket_max_missed_pongs(int count) {
+  websocket_max_missed_pongs_ = count;
+}
+
 inline void WebSocketClient::set_tcp_nodelay(bool on) { tcp_nodelay_ = on; }

 inline void WebSocketClient::set_address_family(int family) {
@@ -136,6 +136,86 @@ TEST_F(WebSocketServerPingIntervalTest, ServerRuntimeInterval) {
  client.close();
 }

+// Verify that the client detects a non-responsive peer via unacked-ping count.
+// Setup: the server's heartbeat is disabled AND its handler never calls
+// read(), so no automatic Pong reply is ever produced. The client sends
+// pings but receives no pongs, and should close itself once the unacked
+// ping count reaches max_missed_pongs.
+class WebSocketPongTimeoutTest : public ::testing::Test {
+protected:
+  void SetUp() override {
+    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");
+    thread_ = std::thread([this]() { svr_.listen_after_bind(); });
+    svr_.wait_until_ready();
+  }
+
+  void TearDown() override {
+    {
+      std::lock_guard<std::mutex> lock(handler_mutex_);
+      release_ = true;
+    }
+    handler_cv_.notify_all();
+    svr_.stop();
+    thread_.join();
+  }
+
+  Server svr_;
+  int port_;
+  std::thread thread_;
+  std::mutex handler_mutex_;
+  std::condition_variable handler_cv_;
+  bool release_ = false;
+};
+
+TEST_F(WebSocketPongTimeoutTest, ClientDetectsNonResponsivePeer) {
+  ws::WebSocketClient client("ws://localhost:" + std::to_string(port_) + "/ws");
+  client.set_websocket_max_missed_pongs(2);
+  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. Poll is_open() up to 6s.
+  auto start = std::chrono::steady_clock::now();
+  while (client.is_open() &&
+         std::chrono::steady_clock::now() - start < std::chrono::seconds(6)) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+  }
+
+  EXPECT_FALSE(client.is_open());
+}
+
+// Verify that a responsive peer does NOT trigger the pong-timeout mechanism,
+// even with a small max_missed_pongs budget. This is the positive counterpart
+// of ClientDetectsNonResponsivePeer: the client must actively drive read() so
+// that incoming Pong frames are consumed and the unacked counter is reset.
+TEST_F(WebSocketHeartbeatTest, ResponsivePeerNeverTimesOut) {
+  ws::WebSocketClient client("ws://localhost:" + std::to_string(port_) + "/ws");
+  client.set_websocket_max_missed_pongs(2);
+  ASSERT_TRUE(client.connect());
+
+  // Interactive loop over ~6s, longer than 2 ping intervals, so the
+  // pong-timeout mechanism would trigger if pongs weren't being consumed.
+  // Each iteration's read() also drains any pending Pong frame.
+  for (int i = 0; i < 6; i++) {
+    std::string text = "keepalive" + std::to_string(i);
+    ASSERT_TRUE(client.send(text));
+    std::string msg;
+    ASSERT_TRUE(client.read(msg));
+    EXPECT_EQ(text, msg);
+    std::this_thread::sleep_for(std::chrono::seconds(1));
+  }
+
+  EXPECT_TRUE(client.is_open());
+  client.close();
+}
+
 // Verify that multiple heartbeat cycles work
 TEST_F(WebSocketHeartbeatTest, MultipleHeartbeatCycles) {
  ws::WebSocketClient client("ws://localhost:" + std::to_string(port_) + "/ws");