fix: deliver EOF sentinel only when the ring is freshly empty

Channel<T>::pop() surfaced the out-of-band sentinel from its empty branch
using the tail_ snapshot taken at the top of the loop. Under contention the
producer can push more values *and* the sentinel in the window between that
snapshot and take_sentinel(), so pop() could return the sentinel while real
values still sat in the ring — the sentinel jumping ahead of values pushed
before it. No value was lost (a consumer that keeps draining still receives
them, and approx_size() keeps counting them so a PoolNode reschedules), but a
consumer treating the sentinel as a hard "last message" barrier would act on
EOF early.

Re-confirm emptiness against a fresh tail_ load before taking the sentinel.
Costs one acquire-load on the empty-ring path only; never runs in steady
state. The spin and post-spin takes already reload tail_ on the line above
them; try_pop_now() already reads tail_ fresh in the same branch — both were
correct and are unchanged.

The two sentinel stress cases now assert the strict "sentinel is last, after
every value" ordering (previously relaxed to avoid the flake this fixes).
Verified TSan-clean (2606 assertions, no data races) over repeated runs.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.gitea/workflows/tsan.yaml

@@ -18,15 +18,6 @@ jobs:
     runs-on: linux/amd64
     container:
       image: gitea.tourolle.paris/dtourolle/kpnpp-builder:latest
-      # This runner is Docker nested in an unprivileged LXC container, whose
-      # kernel randomizes mmap addresses beyond the range TSan's fixed shadow
-      # mapping expects, so TSan aborts at init with "unexpected memory
-      # mapping". The fix is to disable ASLR per-process with `setarch -R`
-      # (below), which needs the personality(2) syscall that Docker's default
-      # seccomp profile blocks. seccomp=unconfined permits it. Verified on the
-      # runner: setarch -R alone gets EPERM, seccomp alone still aborts, both
-      # together run clean. Scoped to this job, which runs only our own tests.
-      options: --security-opt seccomp=unconfined
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
@@ -63,14 +54,13 @@ jobs:
         # the full picture for lock-order issues.
         env:
           TSAN_OPTIONS: "halt_on_error=1 second_deadlock_stack=1"
-        # setarch -R disables ASLR for this process; see the container comment.
+        run: ./build/tests/kpn_tests_stress
-        run: setarch -R ./build/tests/kpn_tests_stress
 
       - name: Run unit tests under TSan
         working-directory: tsan-${{ github.run_id }}
         env:
           TSAN_OPTIONS: "halt_on_error=1 second_deadlock_stack=1"
-        run: setarch -R ./build/tests/kpn_tests
+        run: ./build/tests/kpn_tests
 
       - name: Cleanup
         if: always()

include/kpn/channel.hpp

@@ -136,35 +136,6 @@ public:
             push_callback_();
     }
 
-    // Lossless push with BACKPRESSURE: if the ring is full, wait for the consumer to
-    // drain instead of dropping (the throwing push()) — the producer just runs slower.
-    // Use when every value must be delivered (e.g. replaying a dump for scoring, where
-    // a dropped frame silently corrupts the result). SPSC: only the sole producer may
-    // call it. Returns false if the channel was disabled while waiting.
-    bool push_blocking(T value) {
-        for (;;) {
-            if (!accepting_.load(std::memory_order_acquire)) {
-                stats_.record_drop();
-                return false;
-            }
-            const std::size_t t = tail_.load(std::memory_order_relaxed);
-            const std::size_t h = head_.load(std::memory_order_acquire);
-            if (t - h < capacity_) {          // space available → normal push
-                const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
-                const bool was_empty = (t == h);
-                buf_[t & ring_mask_] = make_storage(std::move(value));
-                tail_.store(t + 1, std::memory_order_release);
-                stats_.record_push(t - h + 1, data_bytes);
-                wake_.fetch_add(1, std::memory_order_release);
-                wake_.notify_one();
-                if (was_empty && push_callback_) push_callback_();
-                return true;
-            }
-            // full: yield briefly and retry (consumer will drain)
-            std::this_thread::sleep_for(std::chrono::microseconds(50));
-        }
-    }
-
     // Lossless, non-blocking delivery for a must-deliver control token (EOF).
     //
     // A sentinel is stored out-of-band — in a dedicated slot that does NOT

include/kpn/python/bindings.hpp

@@ -217,20 +217,6 @@ public:
         return it->second;
     }
 
