// Regression: a blocking push must not park a pool worker.
//
// Node outputs use push_blocking so a full channel costs time rather than data
// (a dropped frame does not degrade a downstream result, it silently changes
// one). But push_blocking sleeps *inside* fire_once, which runs on a pool
// worker — and nodes are pinned to workers by index. Park enough workers in
// that retry loop and there is nobody left to run the consumer that would drain
// the channel, so the whole chain wedges.
//
// This is the failure channel.hpp:174 already warns about for sentinels
// ("a blocking push would park that thread and stop it draining its own input,
// cascading into a hold-and-wait deadlock under backpressure"). The warning
// applies to data pushes too.
//
// Observed in the field as an intermittent hang: frame_source, camera_pos,
// face_detector and face_aligner all asleep in push_blocking at once.
#include <catch2/catch_test_macros.hpp>
#include <kpn/kpn.hpp>

#include <atomic>
#include <chrono>
#include <thread>

namespace {

struct Produce {
    static constexpr std::string_view label() { return "produce"; }
    int n{0};
    int operator()() { return n++; }
};

struct Relay {
    static constexpr std::string_view label() { return "relay"; }
    int operator()(int v) { return v; }
};

// Deliberately slower than the producer, so the channels between them fill.
struct SlowSink {
    static constexpr std::string_view label() { return "slow_sink"; }
    std::atomic<int>* seen;
    void operator()(int) {
        std::this_thread::sleep_for(std::chrono::milliseconds(2));
        seen->fetch_add(1, std::memory_order_relaxed);
    }
};

}  // namespace

TEST_CASE("a saturated chain keeps draining", "[backpressure][deadlock]") {
    std::atomic<int> seen{0};

    Produce  p_fn;
    Relay    r1_fn, r2_fn, r3_fn;
    SlowSink s_fn{&seen};

    // Small channels so they saturate immediately, and a chain longer than a
    // modest pool — the shape that starves workers.
    kpn::ObjectNode<Produce,  kpn::in<>,        kpn::out<"a">, "produce",   0> p (p_fn,  2);
    kpn::ObjectNode<Relay,    kpn::in<"a">,     kpn::out<"b">, "relay1",    0> r1(r1_fn, 2);
    kpn::ObjectNode<Relay,    kpn::in<"b">,     kpn::out<"c">, "relay2",    0> r2(r2_fn, 2);
    kpn::ObjectNode<Relay,    kpn::in<"c">,     kpn::out<"d">, "relay3",    0> r3(r3_fn, 2);
    kpn::ObjectNode<SlowSink, kpn::in<"d">,     kpn::out<>,    "slow_sink", 0> s (s_fn,  2);

    auto net = kpn::make_network(
        kpn::edge(p.output<"a">(),  r1.input<"a">()),
        kpn::edge(r1.output<"b">(), r2.input<"b">()),
        kpn::edge(r2.output<"c">(), r3.input<"c">()),
        kpn::edge(r3.output<"d">(), s.input<"d">())
    );
    net.start();

    // The sink is the slowest stage at 2 ms/item, so 40 items is ~80 ms of real
    // work. Anything approaching the timeout means the chain stopped draining
    // rather than merely running slowly.
    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(20);
    while (seen.load(std::memory_order_relaxed) < 40 &&
           std::chrono::steady_clock::now() < deadline)
        std::this_thread::sleep_for(std::chrono::milliseconds(5));

    const int got = seen.load(std::memory_order_relaxed);
    net.stop();

    INFO("items drained: " << got << " of 40");
    CHECK(got >= 40);
}

// Regression: a saturated chain must never stall.
//
// push_outputs parks from two places: the retry at the top of fire_once, and
// the ordinary push after the node function returns. Both release the worker,
// so both face the same lost wakeup — a space callback firing while queued_ is
// still up is swallowed by try_submit's CAS. Only the retry path re-checked for
// space afterwards. The normal path fell through to on_input_ready(), which
// resubmits only if inputs are ready — and the firing that just parked had
// consumed its input, so they are not.
//
// The strand is permanent under saturation: the node holds its value, its
// consumer waits for exactly that value, and its producer fills the node's
// input channel and parks too. Nothing moves again.
//
// The test above cannot catch it — 40 items drain before any strand occurs.
// This one runs the chain saturated and watches for progress to *freeze*, which
// is the signature of the deadlock. It deliberately does not assert a total:
// capacity-1 channels are slow, and "slow" must never be reported as "wedged".
namespace {

struct FreeRun {
    static constexpr std::string_view label() { return "free_run"; }
    int n{0};
    int operator()() { return n++; }
};

struct CountingSink {
    static constexpr std::string_view label() { return "counting_sink"; }
    std::atomic<int>* seen;
    void operator()(int) { seen->fetch_add(1, std::memory_order_relaxed); }
};

}  // namespace

TEST_CASE("a saturated chain never stalls", "[backpressure][deadlock]") {
    std::atomic<int> seen{0};

    FreeRun      p_fn;
    Relay        r1_fn, r2_fn;
    CountingSink s_fn{&seen};

    // Capacity 1 everywhere: every push contends, so the park path is taken
    // constantly and the race window is sampled millions of times.
    kpn::ObjectNode<FreeRun,      kpn::in<>,     kpn::out<"a">, "free_run", 0> p (p_fn,  1);
    kpn::ObjectNode<Relay,        kpn::in<"a">,  kpn::out<"b">, "relay1",   0> r1(r1_fn, 1);
    kpn::ObjectNode<Relay,        kpn::in<"b">,  kpn::out<"c">, "relay2",   0> r2(r2_fn, 1);
    kpn::ObjectNode<CountingSink, kpn::in<"c">,  kpn::out<>,    "sink",     0> s (s_fn,  1);

    auto net = kpn::make_network(
        kpn::edge(p.output<"a">(),  r1.input<"a">()),
        kpn::edge(r1.output<"b">(), r2.input<"b">()),
        kpn::edge(r2.output<"c">(), s.input<"c">())
    );
    net.start();

    // A live chain moves thousands of items a second, so 3 s with no movement
    // at all is a wedge, not a slow patch. Sampling for 25 s gives the race
    // ample opportunity: the pipeline hit it roughly twice in 30 runs.
    const auto  giveup    = std::chrono::steady_clock::now() + std::chrono::seconds(25);
    int         last      = 0;
    auto        last_move = std::chrono::steady_clock::now();
    bool        stalled   = false;
    int         stall_at  = 0;

    while (std::chrono::steady_clock::now() < giveup) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        const int now_seen = seen.load(std::memory_order_relaxed);
        if (now_seen != last) {
            last      = now_seen;
            last_move = std::chrono::steady_clock::now();
        } else if (std::chrono::steady_clock::now() - last_move >
                   std::chrono::seconds(3)) {
            stalled  = true;
            stall_at = now_seen;
            break;
        }
    }

    net.stop();

    INFO("chain stalled after " << stall_at << " items");
    CHECK_FALSE(stalled);
    // Guard against the test passing because nothing ever ran.
    CHECK(last > 1000);
}