h1 offload test

add_test(httpd_h2_concurrency_test httpd_h2_concurrency_test)

add_executable(httpd_h1_offload_test

    httpd_h1_offload_test.cpp

)

if(WIN32)

    target_link_libraries (httpd_h1_offload_test httppp-static ws2_32)

else()

    target_link_libraries (httpd_h1_offload_test httppp-static)

endif()

add_test(httpd_h1_offload_test httpd_h1_offload_test)

add_executable(httpd_h2_window_credit_race_test

    httpd_h2_window_credit_race_test.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 HTTP/1.1 request offload (_dispatchH1Request /

// shouldOffloadH1Dispatch / netplus::detachConnection+reattachConnection).

// Unlike H2 (where multiple streams multiplex on one connection, so a

// throwaway per-stream tempreq can be handed to a background pool with

// nothing else to prove), H1's cureq *is* the connection -- offloading it

// needs two separate things demonstrated:

//

//   1. Correctness across the detach/reattach round trip: a SECOND request

//      on the SAME persistent connection after an offloaded one must still

//      work, proving reattachConnection() actually restores normal dispatch

//      rather than just letting the first response out the door.

//   2. The actual point of the feature: enough concurrent slow requests to

//      saturate every epoll/kqueue worker thread must not delay an unrelated

//      fast request when offload is enabled (they run on the separate H1

//      pool instead), but MUST delay it when disabled -- proving both that

//      the synchronous fallback path is unchanged and that offload genuinely

//      frees the workers rather than just moving the delay elsewhere.

#include <algorithm>

#include <chrono>

#include <cstring>

#include <iostream>

#include <sstream>

#include <thread>

#include <vector>

#include <netplus/socket.h>

#include "http.h"

#include "httpd.h"

#include "raw_tcp_client.h"

namespace {

// Synthetic slow backend stand-in, same shape as httpd_h2_concurrency_test's

// SlowServer: "/slow/<ms>" sleeps that many milliseconds before responding

// (standing in for a real blocking backend round-trip); any other path

// responds immediately.

class H1SlowServer : public libhttppp::HttpEvent {

public:

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

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

                    h1OffloadThreads) {}

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

        // Gated only by whether the pool exists (h1OffloadThreads), so one

        // class serves both the synchronous and offloaded test runs.

        return true;

    }

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

        const std::string &path = req.getRequestURL();

        int delayMs = 0;

        auto pos = path.rfind('/');

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

            try { delayMs = std::stoi(path.substr(pos + 1)); } catch (...) {}

        }

        if (delayMs > 0)

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

        libhttppp::HttpResponse res;

        res.setContentType("text/plain");

        res.send(req, std::string("ok"), 2);

    }

};

// Same teardown-ordering rationale as httpd_h2_concurrency_test's

// RunningServer: netplus::event's connection registry and Running/Restart

// flags are process-wide statics, so a server's background threads must be

// fully stopped and joined before the next one starts.

struct RunningServer {

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

    std::unique_ptr<H1SlowServer> server;

    std::thread thread;

};

RunningServer startServer(int &port, size_t h1OffloadThreads) {

    RunningServer rs;

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

    rs.sock->bind();

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

    rs.sock->listen();

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

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

    netplus::event::Running = true;

    H1SlowServer *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();

}

// Sends one GET for `path` on an already-connected keep-alive socket and

// reads until the full response (headers + Content-Length body) has

// arrived. Returns the body.

std::string doKeepAliveGet(netplus::tcp &sock, const std::string &path) {

    std::ostringstream req;

    req << "GET " << path << " HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: keep-alive\r\n\r\n";

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

    std::string buf;

    auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);

    for (;;) {

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

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

            size_t clPos = buf.find("Content-Length:");

            if (clPos == std::string::npos || clPos > headerEnd)

                throw std::runtime_error("doKeepAliveGet: no Content-Length in response");

            size_t clStart = clPos + std::strlen("Content-Length:");

            size_t clEnd = buf.find("\r\n", clStart);

            size_t contentLength = std::stoul(buf.substr(clStart, clEnd - clStart));

            size_t bodyStart = headerEnd + 4;

            if (buf.size() >= bodyStart + contentLength)

                return buf.substr(bodyStart, contentLength);

        }

        if (std::chrono::steady_clock::now() > deadline)

            throw std::runtime_error("doKeepAliveGet: timed out waiting for response");

        netplus::buffer rbuf(4096);

        size_t n = sock.recvData(rbuf);

        if (n == 0) throw std::runtime_error("doKeepAliveGet: connection closed early");

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

    }

}

} // namespace

