KPN/tests/test_network.cpp

#include <catch2/catch_test_macros.hpp>
#include <kpn/kpn.hpp>
#include <atomic>
#include <chrono>
#include <mutex>
#include <stdexcept>
#include <string>
#include <thread>

using namespace kpn;

static int increment(int x) { return x + 1; }
static int multiply2(int x) { return x * 2; }

TEST_CASE("network build and run: linear pipeline", "[network]") {
    // Nodes declared first — they own their input channels and must outlive the network
    auto src = make_node<increment>(5);
    auto dst = make_node<multiply2>(5);

    auto& src_in = src.input_channel<0>();
    Channel<int> final_out(5);
    dst.set_output_channel<0>(&final_out);

    Network net;
    net.add("src", src)
       .add("dst", dst)
       .connect("src", src.output<0>(), "dst", dst.input<0>())
       .build();

    net.start();
    src_in.push(5);           // 5 → increment → 6 → multiply2 → 12
    int result = final_out.pop();
    net.stop();

    REQUIRE(result == 12);
}

TEST_CASE("network detects cycle", "[network]") {
    auto a = make_node<increment>(5);
    auto b = make_node<increment>(5);

    Network net;
    net.add("a", a).add("b", b);
    net.connect("a", a.output<0>(), "b", b.input<0>());
    net.connect("b", b.output<0>(), "a", a.input<0>());

    REQUIRE_THROWS_AS(net.build(), NetworkCycleError);
}

TEST_CASE("stop disables input channels — producer push is silently dropped", "[network]") {
    auto node = make_node<increment>(5);
    auto& in_ch = node.input_channel<0>();

    node.start();
    node.stop();

    // After stop, channel is disabled — push must not throw
    in_ch.push(99);
    REQUIRE(in_ch.size() == 0);
}

// Regression: Network::set_error_handler must actually deliver the handler.
//
// The handler was stored in a member and never read. A node's exception was
// discarded at the node boundary and the only surviving evidence was a Closed
// event, which reports that a node stopped but not why — the difference between
// a diagnosis and a guess. StaticNetwork has always wired this; Network
// accepted the handler and silently dropped it, which is worse than not
// offering the setter at all.
//
// The type changed with the fix. It was void(name, exception_ptr), which cannot
// express the keep-running decision the node side needs, so it is now
// NodeErrorHandler like StaticNetwork's.
namespace {

static int throwing_stage(int x) {
    if (x == 42) throw std::runtime_error("boom");
    return x;
}

}  // namespace

TEST_CASE("network error handler receives the node's exception", "[network]") {
    auto src  = kpn::make_node<throwing_stage>(kpn::in<"v">{}, kpn::out<"w">{}, 8);
    kpn::Channel<int> out(8);
    src.set_output_channel<0>(&out);

    kpn::Network net;
    net.add("stage", src).build();

    std::atomic<int>  calls{0};
    std::string       seen_name;
    std::string       seen_what;
    std::mutex        mx;

    net.set_error_handler([&](std::string_view name, std::exception_ptr ep) {
        std::lock_guard lk(mx);
        seen_name = std::string(name);
        try { if (ep) std::rethrow_exception(ep); }
        catch (const std::exception& e) { seen_what = e.what(); }
        calls.fetch_add(1, std::memory_order_relaxed);
        return true;                       // handled: keep the node running
    });

    net.set_watchdog_interval(std::chrono::hours(1));   // keep the report quiet
    net.start();
    src.input_channel<0>().push(42);                    // throws
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    src.input_channel<0>().push(7);                     // must still be running
    const int passed = out.pop();
    net.stop();

    CHECK(calls.load(std::memory_order_relaxed) == 1);
    CHECK(seen_name == "stage");
    CHECK(seen_what == "boom");
    CHECK(passed == 7);
}

// Regression: stopping a network must not wait for the watchdog's next tick.
//
// The watchdog looped on std::this_thread::sleep_for(watchdog_interval_), and
// request_stop() cannot wake a sleeping thread — so stop_watchdog()'s join
// blocked until the current sleep expired. Every teardown paid up to a full
// interval, three seconds by default, and a caller who set a long one to keep
// the periodic report quiet got a stop() that looked like a hang. That is how
// this was found: the error-handler case above set an hour.
TEST_CASE("stopping a network does not wait for the watchdog interval", "[network]") {
    auto node = kpn::make_node<increment>(kpn::in<"v">{}, kpn::out<"w">{}, 4);
    kpn::Channel<int> out(4);
    node.set_output_channel<0>(&out);

    kpn::Network net;
    net.add("inc", node).build();
    net.set_watchdog_interval(std::chrono::hours(1));
    net.start();

    // Let the watchdog actually reach its wait. Without this the test races it:
    // stop_watchdog() runs before the thread has entered the loop, the token is
    // already set when it does, and it exits without ever waiting — which passes
    // against the bug as well as the fix.
    std::this_thread::sleep_for(std::chrono::milliseconds(100));

    const auto t0 = std::chrono::steady_clock::now();
    net.stop();
    const auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - t0).count();

    INFO("stop took " << ms << " ms");
    CHECK(ms < 2000);
}