SinkWorker: one queue per channel, sized by the pool

CLAUDE.md

@@ -27,15 +27,15 @@ under "Unreleased" in `data_tamer_cpp/CHANGELOG.rst`.
     `data_tamer.hpp` (ChannelsRegistry), `data_sink.hpp` (DataSink, SinkWorker, Snapshot,
     SnapshotRef), `logged_value.hpp`, `names.hpp`, `types.hpp` (Schema, BasicType),
     `custom_types.hpp`, `values.hpp`, `fwd.hpp`, `contrib/SerializeMe.hpp`
-    (TypeDefinitionTrait, serialization), `details/` (pool, write mutex, shared state:
+    (TypeDefinitionTrait, serialization), `details/` (pool, write mutex, spin pause, shared state:
     public because templates use them), `sinks/`.
   - `include/data_tamer_parser/data_tamer_parser.hpp`: standalone C++ decoder, no
     dependency on the library.
   - `src/`: implementation; `src/sinks/` for MCAPSink, MCAPRingSink, ROS2PublisherSink.
   - `tests/`: one gtest binary, `datatamer_test`, plus compile-only targets.
   - `examples/`: T01 to T04, `mcap_1m_per_sec`, `mcap_reader`, `ros2_publisher` (ROS only).
   - `benchmarks/`: built only when Google Benchmark is found.
-  - `3rdparty/`: vendored moodycamel ConcurrentQueue and MCAP.
+  - `3rdparty/`: vendored MCAP.
 - `data_tamer_msgs/`: ROS 2 messages `Schema`, `Schemas`, `Snapshot`, `SnapshotBatch`.
 - `python/`: `data_tamer_parser.py`, the reference decoder (standard library only),
   packaged as `data-tamer-parser`; its tests; `ros2_subscriber.py`.
@@ -103,23 +103,34 @@ DATA_TAMER_UPDATE_GOLDEN=1 \
   -Werror`.
 - The real-time path allocates nothing, takes no blocking lock, does no I/O and does not
   throw. It covers `tryTakeSnapshot()` and what it reaches (`SnapshotPool::tryAcquire`,
-  `WriteMutex::tryLockWithSpin`, `ValuePtr` serialization, `SinkWorker::tryPush`),
+  `WriteMutex::tryLockWithSpin`, `ValuePtr` serialization, `SinkWorker::tryPush` and
+  its per-channel single-producer queue),
   scalar `LoggedValue::set()`/`get()`, `trySetEnabled()` and
   `MCAPRingSink::requestDump()`. Tests assert it with `AllocCounter::Scope`
   (`tests/alloc_counter.hpp`): add one when you touch these paths. `takeSnapshot()` may
   block and grow a slot; keep the two variants distinct.
-- `onSchema()`/`onSnapshot()` are serialized by the SinkWorker, and control operations
-  (registration, sinks, `prepare()`) wait for them. Never call a control operation from a
-  sink callback, a serializer or inside `scopedWrite()`.
+- `onSchema()`/`onSnapshot()`/`onStop()`/`onStart()` are serialized by the SinkWorker,
+  and control operations (registration, sinks, `prepare()`) wait for them. Never call a
+  control operation from a sink callback, a serializer or inside `scopedWrite()`.
+  `stop()` runs `onStop()` once per stop, after the last delivery, and `start()` after a
+  stop runs `onStart()`: a sink that needs to finish (close, flush, dump) or reopen does
+  it there.
 - A change to the schema text, the payload encoding, the schema hash or the MCAP and ROS
   message layout is a format revision. Update `docs/wire_format.md`, regenerate the
   vectors, update `python/data_tamer_parser.py` and `data_tamer_parser.hpp`, and bump
   `SCHEMA_VERSION` or `SCHEMA_YAML_VERSION` in `types.hpp` when the text changes. The
   `WireFormat.*` tests and the Python tests fail until all of them agree.
 - LogChannel, ChannelsRegistry, SinkWorker, TypesRegistry, MCAPSink, MCAPRingSink and
-  ROS2PublisherSink keep their state behind a Pimpl. Add members to the Pimpl.
-  `tests/abi_tests.cpp` pins the sizes of the sinks, SinkWorker and SnapshotRef. Record
-  any ABI break in the CHANGELOG.
+  ROS2PublisherSink keep all their state behind a Pimpl (LogChannel's only other base is
+  `enable_shared_from_this`). Add members to the Pimpl. `tests/abi_tests.cpp` pins the
+  sizes of the sinks, SinkWorker, LogChannel, SnapshotRef, DataSink and ChannelDefaults.
+  Record any ABI break in the CHANGELOG.
+- The vtables of `DataSink` and `CustomSerializer` are frozen for 2.x: users derive from
+  them and the library calls through vtables compiled into their binaries. Add no virtual
+  function and no data member; new behaviour goes into non-virtual functions or a
+  separate interface.
+- Only `SnapshotPool::adopt()` builds a SnapshotRef from a pool slot; the constructor is
+  private.
 - `MCAPRingStats` never crosses the library boundary: `MCAPRingSink::stats()` is inline
   and fills it from one exported getter per field. A new field gets a new getter;
   existing getters stay.
@@ -137,7 +148,8 @@ DATA_TAMER_UPDATE_GOLDEN=1 \
   `--all-files`, which rewrites files unrelated to your change.
 - Tests: gtest, `TEST(Suite, BehaviourInCamelCase)`, new files listed in
   `DATATAMER_TEST_SOURCES` (`tests/CMakeLists.txt`). Helpers live in
-  `tests/test_sinks.hpp` (`Attached<T>`). Wait for delivery with `SinkWorker::drain()` or
+  `tests/test_sinks.hpp` (`Attached<T>`, `channelWith`, `acceptedUntilExhausted`) and
+  `tests/mcap_test_utils.hpp` (`tempPath`, `countMessages`). Wait for delivery with `SinkWorker::drain()` or
   `Delivery::Manual`, never with sleeps.
 - Commits: short imperative subject, optionally prefixed by the component
   (`MCAPSink: roll over into new files by default`). The body says why, and what was

README.md

@@ -35,8 +35,10 @@ a sink may save the information immediately in a file (currently, we support [MC
 or publish it using an inter-process communication, for instance, a ROS2 publisher.
 
 You can easily create your own, specialized sinks: implement `DataTamer::DataSink`
-(two callbacks, `onSchema` and `onSnapshot`) and wrap it with `SinkWorker::create<MySink>()`,
-which owns the delivery queue and thread. See `data_tamer/sinks/dummy_sink.hpp` for a small one.
+(two callbacks, `onSchema` and `onSnapshot`, plus optional `onStop` and `onStart` to finish at
+shutdown and reopen on restart)
+and wrap it with `SinkWorker::create<MySink>()`, which owns the delivery thread and the queues
+of the channels attached to it. See `data_tamer/sinks/dummy_sink.hpp` for a small one.
 
 Headers that only name DataTamer types (a `LogChannel&` parameter, a `std::shared_ptr<SinkWorker>`
 member, ...) can include `data_tamer/fwd.hpp` instead of the full headers. Like `<iosfwd>`, it
@@ -82,8 +84,12 @@ on a real-time thread.
 
 - One thread per channel calls `takeSnapshot()`. `prepare()` freezes the schema, pre-allocates
   a pool of 64 snapshots and announces the schema to the sinks; the first `takeSnapshot()` with
-  sinks attached calls it for you. Tune beforehand with `setPoolCapacity()` and
+  sinks attached calls it for you. Tune beforehand with `setPoolCapacity()`, in slots or in
+  time (`setPoolCapacity(200ms, 1ms)` absorbs a 200 ms sink stall at 1 kHz), and
   `setPayloadCapacity()`.
+- The pool is the only bound on snapshots in flight. Each sink gets one queue per channel,
+  as long as the pool, so a snapshot is refused (`partial`, `rejected`) only by a stopped
+  `SinkWorker`. There is no queue size to configure.
 - `takeSnapshot()` returns a `SnapshotResult` (`ok`, `partial`, `rejected`, `no_sinks`,
   `pool_exhausted`, ...). It may wait for a writer holding the mutex and grows a slot whose
   payload no longer fits. `tryTakeSnapshot()` is the real-time variant: the library does no
@@ -185,6 +191,29 @@ Non-scalar `set()`/`get()` lock automatically and can be called inside
 them. Non-scalar pointer proxies also hold the write mutex; release them before
 starting a transaction or taking a snapshot.
 
+## Registry defaults and shutdown
+
+`ChannelsRegistry` creates channels by name, applies default settings to each new channel
+and attaches the default sinks to every channel, including the ones that exist already.
+`SinkWorker::stop()` delivers what is queued and then lets the sink finish itself
+(`DataSink::onStop()`): `MCAPSink` closes its file, `MCAPRingSink` writes a pending dump,
+`ROS2PublisherSink` publishes its partial batch. `stopAll()` does that for every sink of the
+registry, so shutdown needs no ordering rules. `SinkWorker::start()` resumes a stopped sink
+(`DataSink::onStart()`: `MCAPSink` continues in its next numbered file).
+
+```cpp
+using namespace std::chrono_literals;
+auto& registry = DataTamer::ChannelsRegistry::Global();
+DataTamer::ChannelDefaults defaults;
+defaults.pool_stall_tolerance = 200ms;  // pool sized to absorb a 200 ms sink stall
+defaults.pool_snapshot_period = 1ms;    // at 1 kHz: 200 slots per channel
+registry.setChannelDefaults(defaults);  // channels created from now on
+registry.addDefaultSink(DataTamer::MCAPSink::create("run.mcap"));
+auto channel = registry.getChannel("controller");
+// ... register values, prepare(), take snapshots ...
+registry.stopAll();  // every sink stopped once; run.mcap is complete
+```
+
 ## How to register custom types
 
 Containers such as `std::vector` and `std::array` are supported out of the box.
@@ -314,7 +343,7 @@ options.schema_format = DataTamer::SchemaFormat::Yaml;  // optional: shorter sch
 
 auto sink = DataTamer::ROS2PublisherSink::create(node, "/robot", options);
 channel->addDataSink(sink);
-// a partial batch is published by the next snapshot, by flush() or when the sink is destroyed:
+// a partial batch is published by the next snapshot, by flush() or by sink->stop():
 sink->as<DataTamer::ROS2PublisherSink>().flush();
 ```
 
@@ -354,9 +383,8 @@ auto& recorder = worker->as<DataTamer::MCAPRingSink>();
 // From any thread, real-time ones included: one atomic compare-exchange.
 recorder.requestDump(std::chrono::seconds(2));  // also keep 2 s after the event
 
-// At shutdown, write a dump requested just before it.
+// At shutdown, stop() delivers what is queued and writes a dump requested just before it.
 worker->stop();
-recorder.flushPendingDump();
 ```
 
 - The trigger is the first snapshot delivered after the request, and all times are snapshot