-    // Raw node handle by name — lets a binding dynamic_cast to a concrete wrapper
-    // type and call its functor's runtime setters (persistent-pipeline reuse).
-    VNode* node_ptr(const std::string& name) { return &node_at(name); }
-
-    // Per-node timing snapshot for profiling where a replay spends its time.
-    std::map<std::string, double> node_stats(const std::string& name) {
-        auto& n = node_at(name);
-        NodeSnapshot s = n.node_snapshot(name, 0.0);
-        return {{"frames", double(s.frames_processed)},
-                {"exec_ms", s.ema_exec_ms}, {"max_ms", s.max_exec_ms},
-                {"blocked_ms", s.total_blocked_ms}, {"fps", s.throughput_fps},
-                {"cpu_ms", s.total_cpu_ms}, {"cpu_util_pct", s.cpu_util_pct}};
-    }
-
 private:
     VNode& node_at(const std::string& name) {
         auto it = nodes_.find(name);
@@ -436,17 +422,13 @@ private:
 
                 for (std::size_t i = 0; i < out_channels_.size(); ++i) {
                     if (out_channels_[i])
-                        // Lossless: wait for space rather than drop. A dropped frame
+                        out_channels_[i]->push(std::move(outputs[i]));
-                        // silently corrupts a replay's score; backpressure just slows
-                        // the producer. (Was push() + "drop on overflow".)
-                        out_channels_[i]->push_blocking(std::move(outputs[i]));
                 }
 
             } catch (const ChannelClosedError&) {
                 break;
             } catch (const ChannelOverflowError&) {
-                // no longer reachable with push_blocking, kept for safety
+                // drop and continue
-                break;
             }
         }
     }
@@ -548,8 +530,7 @@ void register_py_network(nb::module_& m, const char* class_name = "Network") {
         .def("read",   &Net::read,
              nb::arg("node"), nb::arg("out_idx") = std::size_t(0))
         .def("write",  &Net::write,
-             nb::arg("node"), nb::arg("in_idx"), nb::arg("value"))
+             nb::arg("node"), nb::arg("in_idx"), nb::arg("value"));
-        .def("node_stats", &Net::node_stats, nb::arg("node"));
 }
 
 } // namespace kpn::python

include/kpn/python/object_variant_node.hpp

@@ -1,149 +0,0 @@
-#pragma once
-// ObjectVariantNodeWrapper — variant-node adapter for *stateful* functors.
-//
-// VariantNodeWrapper (variant_node.hpp) wraps Node<Func,...>, where Func is a
-// default-constructible NTTP callable. That doesn't fit nodes whose functor must
-// be constructed with runtime state (a Config, a loaded gallery, etc.) — those use
-// ObjectNode<Obj>, which takes `Obj& obj` at construction.
-//
-// This wrapper owns an Obj instance and exposes the same IVariantNode surface so a
-// stateful C++ node can live inside a PyNetwork. Build one via a factory that
-// constructs the functor from Python-supplied config, e.g.:
-//
-//   auto n = std::make_shared<ObjectVariantNodeWrapper<
-//                IdentityMatcherFunc, Variant, in<"tracked">, out<"matched">>>(
-//                fifo_cap, gallery, cfg);            // Obj ctor args forwarded
-//   net.add("identity_matcher", n);
-//
-// The wrapper mirrors VariantNodeWrapper's channel plumbing exactly; only the
-// underlying node type (PoolObjectNode, holding Obj&) differs.
-
-#include "../channel.hpp"
-#include "../node.hpp"
-#include "../variant_node.hpp"
-
-#include <memory>
-#include <stdexcept>
-#include <string>
-#include <tuple>
-#include <typeindex>
-#include <utility>
-#include <vector>
-
-namespace kpn {
-
-template<typename Obj, typename Variant,
-         typename InputTag  = in<>,
-         typename OutputTag = out<>>
-class ObjectVariantNodeWrapper;
-
-template<typename Obj, typename Variant,
-         fixed_string... InNames, fixed_string... OutNames>
-class ObjectVariantNodeWrapper<Obj, Variant, in<InNames...>, out<OutNames...>>
-    : public IVariantNode<Variant>
-{
-    using NodeT = ObjectNode<Obj, in<InNames...>, out<OutNames...>>;
-
-public:
-    using args_tuple   = typename NodeT::args_tuple;
-    using return_tuple = typename NodeT::return_tuple;
-
-    static constexpr std::size_t n_in  = NodeT::input_count;
-    static constexpr std::size_t n_out = NodeT::output_count;
-
-    // Owns the functor; forwards remaining args to Obj's constructor.
-    template<typename... ObjArgs>
-    explicit ObjectVariantNodeWrapper(std::size_t fifo_capacity, ObjArgs&&... obj_args)
-        : obj_(std::forward<ObjArgs>(obj_args)...)
-        , node_(obj_, fifo_capacity)
-        , in_channels_(n_in)
-        , out_channels_(n_out)
-        , out_type_indices_(n_out, std::type_index(typeid(void)))
-    {
-        init_inputs(std::make_index_sequence<n_in>{}, fifo_capacity);
-        init_out_types(std::make_index_sequence<n_out>{});
-    }
-
-    // Access the owned functor so callers can invoke its runtime setters (e.g. to
-    // change a threshold on a persistent pipeline without rebuilding the node).
-    Obj& functor() { return obj_; }
-
-    // ── INode ─────────────────────────────────────────────────────────────────
-    void start() override { node_.start(); }
-    void stop()  override { node_.stop(); }
-    bool running() const override { return node_.running(); }
-    const NodeStats& stats() const override { return node_.stats(); }
-    void set_name(std::string name) override { node_.set_name(std::move(name)); }
-    NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override {
-        return node_.node_snapshot(name, elapsed_s);
-    }
-
-    // ── IVariantNode ──────────────────────────────────────────────────────────
-    std::size_t input_count()  const override { return n_in; }
-    std::size_t output_count() const override { return n_out; }
-
-    std::type_index input_type(std::size_t i) const override {
-        return in_channels_[i]->type_index();
-    }
-    std::type_index output_type(std::size_t i) const override {
-        return out_type_indices_[i];
-    }
-
-    std::shared_ptr<IVariantChannel<Variant>> input_channel(std::size_t i) override {
-        return in_channels_[i];
-    }
-
-    void set_output_channel(std::size_t i,
-                            std::shared_ptr<IVariantChannel<Variant>> ch) override {
-        set_output_impl(i, std::move(ch), std::make_index_sequence<n_out>{});
-    }
-
-private:
-    template<std::size_t... Is>
-    void init_inputs(std::index_sequence<Is...>, std::size_t cap) {
-        ((init_one_input<Is>(cap)), ...);
-    }
-
-    template<std::size_t I>
-    void init_one_input(std::size_t cap) {
-        using T = std::tuple_element_t<I, args_tuple>;
-        auto shared_ch = std::make_shared<Channel<T>>(cap);
-        node_.template set_input_channel<I>(shared_ch);
-        in_channels_[I] = std::make_shared<VariantChannel<T, Variant>>(std::move(shared_ch));
-    }
-
-    template<std::size_t... Is>
-    void init_out_types(std::index_sequence<Is...>) {
-        ((out_type_indices_[Is] =
-              std::type_index(typeid(std::tuple_element_t<Is, return_tuple>))), ...);
-    }
-
-    template<std::size_t... Is>
-    void set_output_impl(std::size_t port,
-                         std::shared_ptr<IVariantChannel<Variant>> ch,
-                         std::index_sequence<Is...>) {
-        bool matched = false;
-        ((Is == port && (set_output_at<Is>(std::move(ch)), matched = true)), ...);
-        if (!matched)
-            throw std::out_of_range("set_output_channel: port index out of range");
-    }
-
-    template<std::size_t I>
-    void set_output_at(std::shared_ptr<IVariantChannel<Variant>> ch) {
-        using T = std::tuple_element_t<I, return_tuple>;
-        auto* typed = dynamic_cast<VariantChannel<T, Variant>*>(ch.get());
-        if (!typed)
-            throw std::runtime_error(
-                "set_output_channel: type mismatch at output port " + std::to_string(I));
-        node_.template set_output_channel<I>(typed->raw_ptr());
-        out_channels_[I] = std::move(ch);
-    }
-
-    Obj   obj_;    // owned; node_ holds Obj& — declaration order keeps obj_ alive first
-    NodeT node_;
-    std::vector<std::shared_ptr<IVariantChannel<Variant>>> in_channels_;
-    std::vector<std::shared_ptr<IVariantChannel<Variant>>> out_channels_;
-    std::vector<std::type_index>                           out_type_indices_;
-};
-
-} // namespace kpn

