HTTPServer: disconnect clients

diff --git a/src/httpserver.cpp b/src/httpserver.cpp
index 572fa140..8467c1f9 100644
--- a/src/httpserver.cpp
+++ b/src/httpserver.cpp
@@ -986,8 +986,7 @@ void HTTPRequest::WriteReply(HTTPStatusCode status, std::span<const std::byte> r
     bool needs_body{status != HTTP_NO_CONTENT && (status < 100 || status >= 200)};
     bool needs_content_length{false};
 
-    // Keep track of this, will be used in a future commit to disconnect some clients
-    [[maybe_unused]] bool keep_alive{false};
+    bool keep_alive{false};
 
     // See libevent evhttp_make_header_response()
     // Expected response headers depend on protocol version
@@ -1035,6 +1034,8 @@ void HTTPRequest::WriteReply(HTTPStatusCode status, std::span<const std::byte> r
         keep_alive = false;
     }
 
+    m_client->m_keep_alive = keep_alive;
+
     // Serialize the response headers
     const std::string headers{res.StringifyHeaders()};
     const auto headers_bytes{std::as_bytes(std::span{headers})};
@@ -1158,18 +1159,6 @@ void HTTPServer::JoinSocketsThreads()
     }
 }
 
-bool HTTPServer::CloseConnection(const HTTPRemoteClient& http_client)
-{
-    size_t erased = std::erase_if(m_connected,
-                                  [&](auto& client){ return client->m_id == http_client.m_id; });
-    if (erased > 0) {
-         // Report back to the main thread
-         m_connected_size.fetch_sub(erased, std::memory_order_relaxed);
-         return true;
-    }
-    return false;
-}
-
 std::unique_ptr<Sock> HTTPServer::AcceptConnection(const Sock& listen_sock, CService& addr)
 {
     // Make sure we only operate on our own listening sockets
@@ -1276,7 +1265,7 @@ void HTTPServer::SocketHandlerConnected(const IOReadiness& io_readiness) const
                         client->m_origin,
                         client->m_id,
                         NetworkErrorString(err));
-                    // TODO: Disconnect
+                    client->m_disconnect = true;
                 }
             } else if (nrecv == 0) {
                 LogDebug(
@@ -1284,8 +1273,11 @@ void HTTPServer::SocketHandlerConnected(const IOReadiness& io_readiness) const
                     "Received EOF from %s (id=%llu)",
                     client->m_origin,
                     client->m_id);
-                // TODO: Disconnect
+                client->m_disconnect = true;
             } else {
+                // Prevent disconnect until all requests are completely handled.
+                client->m_connection_busy = true;
+
                 // Copy data from socket buffer to client receive buffer
                 client->m_recv_buffer.insert(
                     client->m_recv_buffer.end(),
@@ -1359,6 +1351,9 @@ void HTTPServer::ThreadSocketHandler()
 
         // Accept new connections from listening sockets.
         SocketHandlerListening(io_readiness.events_per_sock);
+
+        // Disconnect any clients that have been flagged.
+        DisconnectClients();
     }
 }
 
