Fix have_consumers staying true after an idle consumer disconnects (#267)

[daverlon]

Fixes #267

The transfer thread only noticed a disconnected peer after a failed write. Outlets that were not pushing never unregistered the queue. Watch the TCP socket for peer close so an idle outlet unregisters the consumer without waiting for a push.

[cboulay]

The idle-disconnect test passes locally, but I found a concurrency issue that needs addressing before merge.

In abort_on_peer_close(), the Asio callback calls q->push_sample(sample_p()) directly. consumer_queue::push_sample() explicitly requires a single producer. Normal sample pushes are serialized by send_buffer::consumers_mut_, but this callback bypasses that mutex, so disconnecting while the outlet is pushing can put two producers in the queue simultaneously. try_push() uses a load/store of the write index rather than a CAS, allowing both producers to claim the same slot and race on its sample pointer.

Please use a separate cancellation/wakeup mechanism that does not enqueue from a second producer, or serialize the wakeup with all normal pushes. The current idle-only regression does not cover this; please also test disconnects concurrent with active sample production, preferably under ThreadSanitizer.

Validation: built the PR on macOS and ran the new [outlet] test; all 4 assertions passed. The race above is a source-level finding, not a claim that the idle test reproduced it.

[cboulay]

Merged current dev (including #299 and #300) into this branch in 246d147. The conflict resolution preserves #300’s queue subscription before the handshake and starts this PR’s disconnect watcher using that existing queue after the header write completes. It also retains the updated synchronous socket handoff. This update resolves integration conflicts; it does not yet implement the recommendations below.

I support fixing idle disconnect detection: a marker outlet should stop reporting a consumer that has already disconnected, even if no samples are pushed afterward. The pending asynchronous read is a reasonable event-driven approach, but I recommend revising the cleanup before merging:

1. Serialize the wakeup with ordinary sample pushes. abort_on_peer_close() currently calls q->push_sample(sample_p()) directly from the IO callback. consumer_queue requires a single producer, while normal outlet pushes are serialized through send_buffer::consumers_mut_. This callback bypasses that lock and can race with a real sample push. A small targeted wakeup helper on send_buffer, using the same mutex and retaining the queue safely for the operation, would preserve the existing empty-sample wakeup convention without adding a concurrent producer. A dedicated queue cancellation operation is another option, but would be a broader change.
2. Account for synchronous outlets. Their sockets are owned by sync_write_handler and bypass this watcher. They still retain idle disconnected consumers until a write fails. This needs explicit coverage here or a clearly scoped follow-up with appropriate socket ownership/synchronization.
3. Preserve Ensure outlets are ready when open_stream returns #300’s early subscription and watcher cleanup. Moving subscription back after the handshake would reintroduce the first-sample race. Failed header writes already release the queue; the pending read must also be cancelled when the transfer thread exits.
4. Expand regression coverage to explicit close/reopen and disconnecting one of two consumers, plus synchronous idle disconnects when supported. Keep the immediate-first-sample tests from Ensure outlets are ready when open_stream returns #300.

#299 makes inlet shutdown deterministic, but the outlet still needs to observe the disconnect. Also, this fix removes stale consumers; it cannot guarantee that a consumer satisfying wait_for_consumers() is the intended recorder or will remain connected. That stronger experiment-start guarantee needs an application-level readiness exchange.

Validation of this merge: macOS universal build passed; [outlet],[open],[reopen],[sync] passed with 1,691 assertions across 17 test cases. These tests do not establish absence of the producer race above.

[daverlon]

Fixed. send_buffer::wake_consumer() does the empty-sample push under consumers_mut_, and
abort_on_peer_close() calls that instead of pushing to the queue directly, so the IO thread
is no longer a second producer.

Before the fix I was able to reproduce the race you described under ThreadSanitizer. It does reproduce once a test pushes concurrently with the disconnect on macOS arm64, AppleClang21, Debug, -fsanitize=thread:

WARNING: ThreadSanitizer: data race
  Read of size 8 at 0x00011061c828 by thread T1:
    #0  lslboost::intrusive_ptr<lsl::sample>::swap(...)
    #1  lslboost::intrusive_ptr<lsl::sample>::operator=(intrusive_ptr&&)
    #2  lsl::consumer_queue::copy_or_move(..., intrusive_ptr&&)
    #3  lsl::consumer_queue::try_push<intrusive_ptr<lsl::sample>>(...)
    #4  lsl::consumer_queue::push_sample<intrusive_ptr<lsl::sample>>(...)
    #5  lsl::client_session::abort_on_peer_close(...)::$_0::operator()(...)
    [... asio dispatch ...]
    #12 asio::io_context::run()
  Previous write of size 8 at 0x00011061c828 by thread T3 (mutexes: write M0):
    #0  lslboost::intrusive_ptr<lsl::sample>::swap(...)
    #1  lslboost::intrusive_ptr<lsl::sample>::operator=(intrusive_ptr const&)
    #2  lsl::consumer_queue::copy_or_move(..., intrusive_ptr const&)
    #3  lsl::consumer_queue::try_push<intrusive_ptr<lsl::sample> const&>(...)
    #4  lsl::consumer_queue::push_sample<intrusive_ptr<lsl::sample> const&>(...)
    #5  lsl::send_buffer::push_sample(intrusive_ptr<lsl::sample> const&)
    #6  lsl::stream_outlet_impl::enqueue<int>(...)
    #9  lsl::stream_outlet::push_sample(int const*, double, bool)
  Location is heap block of size 576016 allocated by thread T1:
    #1  lsl::consumer_queue::consumer_queue(unsigned long, shared_ptr<send_buffer>)
    #9  lsl::send_buffer::new_consumer(int)
    #10 lsl::client_session::handle_read_feedparams(...)
  Mutex M0 created at:
    #2  std::lock_guard<std::mutex>::lock_guard(std::mutex&)
    #3  lsl::send_buffer::push_sample(intrusive_ptr<lsl::sample> const&)

Both threads are in consumer_queue::try_push on the same ring buffer slot. TSan identifies
M0 as the lock taken in send_buffer::push_sample and annotates only T3 with it; the
abort_on_peer_close callback on T1 holds nothing.

To confirm the fix rather than just fail to see the race, I built the pre-fix commit and this
branch with identical flags and ran them interleaved with 550 runs each. Before the change there were 97 queue races in 550 runs, and after it 0 queue races. With the fix all 1100 runs passed, and I wrapped each run in a 120s kill in case the new lock deadlocked and none of them hit it.

Reproduce on 246d147:

cmake -S . -B build-tsan -DCMAKE_BUILD_TYPE=Debug -DLSL_UNITTESTS=ON \
  -DCMAKE_C_FLAGS="-fsanitize=thread -fno-omit-frame-pointer" \
  -DCMAKE_CXX_FLAGS="-fsanitize=thread -fno-omit-frame-pointer" \
  -DCMAKE_EXE_LINKER_FLAGS="-fsanitize=thread" \
  -DCMAKE_SHARED_LINKER_FLAGS="-fsanitize=thread"
cmake --build build-tsan --parallel
for i in $(seq 1 500); do
  TSAN_OPTIONS="history_size=7 halt_on_error=0" \
    ./build-tsan/testing/lsl_test_exported \
    "have_consumers becomes false after disconnect during push"
done

I initially tried a consumer_queue::cancel() but it needs a flag on consumer_queue that never resets, and a data_receiver uses that class too.

Added three new tests:

• disconnect during active production
• explicit close/reopen
• disconnecting one of two inlets.

[outlet],[open],[reopen],[sync] is 1710 assertions across
20 cases. The pre-handshake subscription and the sock_.cancel() on transfer-thread exit are
untouched.

Synchronous idle disconnects are in #303. I'm not planning to pick that one up, but tell me if you'd rather it land here.

TSan also flags a pre-existing factory::new_sample() / data_receiver::pull_sample_typed() race, 21 times pre-fix, 17 after. Not from this change, and it's how I know TSan was still live on the fixed build.

[cboulay]

The producer-race blocker is addressed: disconnect wakeups now use the same send-buffer mutex as normal pushes, retaining the queue for the operation. Early subscription and watcher cleanup are preserved. I also added exception-safe cleanup to the active-production regression test in 4a3e886.

Validation of 4a3e886:

• Universal macOS Debug build passed; the focused [outlet],[open],[reopen],[sync] suite passed 1,710 assertions in 20 cases on both arm64 and x86_64.
• Injected assertion and exception failures confirmed the test stops and joins its pusher during unwinding.
• With Xcode 27 / Apple Clang 21 on native arm64, the same focused suite passed under ThreadSanitizer without warnings.
• 100 repeated disconnect-during-push runs passed all functional assertions, with no reports of the outlet queue producer race. Fifteen runs reported the separate sample-recycling warning now tracked in #305 and consequently exited unsuccessfully under TSan; this is not a claim of a fully TSan-clean stress run.
• All 14 PR CI checks passed.

Synchronous idle disconnect detection remains scoped to #303. Ready to merge this asynchronous-outlet fix.