test

		friend class EventWorker;

		friend class EventWorkerArgs;

		friend struct QuicRfc9000TestAccess;

		friend struct QuicParallelDecryptTestAccess;

	};

	bool isIPAddr(const std::string &host);

endif()

add_test(NAME quic_bbrlite_smoke_test COMMAND quic_bbrlite_smoke_test)

add_executable(quic_parallel_decrypt_test quic_parallel_decrypt_test.cpp)

if(WIN32)

    target_link_libraries(quic_parallel_decrypt_test netplus-static ws2_32)

else()

    target_link_libraries(quic_parallel_decrypt_test netplus-static)

endif()

add_test(NAME quic_parallel_decrypt_test COMMAND quic_parallel_decrypt_test)

add_executable(quic_incremental_dispatch_test quic_incremental_dispatch_test.cpp)

if(WIN32)

    target_link_libraries(quic_incremental_dispatch_test netplus-static ws2_32)

// Correctness tests for Tier 1's opt-in parallel receive-side decrypt

// fan-out (quic::processApplicationPacketsBatchParallel(), src/quic.cpp).

//

// Two distinct things are tested, deliberately kept separate:

//

//   Test A exercises the REAL production code path end to end: a genuine

//   handshake, genuine AEAD key material, a genuine multi-packet recvmmsg

//   batch (forced deterministically by manually driving quic::accept()

//   from a single thread instead of racing a live event loop — see

//   waitForHandshake()'s comment), fed through

//   processApplicationPacketsBatchParallel() via white-box friend access,

//   and verified by checking the reassembled stream content. This catches

//   crashes, decrypt failures, and gross correctness bugs, but — because

//   QUIC stream reassembly is offset-based and already tolerant of

//   reordering by design — it is a weak witness for the one specific risk

//   this feature was scrutinized for: does applying decrypted packets in

//   original arrival order actually survive out-of-order *thread*

//   completion?

//

//   Test B targets exactly that risk directly, without needing real QUIC

//   state at all: it exercises the same ThreadPool + CountdownLatch +

//   indexed-results-array pattern processApplicationPacketsBatchParallel()

//   uses, with synthetic dummy work whose completion order is deliberately

//   reversed (task i sleeps (N-i) milliseconds), and asserts that applying

//   results by original index — not completion order — reconstructs the

//   correct order every time. This is the same failure class ("packet

//   reordering") that a prior report (F30) was rejected over on the send

//   path; this test exists so the receive-side fan-out doesn't quietly

//   regress into it.

#include <iostream>

#include <string>

#include <vector>

#include <thread>

#include <atomic>

#include <chrono>

#include <numeric>

#include <cstring>

#include <shared_mutex>

#include "connection.h"

#include "eventapi.h"

#include "socket.h"

#include "exception.h"

#include "threadpool.h"

#include "https_certs.h"

#include "https_ca_cert.h"

using namespace netplus;

static int g_passed = 0, g_failed = 0;

static void check(bool ok, const char* name) {

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

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

}

// White-box access, mirroring QuicRfc9000TestAccess (quic_rfc9000_test.cpp)

// — there is intentionally no production API for driving accept() manually

// or reaching into _streams/processApplicationPacketsBatchParallel().

namespace netplus {

struct QuicParallelDecryptTestAccess {

    static void processApplicationPacketsBatchParallel(

        quic& endpoint, const std::vector<std::pair<const uint8_t*, size_t>>& packets) {

        endpoint.processApplicationPacketsBatchParallel(packets);

    }

    static std::vector<uint8_t> streamRecvBuffer(const quic& endpoint, uint64_t stream_id) {

        auto it = endpoint._streams.find(stream_id);

        if (it == endpoint._streams.end()) return {};

        return it->second.recv_buffer;

    }

    static bool getHandshakeComplete(const quic& endpoint) {

        return endpoint._handshake_complete;

    }

    static std::shared_ptr<quic> firstChild(quic& listener) {

        std::shared_lock<netplus::WriterPreferringSharedMutex> lock(listener._registry_mutex);

        if (listener._child_connections.empty()) return nullptr;

        return listener._child_connections.begin()->second;

    }

};

} // namespace netplus

// ============================================================================

// Test A: real handshake, deterministically forced multi-packet batch,

// fed through the real processApplicationPacketsBatchParallel() path.

// ============================================================================

// Manually drives the server-side handshake from a single thread instead of

// a live event loop, so nothing else ever touches the server socket

// concurrently with this test — the only way to force a genuine multi-

// datagram recvmmsg batch deterministically rather than racing a live,

// already-fast automatic drain loop (which earlier profiling on this same

// codebase showed keeps up with arrivals closely enough that batches are

// essentially always size 1 under real network timing — see project

// memory). Returns the server-side child connection once the client's

// connect() (run on a separate thread by the caller) has completed.

static std::shared_ptr<quic> waitForHandshakeAndGetChild(

    quic& serverSock, std::atomic<bool>& client_connected, int timeout_ms = 5000)

{

    std::unique_ptr<netplus::socket> new_conn;

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

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

        new_conn.reset();

        try {

            serverSock.accept(new_conn, true);

        } catch (NetException&) {

            // Nothing available this cycle -- keep polling.

        }

        if (client_connected.load(std::memory_order_acquire)) {

            // Drain a couple more cycles so the server side has processed

            // the client's Finished flight too (connect() unblocks on the

            // client as soon as *it* considers the handshake done, which

            // can be a moment before the server-side state fully settles).

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

                std::unique_ptr<netplus::socket> extra;

                try { serverSock.accept(extra, true); } catch (NetException&) {}

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

            }

            break;

        }

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

    }

    return QuicParallelDecryptTestAccess::firstChild(serverSock);

}

