Serve pipelined requests without waiting for the keep-alive timeout

A client may pipeline its requests (RFC 9112 9.3.2). The server read the
following request(s) into a per-request SocketStream buffer, discarded them
with the stream after the first response, and then waited in keep_alive()
for socket data that never came, closing the connection after the
keep-alive timeout. Over TLS the bytes stayed decrypted in the TLS library,
where keep_alive() could not see them either.

Create one stream per connection and serve a request that is already
buffered (Stream::is_readable()) without waiting in keep_alive().

Keeping the buffer means an extra CRLF that some clients send after a
request body is now parsed as the next request line, which answered 400
and closed the connection. Ignore one empty line before the request-line,
as RFC 9112 2.2 recommends.

Fixes #2599

httplib.h

@@ -6624,18 +6624,30 @@ inline bool keep_alive(const std::atomic<socket_t> &svr_sock, socket_t sock,
   return false;
 }
 
-template <typename T>
+// `has_buffered_request` reports whether the connection's stream already holds
-inline bool
+// bytes of the next request. A client may pipeline its requests (RFC 9112
-process_server_socket_core(const std::atomic<socket_t> &svr_sock, socket_t sock,
+// 9.3.2), so reading one request can pull the start of the next one into the
-                           size_t keep_alive_max_count,
+// stream's buffer; that request must be served without waiting for the socket
-                           time_t keep_alive_timeout_sec, T callback) {
+// to become readable again, since its bytes are no longer on the socket.
+// `callback` is told whether keep_alive() has just seen the socket go readable.
+template <typename P, typename T>
+inline bool process_server_socket_core(const std::atomic<socket_t> &svr_sock,
+                                       socket_t sock,
+                                       size_t keep_alive_max_count,
+                                       time_t keep_alive_timeout_sec,
+                                       P has_buffered_request, T callback) {
   assert(keep_alive_max_count > 0);
   auto ret = false;
   auto count = keep_alive_max_count;
-  while (count > 0 && keep_alive(svr_sock, sock, keep_alive_timeout_sec)) {
+  while (count > 0) {
+    auto socket_readable = false;
+    if (!has_buffered_request()) {
+      if (!keep_alive(svr_sock, sock, keep_alive_timeout_sec)) { break; }
+      socket_readable = true;
+    }
     auto close_connection = count == 1;
     auto connection_closed = false;
-    ret = callback(close_connection, connection_closed);
+    ret = callback(socket_readable, close_connection, connection_closed);
     if (!ret || connection_closed) { break; }
     count--;
   }
@@ -6649,14 +6661,16 @@ process_server_socket(const std::atomic<socket_t> &svr_sock, socket_t sock,
                       time_t keep_alive_timeout_sec, time_t read_timeout_sec,
                       time_t read_timeout_usec, time_t write_timeout_sec,
                       time_t write_timeout_usec, T callback) {
+  // One stream per connection: its read buffer can already hold the start of
+  // the next, pipelined request.
+  SocketStream strm(sock, read_timeout_sec, read_timeout_usec,
+                    write_timeout_sec, write_timeout_usec);
   return process_server_socket_core(
       svr_sock, sock, keep_alive_max_count, keep_alive_timeout_sec,
-      [&](bool close_connection, bool &connection_closed) {
+      [&]() { return strm.is_readable(); },
-        SocketStream strm(sock, read_timeout_sec, read_timeout_usec,
+      [&](bool socket_readable, bool close_connection,
-                          write_timeout_sec, write_timeout_usec);
+          bool &connection_closed) {
-        // process_server_socket_core() only gets here once keep_alive() has
+        if (socket_readable) { strm.set_readable_hint(); }
-        // seen the socket go readable.
-        strm.set_readable_hint();
         return callback(strm, close_connection, connection_closed);
       });
 }
@@ -10584,13 +10598,16 @@ inline bool process_server_socket_ssl(
     socket_t sock, size_t keep_alive_max_count, time_t keep_alive_timeout_sec,
     time_t read_timeout_sec, time_t read_timeout_usec, time_t write_timeout_sec,
     time_t write_timeout_usec, T callback) {
+  // See process_server_socket(). The TLS library keeps already decrypted bytes
+  // of a pipelined request, which keep_alive() cannot see on the socket.
+  SSLSocketStream strm(sock, session, read_timeout_sec, read_timeout_usec,
+                       write_timeout_sec, write_timeout_usec);
   return process_server_socket_core(
       svr_sock, sock, keep_alive_max_count, keep_alive_timeout_sec,
-      [&](bool close_connection, bool &connection_closed) {
+      [&]() { return strm.is_readable(); },
-        SSLSocketStream strm(sock, session, read_timeout_sec, read_timeout_usec,
+      [&](bool socket_readable, bool close_connection,
-                             write_timeout_sec, write_timeout_usec);
+          bool &connection_closed) {
-        // See the non-TLS path in process_server_socket().
+        if (socket_readable) { strm.set_readable_hint(); }
-        strm.set_readable_hint();
         return callback(strm, close_connection, connection_closed);
       });
 }
@@ -14447,6 +14464,13 @@ Server::process_request(Stream &strm, const std::string &remote_addr,
   // Connection has been closed on client
   if (!line_reader.getline()) { return false; }
 
+  // RFC 9112 2.2: ignore an empty line received before the request-line. Some
+  // clients send an extra CRLF after a request body, which would otherwise be
+  // parsed as the next request on a persistent connection.
+  if (strcmp(line_reader.ptr(), "\r\n") == 0 && !line_reader.getline()) {
+    return false;
+  }
+
   Request req;
   req.start_time_ = std::chrono::steady_clock::now();
   req.remote_addr = remote_addr;

test/test.cc

@@ -11481,6 +11481,134 @@ TEST(KeepAliveTest, MaxCount) {
   }
 }
 
+// A client may pipeline its requests (RFC 9112 9.3.2), so reading one request
+// can pull the next one into the server's buffer. The server must serve it
+// without waiting for the socket to become readable again. The keep-alive
+// timeout outlasts the client's read timeout, so a request left waiting for it
+// shows up as a missing response.
+static void serve_pipelining_routes(Server &svr,
+                                    const std::function<void(int)> &client) {
+  svr.set_keep_alive_timeout(5);
+  svr.Get("/hi/(\\d+)", [](const Request &req, Response &res) {
+    res.set_content("hi " + req.matches[1].str(), "text/plain");
+  });
+  svr.Post("/echo", [](const Request &req, Response &res) {
+    res.set_content("echo " + req.body, "text/plain");
+  });
+
+  auto port = svr.bind_to_any_port(HOST);
+  thread t = thread([&] { svr.listen_after_bind(); });
+  auto se = detail::scope_exit([&] {
+    svr.stop();
+    t.join();
+  });
+  svr.wait_until_ready();
+
+  client(port);
+}
+
+static std::string
+send_pipelined_requests(const std::vector<std::string> &parts) {
+  Server svr;
+  std::string res;
+  serve_pipelining_routes(svr, [&](int port) {
+    EXPECT_TRUE(send_request_in_parts(2, parts, &res, port));
+  });
+  return res;
+}
+
+static void expect_in_order(const std::string &res,
+                            const std::vector<std::string> &bodies) {
+  size_t pos = 0;
+  for (const auto &body : bodies) {
+    pos = res.find(body, pos);
+    ASSERT_NE(std::string::npos, pos) << "missing or out of order: " << body;
+    pos += body.size();
+  }
+}
+
+TEST(KeepAliveTest, PipelinedRequests) {
+  auto res = send_pipelined_requests(
+      {"GET /hi/1 HTTP/1.1\r\nHost: localhost\r\n\r\n"
+       "POST /echo HTTP/1.1\r\nHost: localhost\r\n"
+       "Content-Length: 4\r\n\r\nbody"
+       "GET /hi/3 HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"});
+  expect_in_order(res, {"hi 1", "echo body", "hi 3"});
+}
+
+// The second request starts in the server's buffer and ends on the socket.
+TEST(KeepAliveTest, PipelinedRequestSplitAcrossReads) {
+  auto res =
+      send_pipelined_requests({"GET /hi/1 HTTP/1.1\r\nHost: localhost\r\n\r\n"
+                               "GET /hi/2 HTTP/1.1\r\nHo",
+                               "st: localhost\r\nConnection: close\r\n\r\n"});
+  expect_in_order(res, {"hi 1", "hi 2"});
+}
+
+// RFC 9112 2.2: an empty line before the request-line is ignored, so an extra
+// CRLF after a body does not turn into a 400 for the next request.
+TEST(KeepAliveTest, EmptyLineBeforeRequestLineIsIgnored) {
+  auto res = send_pipelined_requests(
+      {"POST /echo HTTP/1.1\r\nHost: localhost\r\n"
+       "Content-Length: 4\r\n\r\nbody\r\n"
+       "GET /hi/2 HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"});
+  EXPECT_EQ(std::string::npos, res.find("400 Bad Request"));
+  expect_in_order(res, {"echo body", "hi 2"});
+}
+
+// Where an unparsable request ends is unknown, so what follows it in the buffer
+// must not be served as the next request.
+TEST(KeepAliveTest, PipelinedRequestAfterInvalidRequestIsNotServed) {
+  auto res = send_pipelined_requests(
+      {"INVALID REQUEST LINE\r\n\r\n"
+       "GET /hi/1 HTTP/1.1\r\nHost: localhost\r\n\r\n"});
+  EXPECT_EQ("HTTP/1.1 400 Bad Request", res.substr(0, 24));
+  EXPECT_EQ(std::string::npos, res.find("hi 1"));
+}
+
+#ifdef CPPHTTPLIB_OPENSSL_SUPPORT
+// Over TLS, the next request is held by the TLS library as already decrypted
+// data rather than on the socket.
+TEST(KeepAliveTest, SSLPipelinedRequests) {
+  SSLServer svr(SERVER_CERT_FILE, SERVER_PRIVATE_KEY_FILE);
+  std::string res;
+  serve_pipelining_routes(svr, [&](int port) {
+    auto error = Error::Success;
+    auto sock = detail::create_client_socket(
+        HOST, "", port, AF_UNSPEC, false, false, nullptr,
+        /*connection_timeout_sec=*/5, 0,
+        /*read_timeout_sec=*/2, 0,
+        /*write_timeout_sec=*/5, 0, std::string(), error);
+    ASSERT_NE(INVALID_SOCKET, sock);
+    auto sock_se = detail::scope_exit([&] { detail::close_socket(sock); });
+
+    auto ctx = SSL_CTX_new(TLS_client_method());
+    auto ssl = SSL_new(ctx);
+    auto ssl_se = detail::scope_exit([&] {
+      SSL_free(ssl);
+      SSL_CTX_free(ctx);
+    });
+    SSL_set_fd(ssl, static_cast<int>(sock));
+    ASSERT_EQ(1, SSL_connect(ssl));
+
+    // One write, so all three requests arrive in one TLS record
+    const std::string req =
+        "GET /hi/1 HTTP/1.1\r\nHost: localhost\r\n\r\n"
+        "GET /hi/2 HTTP/1.1\r\nHost: localhost\r\n\r\n"
+        "GET /hi/3 HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n";
+    ASSERT_EQ(static_cast<int>(req.size()),
+              SSL_write(ssl, req.data(), static_cast<int>(req.size())));
+
+    char buf[512];
+    int n;
+    while ((n = SSL_read(ssl, buf, sizeof(buf))) > 0) {
+      res.append(buf, static_cast<size_t>(n));
+    }
+  });
+  expect_in_order(res, {"hi 1", "hi 2", "hi 3"});
+}
+#endif
+
 TEST(KeepAliveTest, Issue1041) {
   Server svr;
   svr.set_keep_alive_timeout(3);