Crafter.Network/tests/ShouldSurviveConnectionChurn/main.cpp

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

// Regression test for connection teardown on the client side.
//
// QUICStream::Stop only *initiates* a shutdown; the msquic handle is closed
// later, off the terminal SHUTDOWN_COMPLETE callback on a worker thread. Two
// things used to go wrong with that, and both need a reconnect loop to show
// up, which is what this test is:
//
//  1. ~ClientQUIC closed the connection straight away, so a connection could
//     be closed while streams belonging to it were still open on the msquic
//     side — an msquic bugcheck, which aborts the process.
//
//  2. StreamStart is asynchronous. If the connection was shut down before the
//     queued start ran, msquic failed the start and left the stream neither
//     started nor shut down, so SHUTDOWN_COMPLETE never arrived and the handle
//     was never closed. The registration's rundown at exit() then blocked
//     forever waiting for it.
//
// The test therefore ends by falling off the end of main() rather than with
// std::_Exit — the static MsQuicRuntime destructor's RegistrationClose is
// where a leaked handle shows up, so skipping it would skip half the point.
//
// Two things about the shape matter:
//
//  - Blocked readers. Each inbound stream gets a thread parked inside
//    RecieveSync for the life of the connection (the shape a long-lived
//    streaming client has), so teardown happens with readers still inside the
//    receive path.
//
//  - A concurrent sender. A client-opened bidirectional control stream with a
//    thread sending on it across the teardown, which is the shape a client
//    with periodic acks or keepalives has. It is the send racing the shutdown
//    that reaches (2); a loop without it passes on the broken build.

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

namespace {
    constexpr int cycles = 25;
    constexpr int streamsPerConnection = 3;
}

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

    // On a build where a teardown deadlocks rather than aborts, fail the test
    // instead of hanging the whole suite.
    std::thread watchdog([] {
        std::this_thread::sleep_for(std::chrono::seconds(90));
        std::println("timed out — a connection teardown never completed");
        std::cout.flush();
        std::_Exit(1);
    });
    watchdog.detach();

    for (int i = 0; i < cycles; ++i) {
        const std::uint16_t port = static_cast<std::uint16_t>(19200 + i);
        const std::string alpn = "churn/1";

        QUICServerCredentials serverCreds;
        serverCreds.selfSigned = true;

        std::mutex serverMutex;
        std::vector<QUICStream> serverStreams;
        ClientQUIC* accepted = nullptr;

        auto listener = std::make_unique<ListenerQUIC>(
            port, alpn, serverCreds, [&](ClientQUIC* peer) {
                // Drain whatever the client opens toward us. Without this the
                // client's control-stream sends would block on flow control
                // and the test would wedge on its own account rather than on
                // anything the library did.
                peer->OnStream([](QUICStream inbound) {
                    auto shared = std::make_shared<QUICStream>(std::move(inbound));
                    std::thread([shared] {
                        try {
                            while (true) (void)shared->RecieveSync();
                        } catch (...) {
                            // Connection closed.
                        }
                    }).detach();
                });
                std::lock_guard lock(serverMutex);
                accepted = peer;
                for (int s = 0; s < streamsPerConnection; ++s) {
                    QUICStream stream = peer->OpenStream(/*unidirectional=*/true);
                    const char payload[] = "hello";
                    stream.SendSync(payload, sizeof(payload) - 1, /*finish=*/false);
                    serverStreams.push_back(std::move(stream));
                }
            });
        listener->ListenAsyncAsync();

        QUICClientCredentials clientCreds;
        clientCreds.insecureNoServerValidation = true;

        {
            ClientQUIC client(std::string("localhost"), port, alpn, clientCreds);

            std::mutex clientMutex;
            std::condition_variable clientCv;
            std::vector<std::thread> readers;
            int started = 0;

            client.OnStream([&](QUICStream stream) {
                auto shared = std::make_shared<QUICStream>(std::move(stream));
                std::thread reader([shared] {
                    try {
                        while (true) (void)shared->RecieveSync();
                    } catch (...) {
                        // Connection closed.
                    }
                });
                std::lock_guard lock(clientMutex);
                readers.push_back(std::move(reader));
                ++started;
                clientCv.notify_all();
            });

            {
                std::unique_lock lock(clientMutex);
                clientCv.wait_for(lock, std::chrono::seconds(5), [&] {
                    return started == streamsPerConnection;
                });
                if (started != streamsPerConnection) {
                    std::println("cycle {}: only {} of {} streams arrived",
                                 i, started, streamsPerConnection);
                    return 1;
                }
            }

            // The sends keep going while the connection is torn down
            // underneath them.
            QUICStream control = client.OpenStream(/*unidirectional=*/false);
            std::atomic<bool> sending{ true };
            std::thread sender([&] {
                while (sending) {
                    try {
                        const char tick[] = "ack";
                        control.SendSync(tick, sizeof(tick) - 1, /*finish=*/false);
                    } catch (...) {
                        return;   // connection went away mid-send
                    }
                    std::this_thread::sleep_for(std::chrono::milliseconds(2));
                }
            });

            client.Stop();
            sending = false;
            if (sender.joinable()) sender.join();
            for (std::thread& reader : readers) {
                if (reader.joinable()) reader.join();
            }
            // `control` and then `client` are destroyed here, once the sender
            // and readers have unwound.
        }

        {
            std::lock_guard lock(serverMutex);
            // Deliberately the other order from the client side: shut the
            // connection down while the app still holds its stream wrappers,
            // so teardown has to cope with streams that are open as far as
            // msquic is concerned.
            serverStreams.clear();
            if (accepted) accepted->Stop();
        }
        listener->Stop();
        listener.reset();
    }

    std::println("survived {} connection teardowns", cycles);
    std::cout.flush();
    // Deliberately no std::_Exit: returning from main runs the static
    // MsQuicRuntime destructor, whose RegistrationClose blocks until every
    // handle opened under the registration has been closed. That wait is the
    // assertion for (2) above.
    return 0;
}