static void test_forced_batch_real_decrypt() {

    std::cout << "\n=== Tier 1: forced multi-packet batch through the real decrypt path ===" << std::endl;

    x509cert cert;

    if (!cert.loadFromBuffer(test_cert_der)) {

        check(false, "load certificate");

        return;

    }

    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;

    const int port = 18543;

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

    quic client;

    client.setTrustPolicy(netplus::TlsTrustPolicy{false});

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

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

    std::thread client_thread([&]() {

        try {

            client.connect("127.0.0.1", port);

            client_connected.store(true, std::memory_order_release);

        } catch (std::exception&) {

            client_failed.store(true, std::memory_order_release);

        }

    });

    auto server_child = waitForHandshakeAndGetChild(serverSock, client_connected);

    client_thread.join();

    if (client_failed.load() || !server_child) {

        check(false, "handshake completed (manually-driven server side)");

        return;

    }

    check(QuicParallelDecryptTestAccess::getHandshakeComplete(*server_child),

          "server-side child connection reports handshake complete");

    // Client sends N small stream-data packets back to back. Nothing drains

    // the server socket during this window (the manual accept() loop above

    // has already returned), so they queue up in the kernel's UDP receive

    // buffer instead of being processed one at a time.

    const int N = 16;

    uint64_t sid = client.openStream(true);

    std::vector<std::vector<uint8_t>> sent_chunks;

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

        std::vector<uint8_t> chunk(200, static_cast<uint8_t>(i));

        sent_chunks.push_back(chunk);

        client.sendStreamData(sid, chunk.data(), chunk.size(), i == N - 1);

    }

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

    // One real recvmmsg call (via the public udp::recvBatchAddrViews) —

    // should now return several of the client's queued datagrams in a

    // single syscall, a genuine multi-packet batch.

    std::vector<DatagramView> datagrams;

    std::vector<sockaddr_storage> addrs;

    size_t got = serverSock.recvBatchAddrViews(datagrams, addrs, 64);

    check(got >= 2, "recvBatchAddrViews returned a genuine multi-datagram batch");

    std::vector<std::pair<const uint8_t*, size_t>> packets;

    for (auto& dgram : datagrams) {

        if (dgram.size() < 1) continue;

        if ((dgram[0] & 0x80) == 0) {  // short-header (1-RTT) only, matching accept()'s own routing

            packets.push_back({dgram.data(), dgram.size()});

        }

    }

    check(!packets.empty(), "at least one short-header packet in the forced batch");

    QuicParallelDecryptTestAccess::processApplicationPacketsBatchParallel(*server_child, packets);

    std::vector<uint8_t> expected;

    for (auto& c : sent_chunks) expected.insert(expected.end(), c.begin(), c.end());

    std::vector<uint8_t> got_data = QuicParallelDecryptTestAccess::streamRecvBuffer(*server_child, sid);

    check(got_data.size() == expected.size(),

          "parallel-decrypted batch reassembled to the expected total size");

    check(got_data == expected,

          "parallel-decrypted batch content matches exactly what the client sent, in order");

    client.close();

}

// ============================================================================

// Test B: does index-ordered result application survive out-of-order

// thread completion? Uses the exact same primitives

// (ThreadPool + CountdownLatch + indexed results array) as

// processApplicationPacketsBatchParallel(), with synthetic work whose

// completion order is deliberately reversed, so this is deterministic

// rather than hoping real scheduling happens to reorder something.

// ============================================================================

static void test_indexed_apply_survives_reversed_completion_order() {

    std::cout << "\n=== Tier 1: indexed apply survives reversed completion order ===" << std::endl;

    const int N = 12;

    ThreadPool pool((std::max<unsigned>)(2, std::thread::hardware_concurrency()));

    struct Result { bool done = false; int value = -1; };

    std::vector<Result> results(N);

    CountdownLatch latch(static_cast<size_t>(N));

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

        pool.submit([i, &results, &latch]() {

            // Task 0 sleeps longest, task N-1 sleeps least -- guarantees

            // completion order is the reverse of submission order, exactly

            // the scenario the plan's rejected-F30 discussion worried about

            // ("a race that only shows up under real jitter/reordering").

            std::this_thread::sleep_for(std::chrono::milliseconds((N - i) * 8));

            results[i].value = i * 10;

            results[i].done = true;

            latch.count_down();

        });

    }

    latch.wait();

    bool all_done = true;

    std::vector<int> applied_in_order;

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

        if (!results[i].done) all_done = false;

        applied_in_order.push_back(results[i].value);

    }

    check(all_done, "every submitted task completed before latch.wait() returned");

    std::vector<int> expected(N);

    for (int i = 0; i < N; ++i) expected[i] = i * 10;

    check(applied_in_order == expected,

          "results applied in original submission order regardless of completion order");

}

int main() {

    std::cout << "=== QUIC Tier 1 Parallel Decrypt Correctness Tests ===" << std::endl;

    test_indexed_apply_survives_reversed_completion_order();

    test_forced_batch_real_decrypt();

    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;

}