test

endif()

add_test(NAME quic_concurrent_test COMMAND quic_concurrent_test)

add_executable(quic_accept_contention_test quic_accept_contention_test.cpp)

if(WIN32)

    target_link_libraries(quic_accept_contention_test netplus-static ws2_32)

else()

    target_link_libraries(quic_accept_contention_test netplus-static)

endif()

add_test(NAME quic_accept_contention_test COMMAND quic_accept_contention_test)

add_executable(quic_retry_test quic_retry_test.cpp)

if(WIN32)

    target_link_libraries(quic_retry_test netplus-static ws2_32)

// quic_accept_contention_test.cpp

//

// Regression test for a real production hang: paritypp::client::connect_to_node()

// on one mediadb cluster node repeatedly failed to reach a healthy peer with

// "Failed to connect to node X", every single retry, for minutes on end — even

// though that peer's own accept loop was alive and idle-looking in a snapshot,

// and it was answering *other* peers just fine.

//

// Root cause, found by attaching gdb to the live stuck process: quic::accept()'s

// datagram-drain loop processes each incoming datagram in order for one

// recvBatchAddr() batch. For a packet belonging to an *existing* connection —

// whether handled inline (a mid-handshake long-header retransmission,

// quic.cpp's `else` branch around the old line ~2178) or deferred into

// pending_app_pkts and flushed before returning a new connection (flush_pending_app,

// ~line 1988) — it took that connection's own quic_mtx() with a *blocking*

// unique_lock. Every one of these paths runs inline in the single thread that

// drains a given listener (see accept()'s own concurrency note), so if that

// connection's mutex happens to be held elsewhere for a while (e.g. a large

// concurrent transfer's sendStreamData()/pumpIncomingLocked() call, itself

// legitimate work, not a bug), the *entire* drain call — and with it, every

// other connection's traffic in the same batch, and any brand-new incoming

// connection sitting later in that batch — stalls until it's released. A

// connection simply being busy could starve every new connection attempt on

// the same listener indefinitely.

//

// Fixed by using try_lock at both sites: if a connection's mutex isn't free

// right now, its packet for this cycle is dropped rather than waited for —

// QUIC's own loss detection retransmits it, which is far cheaper than

// blocking the shared accept path.

//

// This test reproduces the shape of the failure directly (not just via

// existing coverage, which never puts one connection under heavy sustained

// load *while* a second one is trying to connect for the first time): open

// one connection and keep it busy with a large multi-megabyte transfer, then

// — while that transfer is still in flight — open several brand-new

// connections to the same server and verify each completes its handshake

// and a small round-trip promptly. Before the fix, these new connections

// could be starved behind the busy one for however long its mutex stayed

// held; after the fix, they should stay responsive regardless.

#include <atomic>

#include <chrono>

#include <cstring>

#include <iostream>

#include <map>

#include <string>

#include <thread>

#include <vector>

#include "connection.h"

#include "crypto/sha.h"

#include "eventapi.h"

#include "exception.h"

#include "random.h"

#include "socket.h"

#include "https_ca_cert.h"

#include "https_certs.h"

using namespace netplus;

static int g_passed = 0, g_failed = 0;

static void check(bool ok, const std::string& name) {

    if (ok) { std::cout << "  PASS: " << name << std::endl; g_passed++; }

    else    { std::cout << "  FAIL: " << name << std::endl; g_failed++; }

}

static std::atomic<bool> g_quic_ready(false);

class EchoServer : public event {

public:

    EchoServer(std::vector<netplus::socket*> socks, int timeout = 500)

        : event(socks, timeout) {}

    void RequestEvent(con& curcon, const int, ULONG_PTR) override {

        if (!curcon.RecvData.empty()) {

            curcon.SendData.append(curcon.RecvData.data(), curcon.RecvData.size());

            curcon.RecvData.clear();

        }

    }

    void ResponseEvent(con&, const int, ULONG_PTR) override {}

    void ConnectEvent(con&, const int, ULONG_PTR) override {}

    void DisconnectEvent(con&, const int, ULONG_PTR) override {}

    void CreateConnection(std::shared_ptr<con>& res) override {

        res = std::make_shared<con>(this);

    }

};

static void runQuicServer(std::map<std::string, ssl::CertificateBundle>& certs, int port) {

    try {

        quic serverSock(certs, "127.0.0.1", port, 256, -1);

        serverSock.setStreamCallback([](netplus::socket* sock, uint64_t stream_id,

                                        const std::vector<uint8_t>& data, bool fin) {

            netplus::quic* q = dynamic_cast<netplus::quic*>(sock);

            if (!q || data.empty()) return;

            q->sendStreamData(stream_id, data, fin);

        });

        EchoServer srv({&serverSock});

        g_quic_ready.store(true);

        srv.runEventloop();

    } catch (NetException& e) {

        std::cerr << "[QUIC server] NetException: " << e.what() << std::endl;

        g_quic_ready.store(true);

    } catch (std::exception& e) {

        std::cerr << "[QUIC server] Exception: " << e.what() << std::endl;

        g_quic_ready.store(true);

    }

}

// Keeps one connection continuously busy with a large (several-MB) transfer

// for the whole test — real work that legitimately holds the server-side

// child connection's quic_mtx() for stretches, exactly the condition that

// used to starve new connections.

