Error Handling & Events

KPN++ provides three complementary layers for observing and reacting to failures.

---

1. Per-node error handler

Called when a node's function throws an unhandled exception. Return true to skip the failed invocation and keep running; false to stop the node.

    // Return true  → skip this invocation, keep the node running.
    // Return false → stop the node (downstream drains then also stops).
    proc.set_error_handler([](std::string_view name, std::exception_ptr ep) {
        try { std::rethrow_exception(ep); }
        catch (const std::exception& e) {
            std::cerr << "[" << name << "] skipping item — " << e.what() << '\n';
        }
        return true;
    });

When a node stops (either from false return or no handler installed), it:

1. Disables its input channels — upstream stops pushing into dead queues.
2. Disables its output channels — downstream nodes receive ChannelClosedError on their next pop, propagating the shutdown naturally through the graph.

---

2. Per-node overflow callback

Fired with a timestamp each time an output push is dropped because the channel is full. The node name is known at registration so it is not included — keeping the callback zero-overhead when unused.

    // Per-node overflow callback — no node name needed, known at registration.
    std::atomic<int> overflow_count{0};
    src.set_overflow_callback([&](steady_clock::time_point ts) {
        auto ms = duration_cast<milliseconds>(ts.time_since_epoch()).count();
        std::cerr << "[overflow] fast_source at t=" << ms << "ms\n";
        overflow_count.fetch_add(1);
    });

Note

The callback is purely informational — the node always continues after an overflow. To stop the node on overflow, call node.stop() from inside the callback.

A matching set_closed_callback() fires (also with just a timestamp) when the node stops due to a closed upstream channel:

node.set_closed_callback([](std::chrono::steady_clock::time_point ts) {
    std::cerr << "node stopped at t=" << ts.time_since_epoch().count() << '\n';
});

Each node holds two callback slots per event type — one user-set (registered above) and one injected by the network (see below). Both fire independently.

---

3. Network-level event handler

One callback for the whole network. Receives the node name (captured in a closure by the network at build() / start()), a NodeEvent, and a timestamp:

    // Network-level aggregate handler — covers every node, includes node name.
    net.set_event_handler([](std::string_view name, NodeEvent ev,
                             steady_clock::time_point ts) {
        auto ms = duration_cast<milliseconds>(ts.time_since_epoch()).count();
        std::string_view kind = (ev == NodeEvent::Overflow) ? "overflow" : "closed";
        std::cerr << "[net:" << kind << "] node=" << name << " t=" << ms << "ms\n";
    });

NodeEvent values:

Value	Meaning
NodeEvent::Overflow	An output push was dropped (channel full)
NodeEvent::Closed	The node stopped (crash or upstream close cascade)

The network handler and any per-node callbacks are independent — both fire when set.

---

Complete example

examples/16_event_callbacks/main.cpp shows a fast producer overflowing a slow consumer, with both a per-node overflow callback and a network-level event handler active simultaneously.

Node functions:

static std::atomic<int> g_seq{0};

static int fast_source() {
    std::this_thread::sleep_for(milliseconds(2));   // ~500/s
    return g_seq.fetch_add(1);
}

static void slow_sink(int x) {
    std::this_thread::sleep_for(milliseconds(50));  // ~20/s
    std::cout << "  consumed: " << x << '\n';
}

Per-node overflow callback:

    // Per-node overflow callback — no node name needed, known at registration.
    std::atomic<int> overflow_count{0};
    src.set_overflow_callback([&](steady_clock::time_point ts) {
        auto ms = duration_cast<milliseconds>(ts.time_since_epoch()).count();
        std::cerr << "[overflow] fast_source at t=" << ms << "ms\n";
        overflow_count.fetch_add(1);
    });

Network-level event handler:

    // Network-level aggregate handler — covers every node, includes node name.
    net.set_event_handler([](std::string_view name, NodeEvent ev,
                             steady_clock::time_point ts) {
        auto ms = duration_cast<milliseconds>(ts.time_since_epoch()).count();
        std::string_view kind = (ev == NodeEvent::Overflow) ? "overflow" : "closed";
        std::cerr << "[net:" << kind << "] node=" << name << " t=" << ms << "ms\n";
    });