ws: make pong sending lazy

Do not send PONG frames unless there is sufficient space left in the
websocket send buffer. A server might be lazy in reading our data and
intermediary PONG frames can be skipped by a client (RFC 6455, ch.
5.5.3).

Add test case measuring no real RSS increase on a server blasting with
PING frames.

Closes #21911

lib/ws.c

@@ -632,6 +632,7 @@ static CURLcode ws_enc_add_cntrl(struct Curl_easy *data,
                                  size_t plen,
                                  unsigned int frame_type)
 {
+  (void)data;
   DEBUGASSERT(plen <= WS_MAX_CNTRL_LEN);
   if(plen > WS_MAX_CNTRL_LEN)
     return CURLE_BAD_FUNCTION_ARGUMENT;
@@ -641,13 +642,6 @@ static CURLcode ws_enc_add_cntrl(struct Curl_easy *data,
   ws->pending.type = frame_type;
   ws->pending.payload_len = plen;
   memcpy(ws->pending.payload, payload, plen);
-
-  if(!ws->enc.payload_remain) { /* not in the middle of another frame */
-    CURLcode result = ws_enc_add_pending(data, ws);
-    if(!result)
-      (void)ws_flush(data, ws, Curl_is_in_callback(data));
-    return result;
-  }
   return CURLE_OK;
 }
 
@@ -716,7 +710,7 @@ static CURLcode ws_cw_write(struct Curl_easy *data,
 {
   struct ws_cw_ctx *ctx = writer->ctx;
   struct websocket *ws;
-  CURLcode result;
+  CURLcode result = CURLE_OK;
 
   CURL_TRC_WRITE(data, "ws_cw_write(len=%zu, type=%d)", nbytes, type);
   if(!(type & CLIENTWRITE_BODY) || data->set.ws_raw_mode)
@@ -749,7 +743,8 @@ static CURLcode ws_cw_write(struct Curl_easy *data,
     if(result == CURLE_AGAIN) {
       /* insufficient amount of data, keep it for later.
        * we pretend to have written all since we have a copy */
-      return CURLE_OK;
+      result = CURLE_OK;
+      goto out;
     }
     else if(result) {
       failf(data, "[WS] decode payload error %d", (int)result);
@@ -760,10 +755,16 @@ static CURLcode ws_cw_write(struct Curl_easy *data,
   if((type & CLIENTWRITE_EOS) && !Curl_bufq_is_empty(&ctx->buf)) {
     failf(data, "[WS] decode ending with %zu frame bytes remaining",
           Curl_bufq_len(&ctx->buf));
-    return CURLE_RECV_ERROR;
+    result = CURLE_RECV_ERROR;
   }
 
-  return CURLE_OK;
+out:
+  if(!result) {
+    result = ws_flush(data, ws, Curl_is_in_callback(data));
+    if(result == CURLE_AGAIN)
+      result = CURLE_OK;
+  }
+  return result;
 }
 
 /* WebSocket payload decoding client writer. */
@@ -1627,8 +1628,16 @@ CURLcode curl_ws_recv(CURL *curl, void *buffer,
 static CURLcode ws_flush(struct Curl_easy *data, struct websocket *ws,
                          bool blocking)
 {
+  CURLcode result;
+
+  /* If there is space, add any pending control frame */
+  if(Curl_bufq_len(&ws->sendbuf) < ws->sendbuf.chunk_size) {
+    result = ws_enc_add_pending(data, ws);
+    if(result && (result != CURLE_AGAIN))
+      return result;
+  }
+
   if(!Curl_bufq_is_empty(&ws->sendbuf)) {
-    CURLcode result;
     const uint8_t *out;
     size_t outlen, n;
 #ifdef DEBUGBUILD

tests/http/test_20_websockets.py

@@ -24,11 +24,15 @@
 #
 ###########################################################################
 #
+import base64
+import hashlib
 import logging
 import os
+import re
 import shutil
 import socket
 import subprocess
+import threading
 import time
 from datetime import datetime, timedelta
 from typing import Dict
@@ -220,3 +224,77 @@ class TestWebsockets:
         # The CONNECT through the proxy fails as it does not allow it
         r.check_exit_code(7) # CURLE_COULDNT_CONNECT
         assert r.stats[0]['http_connect'] == 403, f'{r}'
+
+    def test_20_11_crazy_pings(self, env: Env):
+        st = {}
+        send_rounds = 1
+
+        def srv():
+            try:
+                with socket.socket() as s:
+                    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                    s.bind(("127.0.0.1", 0))
+                    s.listen(1)
+                    st["p"] = s.getsockname()[1]
+
+                    c, _ = s.accept()
+                    c.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096)
+                    c.settimeout(Env.SERVER_TIMEOUT)
+                    req = b""
+                    while b"\r\n\r\n" not in req:
+                        req += c.recv(4096)
+
+                    k = re.search(rb"(?im)^Sec-WebSocket-Key:\s*(\S+)", req).group(1)
+                    a = base64.b64encode(
+                        hashlib.sha1(k + b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11").digest()
+                    ).decode()
+                    c.sendall(
+                        (
+                            "HTTP/1.1 101 Switching Protocols\r\n"
+                            "Upgrade: websocket\r\n"
+                            "Connection: Upgrade\r\n"
+                            f"Sec-WebSocket-Accept: {a}\r\n\r\n"
+                        ).encode()
+                    )
+
+                    f = b"\x89\x00" * 65536  # PING frames, many
+                    try:
+                        for _ in range(send_rounds):
+                            c.sendall(f)
+                        f = b"\x88\x00"  # CLOSE frame
+                        c.sendall(f)
+                    except OSError:
+                        pass
+                    time.sleep(1)
+                    c.close()
+            except OSError as e:
+                st["err"] = e
+
+        curl = CurlClient(env=env)
+        send_rounds = 2
+        threading.Thread(target=srv, daemon=True).start()
+        while "p" not in st and "err" not in st:
+            time.sleep(0.01)
+        assert "err" not in st, f'ws-ping server failed to start: {st["err"]}'
+
+        url = f'ws://127.0.0.1:{st["p"]}/'
+        r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
+                               with_profile=True)
+        assert r.exit_code in [55, 56], f'{r.dump_logs()}'  # SEND/RECV_ERROR
+        assert r.profile, f'{r}'
+        rss1 = r.profile.stats['rss'] / (1024 * 1024)
+
+        st.clear()
+        send_rounds = 10
+        threading.Thread(target=srv, daemon=True).start()
+        while "p" not in st and "err" not in st:
+            time.sleep(0.01)
+        assert "err" not in st, f'ws-ping server failed to start: {st["err"]}'
+
+        url = f'ws://127.0.0.1:{st["p"]}/'
+        r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
+                               with_profile=True)
+        assert r.exit_code in [55, 56], f'{r.dump_logs()}'  # SEND/RECV_ERROR
+        assert r.profile, f'{r}'
+        rss2 = r.profile.stats['rss'] / (1024 * 1024)
+        assert (rss1 * 1.1) >= rss2, 'bad memory increase'