test

add_test(httpd_h2_dispatch_overload_test httpd_h2_dispatch_overload_test)

add_executable(httpd_h1_dispatch_overload_test

    httpd_h1_dispatch_overload_test.cpp

)

if(WIN32)

    target_link_libraries (httpd_h1_dispatch_overload_test httppp-static ws2_32)

else()

    target_link_libraries (httpd_h1_dispatch_overload_test httppp-static)

endif()

add_test(httpd_h1_dispatch_overload_test httpd_h1_dispatch_overload_test)

add_executable(url_test

    url.cpp

)

/*******************************************************************************

Copyright (c) 2026, Jan Koester jan.koester@gmx.net

All rights reserved.

Redistribution and use in source and binary forms, with or without

modification, are permitted provided that the following conditions are met:

    * Redistributions of source code must retain the above copyright

      notice, this list of conditions and the following disclaimer.

    * Redistributions in binary form must reproduce the above copyright

      notice, this list of conditions and the following disclaimer in the

      documentation and/or other materials provided with the distribution.

    * Neither the name of the <organization> nor the

      names of its contributors may be used to endorse or promote products

      derived from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND

ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE

DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY

DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES

(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;

LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND

ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS

SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

*******************************************************************************/

// H1 analogue of httpd_h2_dispatch_overload_test: _h1DispatchPool used to have no bound at

// all (unlike _h2DispatchPool, which got h2DispatchQueueMax specifically to fix a production

// incident -- see that test's own header comment). Every offloaded H1 request's socket is

// detached from the event loop (netplus::detachConnection) before being queued, so once every

// worker was tied up, an unbounded pile of detached sockets accumulated: invisible to the idle

// reaper, un-timed until a worker finally freed up. A client that gave up in the meantime left

// proxyplus discovering it far too late and starting its own close against a peer that might

// already be gone -- the real root cause behind sockets observed stuck in LAST_ACK under load

// (see the proxyplus 08-16 netstat lead). See HttpEvent's h1DispatchQueueMax ctor param and

// _dispatchH1Request's submit()-rejection branch (src/httpd.cpp).

//

// Same structure as the H2 test: pin the single offload worker on a gate so it never completes

// on its own, fill the (size-1) queue behind it with one more request, then prove a *third*,

// brand-new connection gets an immediate 503 -- not a hang -- while the pool is saturated. Then

// release the gate and confirm the two gated requests still complete normally afterwards.

#include <chrono>

#include <condition_variable>

#include <cstring>

#include <iostream>

#include <mutex>

#include <sstream>

#include <stdexcept>

#include <thread>

#include <vector>

#include <netplus/socket.h>

#include "http.h"

#include "httpd.h"

#include "raw_tcp_client.h"

namespace {

std::mutex g_gateMtx;

std::condition_variable g_gateCv;

bool g_release = false;

void waitForGate() {

    std::unique_lock<std::mutex> lk(g_gateMtx);

    g_gateCv.wait(lk, [] { return g_release; });

}

// Offloads every request and blocks in RequestEvent until the test releases g_release --

// mirrors any real slow/stuck backend call from the production incident this guards against.

class GatedH1Server : public libhttppp::HttpEvent {

public:

    GatedH1Server(std::vector<netplus::socket*> serversocket, size_t h1OffloadThreads,

                  size_t h1DispatchQueueMax)

        : HttpEvent(serversocket, 1000, /*h2OffloadThreads=*/0, /*idleTimeoutSeconds=*/0,

                    h1OffloadThreads, /*h2DispatchQueueMax=*/0, h1DispatchQueueMax) {}

    bool shouldOffloadH1Dispatch(libhttppp::HttpRequest &) const override {

        return true;

    }

    void RequestEvent(libhttppp::HttpRequest &req, const int, ULONG_PTR) override {

        waitForGate();

        libhttppp::HttpResponse res;

        res.setContentType("text/plain");

        static const std::string body = "ok";

        res.send(req, body, static_cast<int>(body.size()));

    }

};

struct RunningServer {

    std::unique_ptr<netplus::tcp> sock;

    std::unique_ptr<GatedH1Server> server;

