fix(http1): close a finished connection instead of holding it until reap

A connection's socket was owned by the registry entry and only released
when the next accept() reaped it, so a peer we had finished with — after
a 408, a 400, or a `connection: close` — never saw EOF and sat waiting
for a server that was done talking. On a server that goes quiet it also
held every descriptor from the last burst indefinitely.

The connection thread now closes its own socket the moment Serve()
returns, under the registry lock so Stop()'s shutdown() can never name a
descriptor that has already been released, and Stop() waits on a
condition variable for the last thread rather than assuming the vector
it moved out is quiescent. Adopt() is also fully guarded: it runs on
ListenerTCP's accept loop, which has no handler, so anything escaping it
would abort the process.

Found by ShouldSurviveAbuseHTTP1, added here: 24 concurrent keep-alive
clients, peers that vanish mid-request or send garbage, and a peer that
stalls forever — the server must keep serving and still stop promptly.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

implementations/Crafter.Network-ListenerHTTP1.cpp

@@ -65,6 +65,9 @@ struct ListenerHTTP1::Impl {
     ListenerHTTP1* owner = nullptr;
     std::unique_ptr<ListenerTCP> listener;
     std::mutex mutex;
+    // Signalled when a connection thread finishes, so Stop() can wait for
+    // the last one instead of polling.
+    std::condition_variable idle;
     std::vector<std::unique_ptr<HTTP1Connection>> connections;
     std::atomic<bool> running{true};
     std::atomic<std::uint64_t> accepted{0};
@@ -79,7 +82,10 @@ struct ListenerHTTP1::Impl {
         });
     }
 
-    void Adopt(ClientTCP* accepted) {
+    void Adopt(ClientTCP* accepted) try {
+        // The unique_ptr takes ownership immediately, so every path out of
+        // here — including the shutdown early-return and a throwing thread
+        // constructor — closes the socket.
         auto connection = std::make_unique<HTTP1Connection>();
         connection->client.reset(accepted);
         HTTP1Connection* pointer = connection.get();
@@ -94,9 +100,24 @@ struct ListenerHTTP1::Impl {
             } catch (...) {
                 // A connection dying must never take the server with it.
             }
-            pointer->finished.store(true);
+            {
+                std::lock_guard lock(mutex);
+                // Close here rather than when the entry is reaped: the peer
+                // has to see the connection end as soon as we are done with
+                // it, and holding the descriptor until the next accept
+                // would be an unbounded leak on a server that goes quiet.
+                // Doing it under the lock keeps Stop()'s shutdown() from
+                // ever touching a descriptor number we have released.
+                pointer->client.reset();
+                pointer->finished.store(true);
+            }
+            idle.notify_all();
         });
         connections.push_back(std::move(connection));