@@ -1371,6 +1366,17 @@ void HTTPServer::MaybeDispatchRequestsFromClient(const std::shared_ptr<HTTPRemot
         try {
             // Stop reading if we need more data from the client to parse a complete request
             if (!client->ReadRequest(*req)) break;
+        } catch (const ContentTooLargeError& e) {
+            LogDebug(
+                BCLog::HTTP,
+                "HTTP request body too large from client %s (id=%llu): %s",
+                client->m_origin,
+                client->m_id,
+                e.what());
+
+            req->WriteReply(HTTP_CONTENT_TOO_LARGE);
+            client->m_disconnect = true;
+            return;
         } catch (const std::runtime_error& e) {
             LogDebug(
                 BCLog::HTTP,
@@ -1380,8 +1386,8 @@ void HTTPServer::MaybeDispatchRequestsFromClient(const std::shared_ptr<HTTPRemot
                 e.what());
 
             // We failed to read a complete request from the buffer
-            // TODO: respond with HTTP_BAD_REQUEST and disconnect
-
+            req->WriteReply(HTTP_BAD_REQUEST);
+            client->m_disconnect = true;
             return;
         }
 
@@ -1399,6 +1405,46 @@ void HTTPServer::MaybeDispatchRequestsFromClient(const std::shared_ptr<HTTPRemot
     }
 }
 
+void HTTPServer::DisconnectClients()
+{
+    size_t erased = std::erase_if(m_connected,
+                                  [&](auto& client) {
+                                        // Disconnect this client due to error or end of communication.
+                                        // May drop unsent data if we are closing due to error.
+                                        if (client->m_disconnect) {
+                                            ;
+                                        } else {
+                                            // Disconnect this client because the server is shutting
+                                            // down and we need to disconnect all clients...
+                                            if (m_disconnect_all_clients) {
+                                                // ...unless we still have data for this client.
+                                                if (client->m_connection_busy) {
+                                                    // There is still data for this healthy-connected client.
+                                                    // Continue the I/O loop until all data is sent or an error is encountered.
+                                                    return false;
+                                                } else {
+                                                    // This is a healthy persistent connection (e.g. keep-alive)
+                                                    // but it's time to say goodbye.
+                                                    ;
+                                                }
+                                            } else {
+                                                // No reason to disconnect.
+                                                return false;
+                                            }
+                                        }
+                                        // No reason NOT to disconnect, log and remove.
+                                        LogDebug(BCLog::HTTP,
+                                                 "Disconnecting HTTP client %s (id=%llu)",
+                                                 client->m_origin,
+                                                 client->m_id);
+                                        return true;
+                                    });
+    if (erased > 0) {
+        // Report back to the main thread
+        m_connected_size.fetch_sub(erased, std::memory_order_relaxed);
+    }
+}
+
 bool HTTPRemoteClient::ReadRequest(HTTPRequest& req)
 {
     LineReader reader(m_recv_buffer, MAX_HEADERS_SIZE);
@@ -1447,6 +1493,7 @@ bool HTTPRemoteClient::MaybeSendBytesFromBuffer()
             if (!IOErrorIsPermanent(err)) {
                 // The error can be safely ignored, try the send again on the next I/O loop.
                 m_send_ready = true;
+                m_connection_busy = true;
                 return true;
             } else {
                 // Unrecoverable error, log and disconnect client.
@@ -1456,7 +1503,9 @@ bool HTTPRemoteClient::MaybeSendBytesFromBuffer()
                     m_origin,
                     m_id,
                     NetworkErrorString(err));
-                // TODO: disconnect
+                m_send_ready = false;
+                m_disconnect = true;
+
                 // Do not attempt to read from this client.
                 return false;
             }
@@ -1479,9 +1528,14 @@ bool HTTPRemoteClient::MaybeSendBytesFromBuffer()
         // on an already-empty m_send_buffer because the connection might have just been opened.
         if (m_send_buffer.empty()) {
             m_send_ready = false;
-        } else {
-            // The send buffer isn't flushed yet, try to push more on the next loop.
-            m_send_ready = true;
+            m_connection_busy = false;
+
+            // Our work is done here
+            if (!m_keep_alive) {
+                m_disconnect = true;
+                // Do not attempt to read from this client.
+                return false;
+            }
         }
     }
 
diff --git a/src/httpserver.h b/src/httpserver.h
index 26345c0f..4018837d 100644
--- a/src/httpserver.h
+++ b/src/httpserver.h
@@ -336,7 +336,7 @@ public:
     virtual ~HTTPServer()
     {
         Assume(!m_thread_socket_handler.joinable()); // Missing call to JoinSocketsThreads()
-        Assume(m_connected.empty()); // Missing call to CloseConnection()
+        Assume(m_connected.empty()); // Missing call to DisconnectClients(), or disconnect flags not set
         Assume(m_listen.empty()); // Missing call to StopListening()
     }
 
@@ -378,13 +378,9 @@ public:
     void InterruptNet() { m_interrupt_net(); }
 
     /**
-     * Destroy a given connection by closing its socket and release resources occupied by it.
-     * This is a temporary method that will be removed in a future commit when it is
-     * replaced by DisconnectClients() and run automatically in the I/O thread.
-     * @param[in] http_client Connection to destroy.
-     * @return Whether the connection existed and its socket was closed by this call.
+     * Start disconnecting clients when possible in the I/O loop
      */
-    bool CloseConnection(const HTTPRemoteClient& http_client);
+    void DisconnectAllClients() { m_disconnect_all_clients = true; }
 
 private:
     /**
@@ -405,6 +401,13 @@ private:
      */
     std::vector<std::shared_ptr<HTTPRemoteClient>> m_connected;
 
+    /**
+     * Flag used during shutdown.
+     * Overrides HTTPRemoteClient flags m_keep_alive and m_connection_busy.
+     * Set by main thread and read by the I/O thread.
+     */
+    std::atomic_bool m_disconnect_all_clients{false};
+
     /**
      * The number of connected sockets.
      * Updated from the I/O thread but safely readable from
@@ -506,6 +509,14 @@ private:
      * @param[in] client The HTTPRemoteClient to read requests from
      */
     void MaybeDispatchRequestsFromClient(const std::shared_ptr<HTTPRemoteClient>& client) const;