    std::thread thread;

};

RunningServer startServer(int &port, size_t h1OffloadThreads, size_t h1DispatchQueueMax) {

    RunningServer rs;

    rs.sock = std::make_unique<netplus::tcp>("127.0.0.1", 0, 16, -1);

    rs.sock->bind();

    port = rawtcp::boundPort(*rs.sock);

    rs.sock->listen();

    rs.server = std::make_unique<GatedH1Server>(

        std::vector<netplus::socket*>{rs.sock.get()}, h1OffloadThreads, h1DispatchQueueMax);

    netplus::event::Running = true;

    GatedH1Server *serverPtr = rs.server.get();

    rs.thread = std::thread([serverPtr] { serverPtr->runEventloop(); });

    std::this_thread::sleep_for(std::chrono::milliseconds(50));

    return rs;

}

void stopServer(RunningServer &rs) {

    netplus::event::requestStop();

    if (rs.thread.joinable())

        rs.thread.join();

}

// One fresh connection, one GET /. Returns the HTTP status code text found on the response's

// first line (e.g. "200" or "503"), empty if none arrived within timeoutMs. Doesn't wait for

// or parse the body -- the status line alone is enough to tell 503 apart from 200 here.

std::string sendRequestAndReadStatus(int port, int timeoutMs) {

    netplus::tcp sock;

    rawtcp::connectLocal(sock, port);

    std::ostringstream req;

    req << "GET / HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n";

    rawtcp::sendAllOrThrow(sock, req.str());

    sock.setTimeout(timeoutMs);

    std::string buf;

    auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeoutMs);

    while (std::chrono::steady_clock::now() < deadline) {

        size_t lineEnd = buf.find("\r\n");

        if (lineEnd != std::string::npos) {

            // Status line looks like "HTTP/1.1 200 OK" -- the code is the second

            // whitespace-delimited token.

            size_t firstSpace = buf.find(' ');

            if (firstSpace != std::string::npos && firstSpace < lineEnd) {

                size_t codeStart = firstSpace + 1;

                size_t codeEnd = buf.find(' ', codeStart);

                if (codeEnd == std::string::npos || codeEnd > lineEnd) codeEnd = lineEnd;

                sock.close();

                return buf.substr(codeStart, codeEnd - codeStart);

            }

        }

        netplus::buffer rbuf(4096);

        size_t n;

        try {

            n = sock.recvData(rbuf);

        } catch (netplus::NetException &e) {

            if (e.getErrorType() == netplus::NetException::Note) continue; // would-block/timeout

            break; // EOF or hard error

        }

        if (n == 0) break;

        buf.append(rbuf.data.buf, n);

    }

    sock.close();

    return "";

}

} // namespace

int main() {

    int port = 0;

    // 1 worker, queue capacity 1: request A occupies the worker, request B fills the one

    // queue slot, request C must be rejected (503) rather than queued or blocked.

    RunningServer rs = startServer(port, /*h1OffloadThreads=*/1, /*h1DispatchQueueMax=*/1);

    bool ok = true;

    try {

        std::string statusA, statusB;

        std::thread threadA([&] { statusA = sendRequestAndReadStatus(port, 5000); });

        std::this_thread::sleep_for(std::chrono::milliseconds(150)); // let the worker dequeue A

        std::thread threadB([&] { statusB = sendRequestAndReadStatus(port, 5000); });

        std::this_thread::sleep_for(std::chrono::milliseconds(150)); // let B land in the queue

        auto start = std::chrono::steady_clock::now();

        std::string statusC = sendRequestAndReadStatus(port, 2000);

        auto elapsedMs = std::chrono::duration_cast<std::chrono::milliseconds>(

                             std::chrono::steady_clock::now() - start)

                             .count();

        if (statusC != "503") {

            std::cerr << "FAIL: expected 503 for the over-capacity request, got '"

                      << statusC << "'\n";

            ok = false;

        }

        if (elapsedMs > 1000) {

            std::cerr << "FAIL: over-capacity request took " << elapsedMs

                      << "ms -- should fail fast, not wait behind the saturated pool\n";

            ok = false;

        }

        {

            std::lock_guard<std::mutex> lk(g_gateMtx);

            g_release = true;

        }

        g_gateCv.notify_all();

        threadA.join();

        threadB.join();

        if (statusA != "200") {

            std::cerr << "FAIL: gated request A expected 200 after release, got '"

                      << statusA << "'\n";

            ok = false;

        }

        if (statusB != "200") {

            std::cerr << "FAIL: queued request B expected 200 after release, got '"

                      << statusB << "'\n";

            ok = false;

        }

    } catch (const std::exception &e) {

        std::cerr << "FAIL: exception: " << e.what() << "\n";

        ok = false;

    }

    stopServer(rs);

    if (!ok) return 1;

    std::cout << "httpd_h1_dispatch_overload_test: all checks passed\n";

    return 0;

}