static void busyClientWorker(const std::string& host, int port,

                              std::atomic<bool>& stop, std::atomic<bool>& started) {

    try {

        quic client;

        client.connect(host, port);

        const size_t payload_size = 8 * 1024 * 1024;

        std::vector<uint8_t> payload(payload_size, 0x42);

        while (!stop.load()) {

            uint64_t sid = client.openStream(true);

            client.sendStreamData(sid, payload, true);

            started.store(true);

            size_t total = 0;

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

            while (total < payload.size() && std::chrono::steady_clock::now() < deadline

                   && !stop.load()) {

                client.pumpNetwork(MSG_DONTWAIT);

                if (client.hasStreamData(sid)) {

                    uint8_t buf[65536];

                    size_t got = client.recvStreamData(sid, buf, sizeof(buf));

                    if (got > 0) total += got;

                }

                std::this_thread::sleep_for(std::chrono::microseconds(100));

            }

        }

        client.close();

    } catch (NetException& e) {

        std::cerr << "[busy client] NetException: " << e.what() << std::endl;

    } catch (std::exception& e) {

        std::cerr << "[busy client] Exception: " << e.what() << std::endl;

    }

}

// A fresh connection attempt made *while* the busy client's transfer is in

// flight: connect, send one small tagged payload, verify the echo, and

// report how long the whole thing took.

static bool newConnectionRoundTrip(int idx, const std::string& host, int port,

                                    double& elapsed_ms_out) {

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

    try {

        quic client;

        client.connect(host, port);

        std::string tag = "NEWCONN-" + std::to_string(idx);

        std::vector<uint8_t> payload(tag.begin(), tag.end());

        uint64_t sid = client.openStream(true);

        client.sendStreamData(sid, payload, true);

        std::vector<uint8_t> out;

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

        while (out.size() < payload.size() && std::chrono::steady_clock::now() < deadline) {

            client.pumpNetwork(MSG_DONTWAIT);

            if (client.hasStreamData(sid)) {

                uint8_t buf[4096];

                size_t got = client.recvStreamData(sid, buf, sizeof(buf));

                if (got > 0) out.insert(out.end(), buf, buf + got);

            }

            std::this_thread::sleep_for(std::chrono::microseconds(200));

        }

        client.close();

        elapsed_ms_out = std::chrono::duration<double, std::milli>(

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

        return out.size() == payload.size() &&

               std::memcmp(out.data(), payload.data(), payload.size()) == 0;

    } catch (NetException& e) {

        std::cerr << "[new conn " << idx << "] NetException: " << e.what() << std::endl;

    } catch (std::exception& e) {

        std::cerr << "[new conn " << idx << "] Exception: " << e.what() << std::endl;

    }

    elapsed_ms_out = std::chrono::duration<double, std::milli>(

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

    return false;

}

int main() {

    std::cout << "=== QUIC Accept-Loop Contention Regression Test ===" << std::endl;

    const int kQuicPort = 19613;

    // Generous but bounded: a healthy accept loop should register a new

    // connection in well under a second even with a large transfer

    // continuously busy on the same listener. This is orders of magnitude

    // below what the original bug produced (the real incident's retries

    // failed after a full 5-second per-attempt deadline, repeatedly, for

    // minutes) — comfortably distinguishes "responsive" from "starved"

    // without being a flaky tight bound on a shared CI box.

    const double kMaxAcceptableMs = 2000.0;

    x509cert cert;

    if (!cert.loadFromBuffer(test_cert_der)) {

        std::cerr << "Failed to load certificate" << std::endl;

        return 1;

    }

    std::map<std::string, ssl::CertificateBundle> certs;

    ssl::CertificateBundle bundle;

    bundle.cert = cert;

    bundle.privateKeyDer = std::vector<uint8_t>(test_key_der.begin(), test_key_der.end());

    bundle.rsa_key = rsa(bundle.privateKeyDer);

    bundle.chain.push_back(std::vector<uint8_t>(MKCERT_ROOT_CA_DER,

        MKCERT_ROOT_CA_DER + MKCERT_ROOT_CA_DER_LEN));

    certs["localhost"] = bundle;

    certs["127.0.0.1"] = bundle;

    std::thread quicServerThread(runQuicServer, std::ref(certs), kQuicPort);

    while (!g_quic_ready.load()) std::this_thread::sleep_for(std::chrono::milliseconds(10));

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

    std::atomic<bool> stop_busy(false);

    std::atomic<bool> busy_started(false);

    std::thread busyThread(busyClientWorker, "127.0.0.1", kQuicPort,

                            std::ref(stop_busy), std::ref(busy_started));

    while (!busy_started.load()) std::this_thread::sleep_for(std::chrono::milliseconds(5));

    // Give the busy transfer a head start so it's genuinely mid-flight.

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

    const int kNumNewConnections = 8;

    for (int i = 0; i < kNumNewConnections; ++i) {

        double elapsed_ms = 0.0;

        bool ok = newConnectionRoundTrip(i, "127.0.0.1", kQuicPort, elapsed_ms);

        std::cout << "  new connection " << i << ": ok=" << ok

                   << " elapsed_ms=" << elapsed_ms << std::endl;

        check(ok && elapsed_ms < kMaxAcceptableMs,

              "new connection " + std::to_string(i) +

              " completed promptly while a large transfer was in flight (" +

              std::to_string(elapsed_ms) + "ms)");

    }

    stop_busy.store(true);

    busyThread.join();

    event::Running = false;

    quicServerThread.join();

    std::cout << "\n==============================" << std::endl;

    std::cout << "Results: " << g_passed << " passed, " << g_failed << " failed" << std::endl;

    std::cout << "==============================" << std::endl;

    return (g_failed > 0) ? 1 : 0;

}