+
+    /**
+     * Close underlying socket connections for flagged clients
+     * by removing their shared pointer from m_connected. If an HTTPRemoteClient
+     * is busy in a worker thread, its connection will be closed once that
+     * job is done and the HTTPRequest is out of scope.
+     */
+    void DisconnectClients();
 };
 
 class HTTPRemoteClient
@@ -559,6 +570,25 @@ public:
      */
     std::shared_ptr<Sock> m_sock GUARDED_BY(m_sock_mutex);
 
+    //! Initialized to true while server waits for first request from client.
+    //! Set to false after data is written to m_send_buffer and then that buffer is flushed to client.
+    //! Reset to true when we receive new request data from client.
+    //! Checked during DisconnectClients(). All of these operations take place in the HTTPServer I/O loop.
+    //! `m_connection_busy=true` can be overridden by `m_disconnect=true` (we disconnect).
+    bool m_connection_busy{true};
+
+    //! Client has requested to keep the connection open after all requests have been responded to.
+    //! Set by (potentially multiple) worker threads and checked in the HTTPServer I/O loop.
+    //! `m_keep_alive=true` can be overridden `by HTTPServer.m_disconnect_all_clients` (we disconnect).
+    std::atomic_bool m_keep_alive{false};
+
+    //! Flag this client for disconnection on next loop.
+    //! Either we have encountered a permanent error, or both sides of the socket are done
+    //! with the connection, e.g. our reply to a "Connection: close" request has been sent.
+    //! Might be set in a worker thread or in the I/O thread. When set to `true` we disconnect,
+    //! possibly overriding all other disconnect flags.
+    std::atomic_bool m_disconnect{false};
+
     explicit HTTPRemoteClient(HTTPServer::Id id, const CService& addr, std::unique_ptr<Sock> socket)
         : m_id(id), m_addr(addr), m_origin(addr.ToStringAddrPort()), m_sock{std::move(socket)} {};
 
diff --git a/src/rpc/protocol.h b/src/rpc/protocol.h
index 0ba17ae0..99e8f05f 100644
--- a/src/rpc/protocol.h
+++ b/src/rpc/protocol.h
@@ -16,6 +16,7 @@ enum HTTPStatusCode : int
     HTTP_FORBIDDEN             = 403,
     HTTP_NOT_FOUND             = 404,
     HTTP_BAD_METHOD            = 405,
+    HTTP_CONTENT_TOO_LARGE     = 413,
     HTTP_INTERNAL_SERVER_ERROR = 500,
     HTTP_SERVICE_UNAVAILABLE   = 503,
 };
@@ -32,6 +33,7 @@ inline std::string_view HTTPStatusReasonString(HTTPStatusCode code)
     case HTTP_FORBIDDEN: return "Forbidden";
     case HTTP_NOT_FOUND: return "Not Found";
     case HTTP_BAD_METHOD: return "Method Not Allowed";
+    case HTTP_CONTENT_TOO_LARGE: return "Content too large";
     case HTTP_INTERNAL_SERVER_ERROR: return "Internal Server Error";
     case HTTP_SERVICE_UNAVAILABLE: return "Service Unavailable";
     }
diff --git a/src/test/httpserver_tests.cpp b/src/test/httpserver_tests.cpp
index 0470e166..c9b8eea5 100644
--- a/src/test/httpserver_tests.cpp
+++ b/src/test/httpserver_tests.cpp
@@ -594,13 +594,18 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests)
     BOOST_CHECK(actual.find("Content-Type: text/html; charset=ISO-8859-1\r\n") != std::string::npos);
     BOOST_CHECK(actual.find("Date: Wed, 11 Dec 2024 00:47:09 GMT\r\n") != std::string::npos);
 
-    // Stop the I/O thread before modifying m_connected. CloseConnection() is a
-    // temporary test-only API that is not thread-safe against GenerateWaitSockets(),
-    // which iterates m_connected on the I/O thread.
+    // Wait up to one minute for connection to be automatically closed, because
+    // keep-alive was not set by the client and we are done responding to their request.
+    attempts = 6000;
+    while (server.GetConnectionsCount() != 0) {
+        std::this_thread::sleep_for(10ms);
+        BOOST_REQUIRE(--attempts > 0);
+    }
+
+    // Stop the I/O loop and shutdown
     server.InterruptNet();
+    // Wait for I/O loop to finish, after all connected sockets are closed
     server.JoinSocketsThreads();
-    // Close connection
-    BOOST_REQUIRE(server.CloseConnection(*client));
     // Close all listening sockets
     server.StopListening();
 }