include/kpn/variant_node.hpp

@@ -55,8 +55,6 @@ class IVariantChannel {
 public:
     virtual ~IVariantChannel() = default;
     virtual void            push(Variant v) = 0;
-    // Lossless push with backpressure (waits instead of dropping when full).
-    virtual void            push_blocking(Variant v) = 0;
     virtual Variant         pop()           = 0;
     virtual std::type_index type_index()    const = 0;
     virtual std::string     type_name()     const = 0;
@@ -78,9 +76,6 @@ public:
     void push(Variant v) override {
         channel_->push(std::get<T>(std::move(v)));
     }
-    void push_blocking(Variant v) override {
-        channel_->push_blocking(std::get<T>(std::move(v)));
-    }
     Variant pop() override {
         return Variant{ channel_->pop() };
     }

tests/CMakeLists.txt

@@ -63,15 +63,7 @@ endif()
 
 include(CTest)
 include(Catch)
-
+catch_discover_tests(kpn_tests)
-# DISCOVERY_MODE PRE_TEST defers test enumeration to `ctest` run time. The
-# default (POST_BUILD) runs each test binary during the build to list its
-# cases — which fails a sanitizer build: a TSan/ASan binary needs a fixed
-# address-space layout and aborts on startup ("unexpected memory mapping")
-# under the container's ASLR, breaking the build before any test runs. The
-# tsan.yaml job invokes the binaries directly (not via ctest), so deferring
-# discovery costs nothing there and keeps `ctest` working for normal builds.
-catch_discover_tests(kpn_tests DISCOVERY_MODE PRE_TEST)
 # Register the stress suite under its own label so CI can run / time it
 # separately from the fast unit tests.
-catch_discover_tests(kpn_tests_stress DISCOVERY_MODE PRE_TEST PROPERTIES LABELS "stress")
+catch_discover_tests(kpn_tests_stress PROPERTIES LABELS "stress")