Crafter.Network/tests/ShouldNotDropEarlyStreams/main.cpp

//SPDX-License-Identifier: LGPL-3.0-only
//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®

// Regression test for the dropped-early-stream race in ListenerHTTP/ListenerQUIC.
//
// The server's connection callback (which receives PEER_STREAM_STARTED) is
// installed in the listener's NEW_CONNECTION handler, but the per-connection
// onConnect setup (which installs the OnStream handler + sends the server's
// SETTINGS) is dispatched to the Crafter ThreadPool. An HTTP/3 peer opens its
// control + request streams the instant the QUIC handshake completes — i.e.
// potentially BEFORE onConnect has run.
//
// The bug: the original code installed only a no-op bootstrap connection
// callback in NEW_CONNECTION and deferred the *real* callback to the same
// ThreadPool task as onConnect. Any PEER_STREAM_STARTED that arrived before
// that task ran was delivered to the bootstrap and silently dropped — the
// stream was never adopted, so the request was never answered. With Chromium
// (which opens streams aggressively right after the handshake) this surfaced
// as an intermittent "WebTransport connection rejected" that cleared on retry.
//
// This test forces that window deterministically: it SATURATES the ThreadPool
// before connecting, so onConnect cannot run until we release it. Meanwhile the
// client opens its request stream immediately. The connection callback runs on
// the msquic thread (not the ThreadPool), so with the fix it still buffers the
// early stream; on the buggy build the bootstrap drops it. After we release the
// pool, onConnect drains the buffer and the request is answered — or, on the
// buggy build, the request stream is gone and the read below hangs until the
// watchdog fails the test.

import Crafter.Network;
import Crafter.Thread;
import std;
using namespace Crafter;

namespace {
    // Read one HTTP/3 frame off `stream` → (frameType, payload). Unconsumed
    // tail bytes are PrependReceived'd back. (Same helper shape as the
    // ShouldEchoWebTransport test.)
    std::pair<std::uint64_t, std::vector<char>> ReadFrame(QUICStream& stream) {
        std::vector<char> buf;
        std::uint64_t type = 0; std::size_t cn = 0;
        while (true) {
            const auto* p = reinterpret_cast<const std::uint8_t*>(buf.data());
            if (HTTP3::DecodeVarint(p, buf.size(), type, cn)) break;
            auto chunk = stream.RecieveSync();
            buf.insert(buf.end(), chunk.begin(), chunk.end());
        }
        buf = std::vector<char>(buf.begin() + cn, buf.end());

        std::uint64_t len = 0; std::size_t lc = 0;
        while (true) {
            const auto* p = reinterpret_cast<const std::uint8_t*>(buf.data());
            if (HTTP3::DecodeVarint(p, buf.size(), len, lc)) break;
            auto chunk = stream.RecieveSync();
            buf.insert(buf.end(), chunk.begin(), chunk.end());
        }
        buf = std::vector<char>(buf.begin() + lc, buf.end());

        while (buf.size() < len) {
            auto chunk = stream.RecieveSync();
            buf.insert(buf.end(), chunk.begin(), chunk.end());
        }
        std::vector<char> payload(buf.begin(), buf.begin() + len);
        std::vector<char> tail(buf.begin() + len, buf.end());
        if (!tail.empty()) stream.PrependReceived(std::move(tail));
        return {type, std::move(payload)};
    }
}

int main() {
    ThreadPool::Start();

    constexpr std::uint16_t kPort = 8086;

    // Watchdog: on the buggy build the request stream is dropped and the
    // response read below blocks forever. Fail the test instead of hanging
    // the whole suite.
    std::thread watchdog([] {
        std::this_thread::sleep_for(std::chrono::seconds(8));
        std::println("timed out waiting for response — early request stream was dropped");
        std::cout.flush();
        std::_Exit(1);
    });
    watchdog.detach();

    // ── Server: a single HTTP route. ─────────────────────────────────────
    QUICServerCredentials serverCreds;
    serverCreds.selfSigned = true;
    std::unordered_map<std::string, std::function<HTTPResponse(const HTTPRequest&)>> routes = {
        {"/", [](const HTTPRequest&) { return CreateResponseHTTP("200", "ok"); }},
    };
    ListenerAsyncHTTP listener(kPort, serverCreds, std::move(routes));

    // ── Saturate the ThreadPool so the server's onConnect can't run yet. ──
    // The connection callback (PEER_STREAM_STARTED) runs on the msquic thread
    // and is unaffected; only onConnect (OnStream install + SETTINGS) is gated.
    std::atomic<bool> release{false};
    std::atomic<int>  parked{0};
    for (int i = 0; i < 256; ++i) {
        ThreadPool::Enqueue([&] {
            parked.fetch_add(1);
            while (!release.load()) std::this_thread::sleep_for(std::chrono::milliseconds(1));
        });
    }
    // Give the pool a moment to actually pick up and block on those tasks.
    std::this_thread::sleep_for(std::chrono::milliseconds(100));

    try {
        // ── Client: open the request stream immediately after the handshake,
        // i.e. while onConnect is still gated behind the saturated pool. ────
        QUICClientCredentials clientCreds;
        clientCreds.insecureNoServerValidation = true;
        ClientQUIC quic("localhost", kPort, std::string(HTTP3::kAlpn), clientCreds);

        // Drain peer-initiated (server) unidi streams once they appear.
        quic.OnStream([](QUICStream stream) {
            try { while (true) (void)stream.RecieveSync(); } catch (...) {}
        });

        // Client control stream + SETTINGS.
        QUICStream control = quic.OpenStream(/*unidirectional=*/true);
        auto prelude = HTTP3::BuildControlStreamPrelude();
        control.SendSync(prelude.data(), static_cast<std::uint32_t>(prelude.size()),
                         /*finish=*/false);

        // GET / on a fresh bidi request stream, FIN immediately.
        QUICStream req = quic.OpenStream(/*unidirectional=*/false);
        std::vector<std::pair<std::string, std::string>> fields = {
            {":method",    "GET"},
            {":scheme",    "https"},
            {":authority", "localhost"},
            {":path",      "/"},
        };
        auto headerPayload = HTTP3::EncodeFieldSection(fields);
        std::vector<std::uint8_t> wire;
        HTTP3::WriteFrame(wire, HTTP3::kFrameHeaders, headerPayload.data(), headerPayload.size());
        req.SendSync(wire.data(), static_cast<std::uint32_t>(wire.size()), /*finish=*/true);

        // Let the early streams reach the server and be buffered (fix) or
        // dropped (bug) while onConnect is still gated.
        std::this_thread::sleep_for(std::chrono::milliseconds(300));

        // Now release the pool — onConnect runs, and with the fix it drains the
        // buffered request stream and answers it.
        release.store(true);

        // Read the response HEADERS frame and check the status.
        auto [respType, respPayload] = ReadFrame(req);
        if (respType != HTTP3::kFrameHeaders) {
            std::println("expected HEADERS response frame, got type {}", respType);
            return 1;
        }
        auto respFields = HTTP3::DecodeFieldSection(
            reinterpret_cast<const std::uint8_t*>(respPayload.data()), respPayload.size());
        std::string status;
        for (auto& [k, v] : respFields) if (k == ":status") status = v;
        if (status != "200") {
            std::println("expected status 200, got '{}'", status);
            return 1;
        }
        std::_Exit(0);
    } catch (std::exception& e) {
        release.store(true);
        std::println("client failed: {}", e.what());
        return 1;
    }
}