ci: fix TSan aborting at init on the nested-LXC runner

The ThreadSanitizer job runs on Docker nested in an unprivileged LXC
container, whose kernel randomizes mmap addresses beyond the range TSan's
fixed shadow mapping expects. TSan aborted at init with "unexpected memory
mapping" before any test ran.

Disable ASLR per-process with `setarch -R`, which needs the personality(2)
syscall that Docker's default seccomp profile blocks; seccomp=unconfined on
the container permits it. Verified on the runner that both are required:
setarch -R alone gets EPERM, seccomp alone still aborts, both together run
clean. Scoped to the tsan job, which runs only our own test binaries.

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

.gitea/workflows/ci.yaml

@@ -63,7 +63,8 @@ jobs:
     if: ${{ needs.changes.outputs.dockerfile == 'true' }}
     uses: ./.gitea/workflows/docker.yaml
     with:
-      push: ${{ github.event_name != 'pull_request' }}
+      # Explicit string, not a boolean expression (act_runner mangles bools).
+      push: ${{ github.event_name == 'pull_request' && 'false' || 'true' }}
 
   # Runs after docker (if docker ran). A skipped docker job is fine; a failed
   # one blocks this via !failure(). Re-run tests when code OR the image changed.
@@ -72,6 +73,13 @@ jobs:
     if: ${{ !failure() && !cancelled() && (needs.changes.outputs.code == 'true' || needs.changes.outputs.dockerfile == 'true') }}
     uses: ./.gitea/workflows/test.yaml
 
+  # ThreadSanitizer run for the lock-free Channel<T>. Same trigger conditions as
+  # test (code or image changed); runs in parallel with test.
+  tsan:
+    needs: [changes, docker]
+    if: ${{ !failure() && !cancelled() && (needs.changes.outputs.code == 'true' || needs.changes.outputs.dockerfile == 'true') }}
+    uses: ./.gitea/workflows/tsan.yaml
+
   docs:
     needs: [changes, docker]
     if: ${{ !failure() && !cancelled() && github.ref == 'refs/heads/master' && (needs.changes.outputs.docs == 'true' || needs.changes.outputs.dockerfile == 'true') }}

.gitea/workflows/docker.yaml

@@ -4,19 +4,22 @@ name: '🐳 Builder Image'
 # It is called by ci.yaml only when Dockerfile.builder or docs/requirements.txt
 # change. It runs on the host runner (NOT inside the builder container) because
 # it needs the Docker CLI/daemon.
+# Note: `push` is a STRING ("true"/"false"), not a boolean. Gitea's act_runner
+# mangles boolean inputs passed from an expression (they arrive as false), so we
+# pass an explicit string and compare with == 'true' below.
 on:
   workflow_call:
     inputs:
       push:
-        description: 'Push the built image to the registry'
+        description: 'Push the built image to the registry ("true"/"false")'
-        type: boolean
+        type: string
-        default: true
+        default: 'true'
   workflow_dispatch:
     inputs:
       push:
-        description: 'Push the built image to the registry'
+        description: 'Push the built image to the registry ("true"/"false")'
-        type: boolean
+        type: string
-        default: true
+        default: 'true'
 
 jobs:
   build:
@@ -48,7 +51,7 @@ jobs:
             .
 
       - name: Push builder image
-        if: ${{ inputs.push }}
+        if: ${{ inputs.push == 'true' }}
         run: |
           docker push gitea.tourolle.paris/dtourolle/kpnpp-builder:latest
           docker push gitea.tourolle.paris/dtourolle/kpnpp-builder:${{ github.sha }}

.gitea/workflows/tsan.yaml

