// 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);
}

// Regression: a node must not start with a wake already outstanding.
//
// 9c5ce5f established the invariant for the running pipeline — a node never
// sleeps with a wake it dropped. start() broke the same invariant before the
// pipeline was even running:
//
//     enable_inputs(...);          // channel goes live here
//     stop_flag_.store(false);
//     queued_.store(false);
//     register_callbacks(...);     // push callback installed here
//
// StaticNetwork starts nodes sources-first, so an upstream node is already
// firing into this one during that gap. A push landing there is accepted by the
// ring but wakes nobody: Channel::push invokes push_callback_ only on the
// empty→non-empty transition, and at that instant the callback is null. Every
// later push sees a non-empty ring and stays silent. The node is never
// submitted, and since a sink has no outputs there is no space callback to
// rescue it either.
//
// The signature is distinctive: **zero** items delivered, not a stall partway.
// The chain reads as wedged from the first frame. Under `ctest -j4` on a loaded
// machine it reproduced 7 times in 24, and never once in 10 unloaded runs —
// contention widens the window between those two statements. That is almost
// certainly the "rare hang, ~1 run in 20 at a 300 s timeout" 28e0667 recorded as
// known-incomplete.
//
// This test needs no contention: it constructs the state the race leaves behind
// directly, by enabling the input and pushing before start() is ever called.
namespace {

int passthrough(int x) { return x; }

}  // namespace

TEST_CASE("a node started with data already queued still fires",
          "[backpressure][startup]") {
    auto pool = std::make_shared<kpn::ThreadPool>(2);
    pool->start();

    auto node = kpn::make_pool_node<passthrough>(pool, 8);
    kpn::Channel<int> out_ch(8);
    node.set_output_channel<0>(&out_ch);

    // The missed edge: the channel is live and already holds a value, but no
    // callback was installed when it arrived, so the wake has been and gone.
    node.input_channel<0>().enable();
    node.input_channel<0>().push(21);

    node.start();

    // Bounded wait — a plain pop() would hang rather than fail on a regression.
    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
    while (out_ch.size() == 0 && std::chrono::steady_clock::now() < deadline)
        std::this_thread::sleep_for(std::chrono::milliseconds(5));

    const bool delivered = out_ch.size() > 0;
    const int  got       = delivered ? out_ch.pop() : -1;

    node.stop();
    pool->stop();

    INFO("value delivered: " << got);
    REQUIRE(delivered);
    CHECK(got == 21);
}

// Regression: a fanout absorbs an unequal pair of consumers by slowing, not by
// dropping.
//
// 6595e6e made node outputs lossless and 28e0667 stopped them parking a worker,
// but FanoutNode was in neither: it kept `catch (ChannelOverflowError&) {}` per
// output, so whichever branch fell behind lost items — silently, and by an
// amount that depended on timing. Two runs of the same input could therefore
// disagree, which is fatal for a fixture the rest of the suite is scored
// against.
//
// The two assertions are the two halves of the requirement:
//   - no gaps: the slow branch receives *every* item, not most of them;
//   - bounded lead: the fast branch is throttled to the slow one rather than
//     racing ahead over a drain that is quietly discarding the difference.
//
// Either alone would pass on a broken implementation. A fanout that pushed only
// to the slow branch has no gaps; one that dropped everything for the slow
// branch keeps a bounded lead by never letting it fall behind.
namespace {

// Records the sequence it sees, so a dropped item shows up as a gap rather than
// merely as a smaller total.
struct SeqCheck {
    std::atomic<int>*  next_expected;
    std::atomic<bool>* saw_gap;
    int                delay_us{0};

    void record(int v) const {
        if (delay_us)
            std::this_thread::sleep_for(std::chrono::microseconds(delay_us));
        const int want = next_expected->load(std::memory_order_relaxed);
        if (v != want) saw_gap->store(true, std::memory_order_relaxed);
        else           next_expected->store(want + 1, std::memory_order_relaxed);
    }
};

struct FastBranch : SeqCheck {
    static constexpr std::string_view label() { return "fast_branch"; }
    void operator()(int v) { record(v); }
};

struct SlowBranch : SeqCheck {
    static constexpr std::string_view label() { return "slow_branch"; }
    void operator()(int v) { record(v); }
};

}  // namespace

