test

add_test(httpd_h2_window_credit_race_test httpd_h2_window_credit_race_test)

add_executable(httpd_h2_dispatch_overload_test

    httpd_h2_dispatch_overload_test.cpp

)

if(WIN32)

    target_link_libraries (httpd_h2_dispatch_overload_test httppp-static ws2_32)

else()

    target_link_libraries (httpd_h2_dispatch_overload_test httppp-static)

endif()

add_test(httpd_h2_dispatch_overload_test httpd_h2_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.

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

// Regression test for a production proxyplus incident: _h2DispatchPool used to have an

// unbounded queue, so once every offload worker got stuck (or was just legitimately busy for a

// long time), every subsequent H2 request -- on every connection, including brand new ones --

// queued forever with no response and no way to notice. See HttpEvent's h2DispatchQueueMax ctor

// param and _dispatchH2Stream's submit()-rejection branch (src/httpd.cpp).

//

// This test pins the single offload worker on a gate so it never completes on its own, fills

// the (size-1) queue behind it with one more request, then proves a *third*, brand-new

// connection gets an immediate 503 -- not a hang -- while the pool is saturated. It then

// releases the gate and confirms the two gated requests still complete normally afterwards,

// proving rejection is a transient "pool is full right now" signal, not a permanent trip.

#include <chrono>

#include <condition_variable>

#include <cstring>

#include <iostream>

#include <mutex>

#include <stdexcept>

#include <thread>

#include <vector>

#include <netplus/socket.h>

#include "hpack.h"

#include "http.h"

#include "httpd.h"

#include "raw_tcp_client.h"

namespace {

constexpr uint8_t FRAME_HEADERS = 0x01;

constexpr uint8_t FRAME_SETTINGS = 0x04;

constexpr uint8_t FLAG_END_STREAM = 0x01;

constexpr uint8_t FLAG_END_HEADERS = 0x04;

constexpr size_t FRAME_HEADER_LEN = 9;

constexpr char CLIENT_PREFACE[] = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";

void appendFrameHeader(std::vector<char> &out, uint32_t len, uint8_t type,

                        uint8_t flags, uint32_t streamId) {

    out.push_back(static_cast<char>((len >> 16) & 0xFF));

    out.push_back(static_cast<char>((len >> 8) & 0xFF));

    out.push_back(static_cast<char>(len & 0xFF));

    out.push_back(static_cast<char>(type));

    out.push_back(static_cast<char>(flags));

    out.push_back(static_cast<char>((streamId >> 24) & 0x7F));

    out.push_back(static_cast<char>((streamId >> 16) & 0xFF));

    out.push_back(static_cast<char>((streamId >> 8) & 0xFF));

    out.push_back(static_cast<char>(streamId & 0xFF));

}

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 proxyplus incident this guards against.

class GatedServer : public libhttppp::HttpEvent {

public:

    GatedServer(std::vector<netplus::socket*> serversocket, size_t h2OffloadThreads,

                size_t h2DispatchQueueMax)

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

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

    bool shouldOffloadH2Dispatch(libhttppp::HttpRequest&, uint32_t) 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<GatedServer> server;

    std::thread thread;

};

RunningServer startServer(int &port, size_t h2OffloadThreads, size_t h2DispatchQueueMax) {

    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<GatedServer>(

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

    netplus::event::Running = true;

    GatedServer *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 / request on stream 1. Returns the decoded :status value

// found in the response HEADERS frame (empty if none arrived within timeoutMs).

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

    netplus::tcp sock;

    rawtcp::connectLocal(sock, port);

    std::string hpack_out = libhttppp::hpack::Encoder::encodeRequestHeaders(

        "GET", "/", "http", "127.0.0.1");

    std::vector<char> req(CLIENT_PREFACE, CLIENT_PREFACE + sizeof(CLIENT_PREFACE) - 1);

    appendFrameHeader(req, 0, FRAME_SETTINGS, 0, 0);

    appendFrameHeader(req, static_cast<uint32_t>(hpack_out.size()), FRAME_HEADERS,

                       FLAG_END_HEADERS | FLAG_END_STREAM, 1);

    req.insert(req.end(), hpack_out.begin(), hpack_out.end());

    rawtcp::sendAllOrThrow(sock, req.data(), req.size());

    sock.setTimeout(timeoutMs);

    libhttppp::hpack::Decoder decoder;

    std::vector<char> buf;

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

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

        netplus::buffer rbuf(8192);

        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

        }

        buf.insert(buf.end(), rbuf.data.buf, rbuf.data.buf + n);

        size_t off = 0;

        while (off + FRAME_HEADER_LEN <= buf.size()) {

            uint32_t len = (static_cast<uint8_t>(buf[off]) << 16) |

                           (static_cast<uint8_t>(buf[off + 1]) << 8) |

                           static_cast<uint8_t>(buf[off + 2]);

            uint8_t type = static_cast<uint8_t>(buf[off + 3]);

            uint32_t sid = ((static_cast<uint8_t>(buf[off + 5]) & 0x7F) << 24) |

                           (static_cast<uint8_t>(buf[off + 6]) << 16) |

                           (static_cast<uint8_t>(buf[off + 7]) << 8) |

                           static_cast<uint8_t>(buf[off + 8]);

            if (off + FRAME_HEADER_LEN + len > buf.size())

                break;

            if (type == FRAME_HEADERS && sid == 1) {

                auto fields = decoder.decode(

                    reinterpret_cast<const uint8_t *>(buf.data() + off + FRAME_HEADER_LEN), len);

                for (auto &f : fields) {

                    if (f.name == ":status") {

                        sock.close();

                        return f.value;

                    }

                }

            }

            off += FRAME_HEADER_LEN + len;

        }

        buf.erase(buf.begin(), buf.begin() + off);

    }

    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, /*h2OffloadThreads=*/1, /*h2DispatchQueueMax=*/1);

    bool ok = true;

    try {

        // Request A: dedicated thread since it blocks on the gate until the end of the test.

        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_h2_dispatch_overload_test: all checks passed\n";

    return 0;

}