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: setarch -R ./build/tests/kpn_tests_stress
+        run: ./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/fanout.hpp

@@ -115,16 +115,6 @@ public:
         out_channels_[I] = ch;
     }
 
-    // Lossless fanout: block until each consumer drains rather than dropping.
-    void set_lossless_output(bool on) override { lossless_ = on; }
-
-    // Opt a single output back out of blocking.  Needed when one branch may
-    // stall indefinitely — a display tap nobody is servicing, say — since
-    // blocking on it would apply backpressure to every other branch too.
-    void set_lossy_output(std::size_t i, bool lossy = true) {
-        if (i < N) lossy_out_[i] = lossy;
-    }
-
 private:
     void run_loop() {
         while (!stop_flag_.load(std::memory_order_relaxed)) {
@@ -136,19 +126,8 @@ private:
 
                 for (std::size_t i = 0; i < N; ++i) {
                     if (out_channels_[i]) {
-                        // Lossless: block until this consumer drains.  Note the
-                        // branches differ in more than blocking — the dropping
-                        // path discards per-output independently and silently,
-                        // so a slow consumer on one branch costs frames on that
-                        // branch only, with no diagnostic.  That is the right
-                        // default for display taps but hides frame loss from
-                        // analysis branches.
-                        if (lossless_ && !lossy_out_[i])
-                            out_channels_[i]->push_blocking(val);
-                        else {
-                            try { out_channels_[i]->push(val); }
-                            catch (const ChannelOverflowError&) {}  // drop independently
-                        }
+                        try { out_channels_[i]->push(val); }
+                        catch (const ChannelOverflowError&) {}  // drop for this output independently
                     }
                 }
 
@@ -163,8 +142,6 @@ private:
 
     std::string                 name_;
     std::size_t                 fifo_capacity_;
-    bool                        lossless_{false};
-    std::array<bool, N>         lossy_out_{};   // per-output opt-out of blocking
     std::shared_ptr<Channel<T>> input_ch_;
     std::array<Channel<T>*, N>  out_channels_{};
     std::atomic<bool>           stop_flag_{false};

include/kpn/inode.hpp

@@ -37,10 +37,6 @@ struct INode {
     // halt(): alias for stop() — immediate, discards in-flight work.
     virtual void halt() { stop(); }
 
-    // Opt into lossless (blocking) output for nodes that support it.  Default
-    // is a no-op so node types with no output channels ignore it.
-    virtual void set_lossless_output(bool) {}
-
     // shutdown(): graceful drain before stopping. Base implementation falls
     // back to stop(). Network and StaticNetwork override with topo-ordered drain.
     virtual void shutdown() { stop(); }

include/kpn/pool_node.hpp

@@ -133,14 +133,6 @@ public:
     void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
     void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
 
-    // Lossless output: block the producer until the consumer drains rather than
-    // dropping on a full channel.  Default is drop, which suits live sources
-    // where a stale frame is worth less than a fresh one.  Enable for offline
-    // batch runs, where a dropped item leaves a gap that downstream analysis
-    // cannot recover.  Must be set before start().
-    void set_lossless_output(bool on) override { lossless_ = on; }
-    void set_lossless(bool on) { set_lossless_output(on); }
-
     void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); }
     void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); }
     void set_closed_callback(NodeEventCallback cb)   { closed_callbacks_[0] = std::move(cb); }
@@ -408,15 +400,6 @@ private:
             ch->push_sentinel(std::move(val));
             return;
         }
-        // Lossless mode: block until the consumer drains instead of dropping.
-        // Dropping is the right default for live sources (a stale frame is
-        // worth less than a fresh one), but for offline batch work every sample
-        // matters — dropped frames leave a non-uniformly sampled series, which
-        // silently invalidates any fixed-rate spectral analysis downstream.
-        if (lossless_) {
-            ch->push_blocking(std::move(val));
-            return;
-        }
         try {
             ch->push(std::move(val));
         } catch (const ChannelOverflowError&) {
@@ -439,7 +422,6 @@ private:
 
     std::shared_ptr<IScheduler>              scheduler_;
     std::string                              name_;
-    bool                                     lossless_{false};
     std::size_t                              fifo_capacity_;
     input_channels_t                         input_channels_;
     output_channels_t                        output_channels_{};
@@ -516,14 +498,6 @@ public:
     void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
     void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
 
-    // Lossless output: block the producer until the consumer drains rather than
-    // dropping on a full channel.  Default is drop, which suits live sources
-    // where a stale frame is worth less than a fresh one.  Enable for offline
-    // batch runs, where a dropped item leaves a gap that downstream analysis
-    // cannot recover.  Must be set before start().
-    void set_lossless_output(bool on) override { lossless_ = on; }
-    void set_lossless(bool on) { set_lossless_output(on); }
-
     void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); }
     void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); }
     void set_closed_callback(NodeEventCallback cb)   { closed_callbacks_[0] = std::move(cb); }
@@ -732,11 +706,6 @@ private:
             ch->push_sentinel(std::move(val));
             return;
         }
-        // Lossless mode: block until the consumer drains instead of dropping.
-        if (lossless_) {
-            ch->push_blocking(std::move(val));
-            return;
-        }
         try {
             ch->push(std::move(val));
         } catch (const ChannelOverflowError&) {
@@ -748,7 +717,6 @@ private:
     Obj&                                     obj_;
     std::shared_ptr<IScheduler>              scheduler_;
     std::string                              name_;
-    bool                                     lossless_{false};
     std::size_t                              fifo_capacity_;
     input_channels_t                         input_channels_;
     output_channels_t                        output_channels_{};

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
-                        // 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]));
+                        out_channels_[i]->push(std::move(outputs[i]));
                 }
 
             } catch (const ChannelClosedError&) {
                 break;
             } catch (const ChannelOverflowError&) {
-                // no longer reachable with push_blocking, kept for safety
-                break;
+                // drop and continue
             }
         }
     }
@@ -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"))
-        .def("node_stats", &Net::node_stats, nb::arg("node"));
+             nb::arg("node"), nb::arg("in_idx"), nb::arg("value"));
 }
 
 } // 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/static_network.hpp

@@ -219,24 +219,6 @@ public:
 
     FanoutStorage& fanouts_storage() { return *fanouts_; }
 
-    // Make every node in this network push losslessly (block until the consumer
-    // drains) instead of dropping on a full channel.  Includes the fanout nodes
-    // make_network() inserts automatically, which is the part user code cannot
-    // reach: they are unnamed, and they drop silently per-output, so a network
-    // whose own nodes are all lossless can still lose items at a fanout.
-    //
-    // Only safe when every consumer eventually drains.  A branch that can stall
-    // indefinitely — a display node nobody is servicing, say — will block the
-    // whole pipeline through backpressure.  Call before start().
-    void set_lossless(bool on = true) {
-        for (auto* n : user_nodes_topo_)  if (n) n->set_lossless_output(on);
-        for (auto* n : fanout_nodes_ptr_) if (n) n->set_lossless_output(on);
-    }
-
-    // Block until every channel is empty.  Useful before stop() so work already
-    // in flight completes rather than being discarded at teardown.
-    void drain() const { drain_all_channels(); }
-
 private:
     struct Snapshots {
         std::vector<NodeSnapshot>     nodes;

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)
-
-# 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)
+catch_discover_tests(kpn_tests)
 # 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")