TEST_CASE("a fanout absorbs an unequal pair by slowing, not dropping",
          "[backpressure][fanout]") {
    std::atomic<int>  fast_next{0}, slow_next{0};
    std::atomic<bool> fast_gap{false}, slow_gap{false};

    FreeRun    p_fn;
    FastBranch fast_fn{{&fast_next, &fast_gap, 0}};
    SlowBranch slow_fn{{&slow_next, &slow_gap, 500}};   // 0.5 ms/item

    // Small channels so the slow branch saturates in the first few milliseconds
    // and stays saturated for the whole run.
    kpn::ObjectNode<FreeRun,    kpn::in<>,       kpn::out<"v">, "free_run", 0> p (p_fn,    8);
    kpn::ObjectNode<FastBranch, kpn::in<"fast">, kpn::out<>,    "fast",     0> fa(fast_fn, 8);
    kpn::ObjectNode<SlowBranch, kpn::in<"slow">, kpn::out<>,    "slow",     0> sl(slow_fn, 8);

    // Two edges from one output port: make_network auto-inserts FanoutNode<int,2>.
    auto net = kpn::make_network(
        kpn::edge(p.output<"v">(), fa.input<"fast">()),
        kpn::edge(p.output<"v">(), sl.input<"slow">())
    );
    net.start();
    std::this_thread::sleep_for(std::chrono::seconds(1));
    net.stop();

    const int fast_seen = fast_next.load(std::memory_order_relaxed);
    const int slow_seen = slow_next.load(std::memory_order_relaxed);

    INFO("fast branch " << fast_seen << " items, slow branch " << slow_seen);
    CHECK_FALSE(fast_gap.load(std::memory_order_relaxed));
    CHECK_FALSE(slow_gap.load(std::memory_order_relaxed));
    // Guard against passing because nothing ran: 1 s at 0.5 ms/item is ~2000.
    CHECK(slow_seen > 200);
    // The lead is bounded by the buffering between the two — the fanout's own
    // input, the two output channels, and one item in each node's hand. A
    // dropping fanout has no such bound: the fast branch runs at full speed and
    // the difference is the loss.
    CHECK(fast_seen - slow_seen < 200);
}

// Regression: a filter must not drop an EOF sentinel into a full output.
//
// RouterNode and FilterNode were the last nodes on a data path still using the
// throwing push() and swallowing the result:
//
//     try { out_ch_->push(val); } catch (const ChannelOverflowError&) {}
//
// 6595e6e made node outputs lossless, 28e0667 stopped them parking a worker,
// a8cfe73 did the same for FanoutNode. These two were in none of them.
//
// For ordinary values that is the familiar silent-loss problem. For a sentinel
// it is a hang. EOF is what tells every downstream node to shut down, and
// nothing comes after it to retry — so a filter that passes EOF by predicate
// but drops it by backpressure produces a pipeline that never terminates. The
// scene-actor-extraction decimator is exactly this shape: `if (f.eof) return
// true;` in the predicate, feeding a chain whose slowest node is an ONNX
// embedder, so the output is reliably full at the moment EOF arrives.
//
// The test forces that state rather than racing for it: the sink is slow enough
// that the filter's output channel is saturated for the whole run, so EOF meets
// a full ring with certainty.
//
// Both assertions are needed. `saw_eof` alone would pass on an implementation
// that dropped every ordinary value and delivered only the sentinel; `count`
// alone would pass on the broken one, which delivers plenty of values and loses
// only the token that matters.
namespace {

struct EofFrame {
    int  seq{0};
    bool eof{false};
};

// EOF is emitted exactly once, as a real source does. Everything after it is a
// filler frame the predicate rejects, which keeps the node alive without
// re-offering the sentinel — a source that retried EOF would mask the bug,
// since a later attempt could find the channel drained.
struct EofSource {
    static constexpr std::string_view label() { return "eof_source"; }
    int  n{0};
    int  total{0};
    EofFrame operator()() {
        if (n > total) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            return {-1, false};          // filler: dropped by the predicate
        }
        EofFrame f{n, n == total};
        ++n;
        return f;
    }
};

struct EofSink {
    static constexpr std::string_view label() { return "eof_sink"; }
    std::atomic<int>*  count;
    std::atomic<bool>* saw_eof;
    void operator()(EofFrame f) {
        std::this_thread::sleep_for(std::chrono::microseconds(200));
        if (f.eof) saw_eof->store(true, std::memory_order_release);
        else       count->fetch_add(1, std::memory_order_relaxed);
    }
};

}  // namespace

TEST_CASE("a filter delivers EOF into a saturated output", "[backpressure][filter]") {
    std::atomic<int>  count{0};
    std::atomic<bool> saw_eof{false};

    EofSource src_fn{0, 40};
    EofSink   sink_fn{&count, &saw_eof};

    // Every real frame passes the predicate, so the only thing between source
    // and sink is backpressure. Small channels keep the output saturated.
    auto filt = kpn::make_filter<EofFrame>(
        [](const EofFrame& f) { return f.seq >= 0; }, 4);

    kpn::ObjectNode<EofSource, kpn::in<>,      kpn::out<"f">, "eof_source", 0> s(src_fn,  4);
    kpn::ObjectNode<EofSink,   kpn::in<"f">,   kpn::out<>,    "eof_sink",   0> k(sink_fn, 4);

    auto net = kpn::make_network(
        kpn::edge(s.output<"f">(),      filt.input<0>()),
        kpn::edge(filt.output<0>(),     k.input<"f">())
    );
    net.start();

    // Generous relative to 41 frames at 200 us, and this is a liveness test:
    // the broken implementation never sets saw_eof no matter how long it runs.
    for (int i = 0; i < 200 && !saw_eof.load(std::memory_order_acquire); ++i)
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    net.stop();

    INFO("values delivered: " << count.load() << " of 40");
    CHECK(saw_eof.load(std::memory_order_acquire));
    CHECK(count.load(std::memory_order_relaxed) == 40);
}