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