int main() {

    int failures = 0;

    // Test 1: correctness + keep-alive survives an offloaded request -- two

    // sequential requests on the SAME connection, both offloaded. Proves

    // reattachConnection() actually restores normal dispatch rather than

    // leaving the connection unusable after the first request.

    {

        int port = 0;

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

        netplus::tcp sock;

        rawtcp::connectLocal(sock, port);

        try {

            std::string body1 = doKeepAliveGet(sock, "/fast");

            std::string body2 = doKeepAliveGet(sock, "/fast");

            if (body1 != "ok" || body2 != "ok") {

                std::cerr << "FAIL: keep-alive offloaded requests returned wrong body ('"

                          << body1 << "', '" << body2 << "')" << std::endl;

                ++failures;

            } else {

                std::cout << "[keep-alive] two sequential offloaded requests on one "

                             "connection both succeeded" << std::endl;

            }

        } catch (std::exception &e) {

            std::cerr << "FAIL: keep-alive test threw: " << e.what() << std::endl;

            ++failures;

        }

        sock.close();

        stopServer(rs);

    }

    // Test 2: enough concurrent slow requests to saturate every epoll/kqueue

    // worker thread must not delay an unrelated fast request when H1 offload

    // is enabled, but must delay it when disabled (default/off).

    const unsigned workerCount = (std::max)(1u, std::thread::hardware_concurrency());

    const int slowDelayMs = 300;

    auto runSaturationCase = [&](size_t h1OffloadThreads) -> long long {

        int port = 0;

        RunningServer rs = startServer(port, h1OffloadThreads);

        std::vector<std::unique_ptr<netplus::tcp>> slowSocks(workerCount);

        for (unsigned i = 0; i < workerCount; ++i) {

            slowSocks[i] = std::make_unique<netplus::tcp>();

            rawtcp::connectLocal(*slowSocks[i], port);

            std::ostringstream req;

            req << "GET /slow/" << slowDelayMs << " HTTP/1.1\r\nHost: 127.0.0.1\r\n"

                   "Connection: close\r\n\r\n";

            rawtcp::sendAllOrThrow(*slowSocks[i], req.str());

        }

        // Let every slow request actually land in RequestEvent (accepted +

        // dispatched -- both cheap/near-instant compared to slowDelayMs)

        // before racing the fast one against them.

        std::this_thread::sleep_for(std::chrono::milliseconds(slowDelayMs / 3));

        netplus::tcp fastSock;

        rawtcp::connectLocal(fastSock, port);

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

        std::string body = doKeepAliveGet(fastSock, "/fast");

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

            std::chrono::steady_clock::now() - t0).count();

        if (body != "ok")

            throw std::runtime_error("fast request got wrong body");

        fastSock.close();

        for (auto &s : slowSocks) s->close();

        stopServer(rs);

        return elapsedMs;

    };

    long long syncMs = runSaturationCase(/*h1OffloadThreads=*/0);

    std::cout << "[sync]    fast request latency while " << workerCount

              << " slow requests saturate every worker: " << syncMs << "ms "

                 "(expect >= ~" << slowDelayMs / 2 << "ms, blocked)" << std::endl;

    if (syncMs < slowDelayMs / 2) {

        std::cerr << "FAIL: synchronous path did not show the expected worker-pool "

                     "starvation (latency=" << syncMs << "ms) -- either the fallback "

                     "path regressed, or this environment has more schedulable epoll/"

                     "kqueue workers than hardware_concurrency() reported" << std::endl;

        ++failures;

    }

    long long offloadMs = runSaturationCase(/*h1OffloadThreads=*/workerCount + 4);

    std::cout << "[offload] fast request latency while " << workerCount

              << " slow requests saturate every worker: " << offloadMs << "ms "

                 "(expect < ~" << slowDelayMs / 2 << "ms, unaffected)" << std::endl;

    if (offloadMs >= slowDelayMs / 2) {

        std::cerr << "FAIL: offloaded slow requests still starved the fast one "

                     "(latency=" << offloadMs << "ms)" << std::endl;

        ++failures;

    }

    std::cout << (failures == 0 ? "PASS" : "FAIL") << std::endl;

    return failures == 0 ? 0 : 1;

}