@@ -0,0 +1,77 @@
+name: '🧵 ThreadSanitizer'
+
+# Reusable workflow: builds the channel stress suite with ThreadSanitizer and
+# runs it. This is the dynamic half of verifying the lock-free SPSC Channel<T>
+# (the static half is the CDSChecker model-check harness in verify/).
+#
+# Triggering and path filtering are owned by ci.yaml (the orchestrator), which
+# calls this only when code changed. workflow_dispatch is kept for manual runs.
+#
+# Runs in the prebuilt builder image (gcc:14), which already ships libtsan — no
+# package installs at job time.
+on:
+  workflow_call:
+  workflow_dispatch:
+
+jobs:
+  tsan:
+    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
+        with:
+          path: tsan-${{ github.run_id }}
+
+      - name: Cache FetchContent dependencies
+        uses: actions/cache@v3
+        with:
+          path: ~/.cmake/fetchcontent
+          key: cmake-fetchcontent-${{ hashFiles('**/CMakeLists.txt') }}
+          restore-keys: cmake-fetchcontent-
+
+      - name: Configure (TSan)
+        working-directory: tsan-${{ github.run_id }}
+        run: |
+          cmake -S . -B build \
+            -G Ninja \
+            -DCMAKE_BUILD_TYPE=Debug \
+            -DKPN_SANITIZER=thread \
+            -DKPN_BUILD_TESTS=ON \
+            -DKPN_BUILD_EXAMPLES=OFF \
+            -DKPN_BUILD_PYTHON=OFF \
+            -DFETCHCONTENT_BASE_DIR=$HOME/.cmake/fetchcontent
+
+      - name: Build (TSan)
+        working-directory: tsan-${{ github.run_id }}
+        run: cmake --build build --parallel --target kpn_tests kpn_tests_stress
+
+      - name: Run stress suite under TSan
+        working-directory: tsan-${{ github.run_id }}
+        # halt_on_error=1 makes the first detected race fail the job; the report
+        # (with both stacks) is printed to the log. second_deadlock_stack gives
+        # 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
+
+      - 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
+
+      - name: Cleanup
+        if: always()
+        run: rm -rf tsan-${{ github.run_id }}

CMakeLists.txt

@@ -10,6 +10,30 @@ option(KPN_BUILD_PYTHON   "Build Python bindings (requires nanobind)" ON)
 option(KPN_BUILD_EXAMPLES "Build examples"                            ON)
 option(KPN_WEB_DEBUG      "Enable web debug UI (cpp-httplib)"         OFF)
 