+    } catch (...) {
+        // This runs on the accept loop, which has no handler of its own:
+        // letting anything escape (a thread that could not be spawned, say)
+        // would abort the process.
     }
 
     void Send(ClientTCP& client, const std::string& wire) {
@@ -261,12 +282,18 @@ void ListenerHTTP1::Stop() {
 
     std::vector<std::unique_ptr<HTTP1Connection>> closing;
     {
-        std::lock_guard lock(impl->mutex);
+        std::unique_lock lock(impl->mutex);
         for (auto& connection : impl->connections) {
             // Wake the serving thread out of poll()/recv() without closing
-            // the descriptor underneath it.
+            // the descriptor underneath it — the thread owns that.
             if (connection->client) shutdown(connection->client->socketid, SHUT_RDWR);
         }
+        impl->idle.wait(lock, [&] {
+            return std::ranges::all_of(impl->connections,
+                                       [](const std::unique_ptr<HTTP1Connection>& connection) {
+                                           return connection->finished.load();
+                                       });
+        });
         closing = std::move(impl->connections);
         impl->connections.clear();
     }

project.cpp

@@ -131,6 +131,7 @@ extern "C" Configuration CrafterBuildProject(std::span<const std::string_view> a
         cfg.AddTest("ShouldSendRecieveLargeHTTP").Dependencies({ &cfg });
         cfg.AddTest("ShouldSendRecieveQUICDatagram").Dependencies({ &cfg });
         cfg.AddTest("ShouldSendRecieveQUICStream").Dependencies({ &cfg });
+        cfg.AddTest("ShouldSurviveAbuseHTTP1").Dependencies({ &cfg });
     }
 
     return cfg;

tests/ShouldSurviveAbuseHTTP1/main.cpp

@@ -0,0 +1,127 @@
+//SPDX-License-Identifier: LGPL-3.0-only
+//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
+
+// A server is only as good as its worst peer. This drives ListenerHTTP1
+// with many concurrent clients and then with peers that behave badly —
+// vanishing mid-request, sending garbage, opening and dropping connections,
+// stalling forever — and requires that it keeps serving correctly
+// throughout and still shuts down promptly at the end.
+
+import Crafter.Network;
+import std;
+using namespace Crafter;
+
+namespace {
+    std::atomic<int> failures{0};
+
+    void Check(bool condition, std::string_view what) {
+        if (!condition) {
+            std::println("FAIL: {}", what);
+            failures.fetch_add(1);
+        }
+    }
+
+    constexpr std::uint16_t port = 8095;
+
+    // Raw socket, no HTTP client in the way, so the test can send things a
+    // well-behaved client never would.
+    void SendRaw(std::string_view bytes, bool waitForReply) {
+        ClientTCP socket("localhost", port);
+        if (!bytes.empty()) socket.Send(bytes.data(), static_cast<std::uint32_t>(bytes.size()));
+        if (waitForReply) {
+            try {
+                (void)socket.RecieveSync();
+            } catch (const std::exception&) {
+                // A dropped connection is an acceptable answer to nonsense.
+            }
+        }
+    }
+}
+
+int main() {
+    std::unordered_map<std::string, std::function<HTTPResponse(const HTTPRequest&)>> routes;
+    routes["/work"] = [](const HTTPRequest& request) {
+        return CreateResponseHTTP("200", request.body.empty() ? std::string("empty") : request.body);
+    };
+
+    try {
+        ListenerAsyncHTTP1 listener(port, std::move(routes));
+        // Keep the stalled-peer case quick; the defaults are measured in
+        // tens of seconds, which is right for a real server and wrong for a
+        // test.
+        listener.listener.requestTimeout = std::chrono::milliseconds(300);
+        listener.listener.keepAliveTimeout = std::chrono::milliseconds(500);
+
+        // ── Many clients at once, each reusing its own connection ──────
+        constexpr int clientCount = 24;
+        constexpr int requestsPerClient = 8;
+        std::atomic<int> completed{0};
+        std::vector<std::thread> clients;
+        for (int i = 0; i < clientCount; ++i) {
+            clients.emplace_back([i, &completed] {
+                try {
+                    ClientHTTP1 client("localhost", port);
+                    for (int request = 0; request < requestsPerClient; ++request) {
+                        const std::string payload = std::format("client-{}-request-{}", i, request);
+                        HTTPResponse response = client.Send(
+                            CreateRequestHTTP("POST", "/work", "localhost", payload));
+                        Check(response.status == "200", "concurrent request status");
+                        Check(response.body == payload, "concurrent request body matches its sender");
+                        completed.fetch_add(1);
+                    }
+                } catch (const std::exception& error) {
+                    std::println("client threw: {}", error.what());
+                    failures.fetch_add(1);
+                }
+            });
+        }
+        for (auto& client : clients) client.join();
+        Check(completed.load() == clientCount * requestsPerClient, "every concurrent request finished");
+        Check(listener.listener.AcceptedCount() == clientCount,
+              "each client used exactly one connection");
+
+        // ── Peers that misbehave ──────────────────────────────────────
+        SendRaw("", false);                                     // connect, say nothing, leave
+        SendRaw("GET /work HTTP/1.1\r\nHost: x\r\n", false);     // headers cut short
+        SendRaw("POST /work HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\nshort", false);
+        SendRaw("not http at all\r\n\r\n", true);                // garbage start line
+        SendRaw("GET / HTTP/1.1\r\nHost : x\r\n\r\n", true);     // smuggling-shaped header
+        SendRaw(std::string("GET /") + std::string(16 * 1024, 'z') + " HTTP/1.1\r\n\r\n", true);
+
+        // A peer that opens a request and then just sits there must be cut
+        // loose by the request timeout rather than holding its thread.
+        {
+            ClientTCP stalled("localhost", port);
+            const std::string partial = "POST /work HTTP/1.1\r\nHost: x\r\nContent-Length: 10\r\n\r\nab";
+            stalled.Send(partial.data(), static_cast<std::uint32_t>(partial.size()));
+            const auto start = std::chrono::steady_clock::now();
+            try {
+                (void)stalled.RecieveUntilCloseSync();
+            } catch (const std::exception&) {}
+            const auto elapsed = std::chrono::steady_clock::now() - start;
+            Check(elapsed < std::chrono::seconds(5), "a stalled peer is timed out, not waited on");
+        }
+
+        // ── Still healthy afterwards ──────────────────────────────────
+        ClientHTTP1 client("localhost", port);
+        HTTPResponse response = client.Send(CreateRequestHTTP("POST", "/work", "localhost",
+                                                              std::string("still here")));
+        Check(response.status == "200", "the server survived the abuse");
+        Check(response.body == "still here", "and still answers correctly");
+
+        // Shutdown has to finish quickly even with connections open.
+        const auto start = std::chrono::steady_clock::now();
+        listener.Stop();
+        const auto elapsed = std::chrono::steady_clock::now() - start;
+        Check(elapsed < std::chrono::seconds(5), "Stop() returns promptly with a connection still open");
+    } catch (const std::exception& error) {
+        std::println("threw: {}", error.what());
+        return 1;
+    }
+
+    if (failures.load() != 0) {
+        std::println("{} check(s) failed", failures.load());
+        return 1;
+    }
+    return 0;
+}