+# Sanitizer build. Empty = off. Accepts "thread", "address", "undefined",
+# or a combination like "address,undefined". Applied to all kpn targets via
+# the kpn_sanitizer_flags() helper below.
+#
+# The lock-free SPSC Channel<T> (include/kpn/channel.hpp) has hand-reasoned
+# acquire/release ordering; -DKPN_SANITIZER=thread + the channel stress test
+# (tests/test_channel_stress.cpp) is the dynamic half of verifying it. The
+# static half is the CDSChecker model-check harness (see verify/).
+set(KPN_SANITIZER "" CACHE STRING
+    "Build with sanitizer: thread | address | undefined | <combo> (empty = off)")
+
+# Translate KPN_SANITIZER into compile/link flags. No-op when empty.
+function(kpn_sanitizer_flags out_var)
+    if(KPN_SANITIZER)
+        set(${out_var}
+            -fsanitize=${KPN_SANITIZER}
+            -fno-omit-frame-pointer
+            -g
+            PARENT_SCOPE)
+    else()
+        set(${out_var} "" PARENT_SCOPE)
+    endif()
+endfunction()
+
 # ── Core library (header-only) ────────────────────────────────────────────────
 add_library(kpn INTERFACE)
 target_include_directories(kpn INTERFACE

Dockerfile.builder

@@ -1,4 +1,4 @@
-# KPN++ Builder Image  (CI: pipeline trigger)
+# KPN++ Builder Image  (CI: pipeline trigger v2)
 # Pre-built image with GCC, CMake, Ninja, and Python dev headers for building and testing KPN++
 # Build: docker build -f Dockerfile.builder -t gitea.tourolle.paris/dtourolle/kpnpp-builder:latest .
 # Push:  docker push gitea.tourolle.paris/dtourolle/kpnpp-builder:latest

README.md

@@ -2,6 +2,8 @@
 
 A C++20 Kahn Process Network (KPN) library. Each node wraps a function and runs in its own thread, communicating with downstream nodes via bounded FIFO channels. Includes Python bindings via nanobind.
 
+📖 **[Documentation](https://pages.tourolle.paris/dtourolle/kpn/)**
+
 ---
 
 ## Requirements
@@ -398,8 +400,8 @@ Violating the second rule deadlocks.
 | `04_storage_policy` | `channel_storage_policy` default and specialisation |
 | `05_error_handling` | `ChannelOverflowError`, `ErrorHandler` |
 | `06_watchdog` | Watchdog interval, stall detection |
-| `07_python_network` | PyNetwork, pure Python node *(pending)* |
+| `07_python_network` | PyNetwork with a pure-Python node between a C++ source and sink |
-| `08_python_subport` | `net.read`, `net.write`, sub-port tap *(pending)* |
+| `08_python_subport` | Drive a Python node from Python via `net.write`/`net.read` sub-port taps |
 | `09_opencv_cellshade` | Real-time cell-shading on webcam/pattern; requires OpenCV ≥ 4 |
 
 Run the cell-shading example:

README.md.in

@@ -2,6 +2,8 @@
 
 A C++20 Kahn Process Network (KPN) library. Each node wraps a function and runs in its own thread, communicating with downstream nodes via bounded FIFO channels. Includes Python bindings via nanobind.
 
+📖 **[Documentation](https://pages.tourolle.paris/dtourolle/kpn/)**
+
 ---
 
 ## Requirements
@@ -220,8 +222,8 @@ Violating the second rule deadlocks.
 | `04_storage_policy` | `channel_storage_policy` default and specialisation |
 | `05_error_handling` | `ChannelOverflowError`, `ErrorHandler` |
 | `06_watchdog` | Watchdog interval, stall detection |
-| `07_python_network` | PyNetwork, pure Python node *(pending)* |
+| `07_python_network` | PyNetwork with a pure-Python node between a C++ source and sink |
-| `08_python_subport` | `net.read`, `net.write`, sub-port tap *(pending)* |
+| `08_python_subport` | Drive a Python node from Python via `net.write`/`net.read` sub-port taps |
 | `09_opencv_cellshade` | Real-time cell-shading on webcam/pattern; requires OpenCV ≥ 4 |
 
 Run the cell-shading example:

examples/07_python_network/example.py

@@ -30,3 +30,8 @@ net.build()
 net.start()
 time.sleep(0.1)
 net.stop()
+
+# Drop the network deterministically: it holds the Python callable, which forms
+# a reference cycle via globals(). Deleting the global breaks it so the network
+# is reclaimed now rather than lingering to interpreter shutdown.
+del net

examples/08_python_subport/example.py

@@ -1,38 +1,55 @@
 """
-08_python_subport — tap a C++ node's output from Python using net.read().
+08_python_subport — drive a *Python* node from Python via write()/read() taps.
 
 Graph:
-  [ProduceNode] --int--> [DoubleItNode] --int--> (tapped by net.read())
+  (fed by net.write()) --int--> [py_triple] --int--> (tapped by net.read())
 
-The sink is Python: instead of connecting a PrintItNode, we call net.read()
+Unlike 07, there is no C++ source or sink here: the only node in the network is
-to pull values out of DoubleItNode's output directly into Python.
+a pure-Python function, py_triple. Python plays *both* the producer and the
-We also demonstrate net.write() by injecting a value into DoubleItNode's input.
+consumer by using the subport taps:
+
+  * net.write("py", 0, v)  injects v into py_triple's input   (Python -> network)
+  * net.read("py", 0)      pulls py_triple's output back out   (network -> Python)
+
+This closes the loop the old version left as a "#todo": a value flows from
+Python, through a Python node running inside the network, and back to Python.
 """
 
 import sys
-import time
+sys.path.insert(0, "build/python")  # for `python examples/.../example.py` from repo root
-import threading
-sys.path.insert(0, "build/python")
 
 import kpn_python as kpn
 
+
+def py_triple(x: int) -> int:
+    return x * 3
+
+
 net = kpn.Network()
 
-net.add("src", kpn.make_produce())
+# The whole network is a single Python node with a tapped input and output.
-net.add("dbl", kpn.make_double_it())
+net.add_node("py", py_triple, inputs=["int"], outputs=["int"])
 
-net.connect("src", 0, "dbl", 0)
 net.build()
 net.start()
 
-# Collect a few values from DoubleItNode's output via Python tap
+# Push values in from Python and read the Python node's results back out.
+inputs = [1, 2, 7, 10, 100]
 results = []
-for _ in range(5):
+for v in inputs:
-    val = net.read("dbl", 0)
+    net.write("py", 0, v)          # Python -> py_triple input
-    results.append(val)
+    results.append(net.read("py", 0))  # py_triple output -> Python
 
 net.stop()
 
-print("values read from C++ DoubleItNode output:", results)
+print("inputs written from Python: ", inputs)
-assert all(v == 84 for v in results), f"expected all 84, got {results}"
+print("outputs read from py_triple:", results)
-print("all correct (42 * 2 = 84)")
+
+expected = [v * 3 for v in inputs]
+assert results == expected, f"expected {expected}, got {results}"
+print("all correct (x * 3 computed by a Python node inside the network)")
+
+# Drop the network deterministically. The network holds the Python callable,
+# which (via globals) forms a reference cycle; deleting the global breaks it so
+# the network is reclaimed promptly instead of lingering to interpreter exit.
+del net

examples/CMakeLists.txt

@@ -13,6 +13,25 @@ function(kpn_example name)
     )
 endfunction()
 
+# Register a Python example script as a CTest smoke test. Runs the script with
+# PYTHONPATH pointing at the freshly-built kpn_python module, so it does not
+# depend on the caller's working directory or a hard-coded "build/python" path.
+function(kpn_python_example name)
+    if(NOT KPN_BUILD_PYTHON)
+        return()
+    endif()
+    add_test(
+        NAME    example_${name}
+        COMMAND ${CMAKE_COMMAND} -E env
+                "PYTHONPATH=$<TARGET_FILE_DIR:kpn_python>"
+                ${Python_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/${name}/example.py
+    )
+    set_tests_properties(example_${name} PROPERTIES
+        TIMEOUT 15
+        LABELS  examples
+    )
+endfunction()
+
 kpn_example(01_hello_pipeline)
 kpn_example(02_named_ports)
 kpn_example(03_multi_output)
@@ -31,7 +50,9 @@ if(KPN_WEB_DEBUG)
     target_link_libraries(14_debug_hub PRIVATE kpn)
     kpn_target_enable_web_debug(14_debug_hub)
 endif()
-# 07 and 08 are Python scripts — no compiled target needed.
+# 07 and 08 are Python scripts — no compiled target, but run as smoke tests.
+kpn_python_example(07_python_network)
+kpn_python_example(08_python_subport)
 
 # 09 requires OpenCV — only build if found
 find_package(OpenCV QUIET COMPONENTS core imgproc highgui videoio)

include/kpn/channel.hpp

@@ -180,7 +180,16 @@ public:
             if (h == t) {
                 // Ring drained — deliver any pending out-of-band sentinel (EOF)
                 // now, so it always arrives after the data pushed before it.
-                { T s; if (take_sentinel(s)) return s; }
+                //
+                // Re-confirm emptiness against a fresh tail_ first: the snapshot
+                // at the top of the loop may be stale (the producer can push more
+                // values *and* the sentinel in the window since), and the sentinel
+                // must never jump ahead of ring values pushed before it. The spin
+                // and post-spin takes below already reload tail_ on the line above
+                // them; this is the one take that used the loop-top snapshot.
+                if (h == tail_.load(std::memory_order_acquire)) {
+                    T s; if (take_sentinel(s)) return s;
+                }
 
                 if (!accepting_.load(std::memory_order_acquire))
                     throw ChannelClosedError{};

include/kpn/python/auto_bind.hpp

@@ -247,7 +247,7 @@ void bind_network(nb::module_& m) {
 
     nb::class_<IVariantNode<Variant>>(m, "INode");
 
-    nb::class_<Net>(m, "Network")
+    nb::class_<Net>(m, "Network", nb::type_slots(network_type_slots<Variant>()))
         .def("__init__", [](Net* self) {
             new (self) Net();
             register_all_converters<Registry>(*self);

include/kpn/python/bindings.hpp

@@ -35,6 +35,16 @@ public:
     using VNode    = IVariantNode<Variant>;
     using VChannel = IVariantChannel<Variant>;
 
+    // ── GC support ────────────────────────────────────────────────────────────
+    // Visit every Python object this network transitively holds (currently the
+    // callable of each PyNode). Used by the Network type's tp_traverse slot so
+    // Python's cyclic GC can discover instance → callable → globals() cycles.
+    // Defined out-of-line below, once PyNode is a complete type.
+    template<typename Fn>
+    void visit_python_objects(Fn&& visit) const;
+    // Drop all Python references held by nodes, breaking any cycle (tp_clear).
+    void clear_python_objects();
+
     // ── Builder API ───────────────────────────────────────────────────────────
 
     void add(std::string name, std::shared_ptr<VNode> node) {
@@ -367,6 +377,14 @@ public:
         out_channels_[i] = std::move(ch);
     }
 
+    // ── GC support (tp_traverse / tp_clear on the owning Network) ──────────────
+    // The node holds a Python callable, which typically forms an
+    // instance → callable → globals() → instance cycle. Expose the callable so
+    // the Network's GC slots can traverse and clear it. See bindings.hpp's
+    // network_tp_traverse/network_tp_clear.
+    const nb::object& python_callable() const { return callable_; }
+    void clear_python_callable() { callable_ = nb::object(); }
+
 private:
     void run_loop() {
         while (!stop_flag_.load(std::memory_order_relaxed)) {
@@ -439,6 +457,60 @@ private:
     NodeStats          stats_;
 };
 
+// ── PyNetwork GC helpers (defined here: PyNode is now complete) ────────────────
+
+template<typename Variant>
+template<typename Fn>
+void PyNetwork<Variant>::visit_python_objects(Fn&& visit) const {
+    for (const auto& [name, node] : nodes_)
+        if (auto* py = dynamic_cast<const PyNode<Variant>*>(node.get()))
+            visit(py->python_callable());
+}
+
+template<typename Variant>
+void PyNetwork<Variant>::clear_python_objects() {
+    for (auto& [name, node] : nodes_)
+        if (auto* py = dynamic_cast<PyNode<Variant>*>(node.get()))
+            py->clear_python_callable();
+}
+
+// ── GC type slots for the Network binding ─────────────────────────────────────
+// The Network holds Python callables (via PyNode), forming uncollectable
+// instance → callable → globals() → instance cycles at interpreter shutdown.
+// These slots let Python's cyclic collector traverse and break them, silencing
+// nanobind's leak warnings. See the nanobind "Reference leaks" documentation.
+
+template<typename Variant>
+int network_tp_traverse(PyObject* self, visitproc visit, void* arg) {
+    Py_VISIT(Py_TYPE(self));
+    if (!nb::inst_ready(self))
+        return 0;
+    auto* net = nb::inst_ptr<PyNetwork<Variant>>(self);
+    int rv = 0;
+    net->visit_python_objects([&](const nb::object& obj) {
+        if (rv == 0 && obj.is_valid())
+            rv = visit(obj.ptr(), arg);
+    });
+    return rv;
+}
+
+template<typename Variant>
+int network_tp_clear(PyObject* self) {
+    auto* net = nb::inst_ptr<PyNetwork<Variant>>(self);
+    net->clear_python_objects();
+    return 0;
+}
+
+template<typename Variant>
+PyType_Slot* network_type_slots() {
+    static PyType_Slot slots[] = {
+        { Py_tp_traverse, reinterpret_cast<void*>(&network_tp_traverse<Variant>) },
+        { Py_tp_clear,    reinterpret_cast<void*>(&network_tp_clear<Variant>) },
+        { 0, nullptr }
+    };
+    return slots;
+}
+
 // ── register_py_network (legacy helper) ───────────────────────────────────────
 // Registers PyNetwork<Variant> with the given nanobind module.
 // Prefer bind_network<Registry> from auto_bind.hpp for new code.
@@ -447,7 +519,7 @@ template<typename Variant>
 void register_py_network(nb::module_& m, const char* class_name = "Network") {
     using Net = PyNetwork<Variant>;
 
-    nb::class_<Net>(m, class_name)
+    nb::class_<Net>(m, class_name, nb::type_slots(network_type_slots<Variant>()))
         .def(nb::init<>())
         .def("connect", &Net::connect,
              nb::arg("src"), nb::arg("out_idx"),

tests/CMakeLists.txt

@@ -43,6 +43,35 @@ target_link_libraries(kpn_tests PRIVATE
     GTest::gtest
 )
 
+# ── Channel stress suite (separate executable) ────────────────────────────────
+# Contended SPSC tests for the lock-free Channel<T>. Kept out of kpn_tests
+# because each case runs many reps / tens of thousands of items and is slow.
+# Most valuable under -DKPN_SANITIZER=thread, but correct (and run) without it.
+add_executable(kpn_tests_stress test_channel_stress.cpp)
+target_link_libraries(kpn_tests_stress PRIVATE kpn Catch2::Catch2WithMain)
+
+# ── Sanitizer flags ───────────────────────────────────────────────────────────
+# kpn_sanitizer_flags() is defined in the top-level CMakeLists and is a no-op
+# unless -DKPN_SANITIZER=... is set. Sanitizer must be on both compile and link.
+kpn_sanitizer_flags(_kpn_san)
+if(_kpn_san)
+    foreach(_t kpn_tests kpn_tests_stress)
+        target_compile_options(${_t} PRIVATE ${_kpn_san})
+        target_link_options(${_t}    PRIVATE ${_kpn_san})
+    endforeach()
+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")

tests/test_channel_stress.cpp

@@ -0,0 +1,281 @@
+// Contended stress tests for the lock-free SPSC Channel<T>.
+//
+// The other channel tests (test_channel.cpp) are single-threaded or use a
+// single 20 ms sleep to order two threads — they never actually contend on the
+// ring, so they exercise neither the memory-ordering pairing nor the
+// spin/futex/lost-wakeup logic in pop().
+//
+// These tests are written to be run under ThreadSanitizer:
+//
+//   cmake -B build -DKPN_SANITIZER=thread -DKPN_BUILD_EXAMPLES=OFF -DKPN_BUILD_PYTHON=OFF
+//   cmake --build build --target kpn_tests_tsan
+//   ./build/tests/kpn_tests_tsan
+//
+// They are also valid (and meaningful) without a sanitizer: the value/sequence
+// assertions catch lost or duplicated items regardless of build flags. TSan
+// adds detection of the underlying data race even on runs where the race did
+// not corrupt observable state.
+//
+// Channel<T> is SPSC: exactly one producer thread and one consumer thread per
+// channel. Every scenario below honours that contract.
+
+#include <catch2/catch_test_macros.hpp>
+#include <atomic>
+#include <chrono>
+#include <kpn/channel.hpp>
+#include <thread>
+#include <vector>
+
+using namespace kpn;
+using namespace std::chrono_literals;
+
+namespace {
+
+// Repeat each scenario enough times that rare interleavings (spin window just
+// missing / just catching the next push, disable landing inside the futex
+// wait) actually occur across a run. Kept modest so a TSan run stays minutes,
+// not hours.
+constexpr int kReps = 200;
+
+} // namespace
+
+TEST_CASE("SPSC: every pushed item is popped exactly once, in order", "[channel][stress]") {
+    // Small capacity forces frequent full/empty transitions, so both the
+    // producer's overflow-retry and the consumer's spin->futex path are hit
+    // many times. The producer retries on overflow rather than dropping, so
+    // the consumer must observe a strictly contiguous 0..N-1 sequence.
+    constexpr int N = 50'000;
+    Channel<int> ch(/*capacity=*/4, /*spin_count=*/16);
+
+    std::thread producer([&] {
+        for (int i = 0; i < N; ++i) {
+            for (;;) {
+                try { ch.push(i); break; }
+                catch (const ChannelOverflowError&) { std::this_thread::yield(); }
+            }
+        }
+    });
+
+    int expected = 0;
+    bool in_order = true;
+    for (int i = 0; i < N; ++i) {
+        int v = ch.pop();
+        if (v != expected) in_order = false;
+        ++expected;
+    }
+    producer.join();
+
+    REQUIRE(in_order);
+    REQUIRE(expected == N);
+    REQUIRE(ch.size() == 0);
+}
+
+TEST_CASE("SPSC: tight empty<->non-empty transitions exercise spin/futex boundary",
+          "[channel][stress]") {
+    // spin_count=0 forces every empty pop() straight into atomic::wait, so this
+    // hammers the lost-wakeup guard (snapshot wake_, re-check tail_, then wait).
+    // The producer pushes one item then waits to go empty again, maximising the
+    // number of empty->non-empty edges relative to item count.
+    constexpr int N = 20'000;
+    Channel<int> ch(/*capacity=*/2, /*spin_count=*/0);
+
+    std::thread producer([&] {
+        for (int i = 0; i < N; ++i) {
+            for (;;) {
+                try { ch.push(i); break; }
+                catch (const ChannelOverflowError&) { std::this_thread::yield(); }
+            }
+        }
+    });
+
+    long sum = 0;
+    for (int i = 0; i < N; ++i) sum += ch.pop();
+    producer.join();
+
+    // Sum of 0..N-1 — detects any lost or duplicated item.
+    REQUIRE(sum == static_cast<long>(N) * (N - 1) / 2);
+}
+
+TEST_CASE("SPSC: disable() while consumer is blocked in pop() unblocks cleanly",
+          "[channel][stress]") {
+    // The data race of record: consumer blocked in pop() (spinning or parked in
+    // the futex) while the owner thread calls disable(). pop() must observe the
+    // close and throw ChannelClosedError — it must not hang and must not read
+    // past the ring. Repeated so disable() lands at many points in pop()'s loop.
+    for (int rep = 0; rep < kReps; ++rep) {
+        Channel<int> ch(/*capacity=*/4, /*spin_count=*/8);
+        std::atomic<bool> threw{false};
+        std::atomic<bool> finished{false};
+
+        std::thread consumer([&] {
+            try {
+                ch.pop();              // empty channel: will block
+            } catch (const ChannelClosedError&) {
+                threw.store(true, std::memory_order_relaxed);
+            }
+            finished.store(true, std::memory_order_relaxed);
+        });
+
+        // Give the consumer a chance to reach the wait, then close.
+        std::this_thread::sleep_for(50us);
+        ch.disable();
+
+        consumer.join();
+        REQUIRE(finished.load());
+        REQUIRE(threw.load());
+    }
+}
+
+TEST_CASE("SPSC: producer racing a disable() never throws and never hangs",
+          "[channel][stress]") {
+    // Mirror of the above from the producer side: push() racing disable() must
+    // either enqueue or silently drop, never throw ChannelClosedError and never
+    // wedge. Overflow is still a legal outcome (full accepting channel) and is
+    // tolerated here.
+    for (int rep = 0; rep < kReps; ++rep) {
+        Channel<int> ch(/*capacity=*/8, /*spin_count=*/8);
+        std::atomic<bool> bad{false};
+
+        std::thread producer([&] {
+            for (int i = 0; i < 1000; ++i) {
+                try { ch.push(i); }
+                catch (const ChannelOverflowError&) { /* legal: full */ }
+                catch (...) { bad.store(true, std::memory_order_relaxed); break; }
+            }
+        });
+
+        std::this_thread::sleep_for(20us);
+        ch.disable();           // owner closes mid-stream
+        producer.join();
+
+        REQUIRE_FALSE(bad.load());
+    }
+}
+
+TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition",
+          "[channel][stress]") {
+    // The empty->non-empty callback ([channel.hpp] was_empty branch) is read by
+    // the consumer-side notification path. Run it under contention to make sure
+    // the was_empty detection isn't torn by a concurrent pop().
+    Channel<int> ch(/*capacity=*/4, /*spin_count=*/4);
+    std::atomic<int> callbacks{0};
+    ch.set_push_callback([&] { callbacks.fetch_add(1, std::memory_order_relaxed); });
+
+    constexpr int N = 10'000;
+    std::thread producer([&] {
+        for (int i = 0; i < N; ++i) {
+            for (;;) {
+                try { ch.push(i); break; }
+                catch (const ChannelOverflowError&) { std::this_thread::yield(); }
+            }
+        }
+    });
+
+    for (int i = 0; i < N; ++i) (void)ch.pop();
+    producer.join();
+
+    // At least one transition, at most one per item; mainly we assert the run
+    // completed without TSan flagging a race on push_callback_/was_empty.
+    REQUIRE(callbacks.load() >= 1);
+    REQUIRE(callbacks.load() <= N);
+}
+
+// Ordering contract of the out-of-band sentinel under contention.
+//
+// push_sentinel() publishes has_eof_ (release) after the producer's N ring
+// pushes; a consumer that observes has_eof_ (acquire) therefore also observes
+// every value pushed before it. Both pop() and try_pop_now() only surface the
+// sentinel once the ring is *freshly* observed empty, so the sentinel is the
+// strictly last item received — it never jumps ahead of a ring value pushed
+// before it. These tests treat the sentinel as a hard "last message" barrier
+// (the consumer stops draining the moment it sees it) and assert that all N
+// values arrived, in a contiguous 0..N-1 sequence, before it.
+//
+// Regression guard: an earlier version of pop() checked emptiness against a
+// stale tail_ snapshot from the top of its loop, so under load the sentinel
+// could surface with a few real values still queued — breaking in_order /
+// values==N here. Under TSan these also cover the has_eof_/eof_value_
+// acquire/release handshake and the spin/futex wakeup on push_sentinel().
+
+TEST_CASE("SPSC: sentinel is strictly last, after every value (blocking pop)",
+          "[channel][stress]") {
+    constexpr int N        = 20'000;
+    constexpr int SENTINEL = -1;
+
+    for (int rep = 0; rep < kReps; ++rep) {
+        // Small ring + tiny spin window so the ring is frequently empty exactly
+        // when the sentinel is published — the interleaving under test.
+        Channel<int> ch(/*capacity=*/4, /*spin_count=*/8);
+
+        std::thread producer([&] {
+            for (int i = 0; i < N; ++i) {
+                for (;;) {
+                    try { ch.push(i); break; }
+                    catch (const ChannelOverflowError&) { std::this_thread::yield(); }
+                }
+            }
+            ch.push_sentinel(SENTINEL);   // must-deliver, never overflows/blocks
+        });
+
+        int  expected = 0;
+        bool in_order = true;
+        bool saw_sentinel = false;
+        // Treat the sentinel as EOF: stop draining the instant it appears.
+        for (;;) {
+            int v = ch.pop();
+            if (v == SENTINEL) { saw_sentinel = true; break; }
+            if (v != expected) in_order = false;
+            ++expected;
+        }
+        producer.join();
+
+        REQUIRE(saw_sentinel);
+        REQUIRE(in_order);
+        REQUIRE(expected == N);        // all N values received before the sentinel
+        REQUIRE(ch.size() == 0);
+        REQUIRE(ch.approx_size() == 0);
+    }
+}
+
+TEST_CASE("SPSC: sentinel is strictly last, after every value (try_pop_now)",
+          "[channel][stress]") {
+    // The pool-node consume path is try_pop_now(), not pop(): it must surface
+    // the out-of-band sentinel only once the ring is freshly observed empty.
+    // The consumer spins with no sleeps, racing the producer at full tilt
+    // across the empty-ring boundary where take_sentinel() is reached.
+    constexpr int N        = 20'000;
+    constexpr int SENTINEL = -1;
+
+    for (int rep = 0; rep < kReps; ++rep) {
+        Channel<int> ch(/*capacity=*/4, /*spin_count=*/0);
+
+        std::thread producer([&] {
+            for (int i = 0; i < N; ++i) {
+                for (;;) {
+                    try { ch.push(i); break; }
+                    catch (const ChannelOverflowError&) { std::this_thread::yield(); }
+                }
+            }
+            ch.push_sentinel(SENTINEL);
+        });
+
+        int  expected = 0;
+        bool in_order = true;
+        bool saw_sentinel = false;
+        int  v;
+        for (;;) {
+            if (!ch.try_pop_now(v)) { std::this_thread::yield(); continue; }
+            if (v == SENTINEL) { saw_sentinel = true; break; }
+            if (v != expected) in_order = false;
+            ++expected;
+        }
+        producer.join();
+
+        REQUIRE(saw_sentinel);
+        REQUIRE(in_order);
+        REQUIRE(expected == N);
+        // Sentinel held no ring slot; once taken the channel is fully empty.
+        REQUIRE(ch.size() == 0);
+        REQUIRE(ch.approx_size() == 0);
+    }
+}