From 001967679f1256d01805614f3cd9983c8bf5adcb Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Sun, 12 Jul 2026 11:10:07 +0200 Subject: [PATCH] fix(subscriptions): emit commit diagnostic on the worker thread to stop a _positions data race CheckpointCommitHandler.Commit built the commit diagnostic on the ack caller's thread and read _positions.Min there, while the commit worker thread mutates the same non-thread-safe SortedSet (UnionWith/RemoveWhere). SortedSet.Min walks .Left pointers that RemoveWhere transiently nulls during red-black-tree rebalancing, so the read can throw NullReferenceException out of the ack path -> Dropped(SubscriptionError) -> resubscribe that never resumes committing -> the subscription wedges with the checkpoint frozen and no self-recovery. The race only surfaces when a subscription diagnostic listener is attached (Diagnostic.IsEnabled("Commit") is true), which is why it hit an app using AddEventuousSubscriptions() but never the test suite. #540 guarded the null CheckpointCommitHandler but not this cross-thread _positions read. Move the diagnostic emission into the worker's Process loop, where _positions is owned and read safely, reading Max (last processed, for the gap metric) and Min (FirstPending, for the pending-checkpoints gauge). Commit no longer touches _positions. Telemetry is preserved. Adds a deterministic regression test asserting the commit diagnostic is not emitted on the caller thread (Commit invoked from a dedicated Thread so the thread-id comparison cannot collide with a pool thread under load). Co-Authored-By: Claude Opus 4.8 --- .../Checkpoints/CheckpointCommitHandler.cs | 13 +++- ...CheckpointCommitHandlerConcurrencyTests.cs | 65 +++++++++++++++++++ 2 files changed, 76 insertions(+), 2 deletions(-) create mode 100644 src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerConcurrencyTests.cs diff --git a/src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs b/src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs index 5c0dd38a3..b0ed302e9 100644 --- a/src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs +++ b/src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs @@ -61,8 +61,17 @@ public CheckpointCommitHandler( return; + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Diagnostics only")] async ValueTask Process(IReadOnlyList list, CancellationToken cancellationToken) { _positions.UnionWith(list); + + // Emit the commit diagnostic here, on the single worker thread that owns _positions. It used + // to be written from Commit() on the ACK caller thread, whose _positions.Min read raced this + // loop's UnionWith/RemoveWhere on the non-thread-safe SortedSet — a data race that threw + // NullReferenceException out of the ack path and wedged the subscription (AI-1329). + if (Diagnostic.IsEnabled(CommitOperation)) + Diagnostic.Write(CommitOperation, new CommitEvent(_subscriptionId, _positions.Max, _positions.Min)); + var next = GetCommitPosition(false); if (!next.Valid) return; @@ -78,9 +87,9 @@ async ValueTask Process(IReadOnlyList list, CancellationToken ca /// Cancellation token /// [PublicAPI] - [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Diagnostics only")] public ValueTask Commit(CommitPosition position, CancellationToken cancellationToken) { - if (Diagnostic.IsEnabled(CommitOperation)) Diagnostic.Write(CommitOperation, new CommitEvent(_subscriptionId, position, _positions.Min)); + // No _positions access here — that read moved to the worker thread's Process (AI-1329). This + // runs on the ack caller thread and must never touch the worker-owned, non-thread-safe set. position.LogContext?.PositionReceived(position); return _worker.Write(position, cancellationToken); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerConcurrencyTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerConcurrencyTests.cs new file mode 100644 index 000000000..eace80a28 --- /dev/null +++ b/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerConcurrencyTests.cs @@ -0,0 +1,65 @@ +using Eventuous.Diagnostics; +using Eventuous.Subscriptions.Checkpoints; +using Shouldly; + +namespace Eventuous.Tests.Subscriptions; + +/// +/// Guards against the production wedge (AI-1329) where +/// read the non-thread-safe _positions ON THE ACK CALLER'S THREAD +/// (to build the commit diagnostic's FirstPending) while the commit worker thread mutated it — +/// a data race that threw out of the ack path and dropped the +/// subscription. The commit diagnostic must be emitted from the worker thread, where _positions +/// is owned, so it can never race the caller. +/// +public class CheckpointCommitHandlerConcurrencyTests { + /// + /// Captures the managed thread id on which the commit diagnostic is emitted. Attaching any + /// listener with the commit-handler diagnostic name also flips Diagnostic.IsEnabled("Commit") + /// to true — exactly what AddEventuousSubscriptions() does in a real app, and why production + /// hit this while the test suite (no listener) never had. + /// + sealed class ThreadCapturingListener(string subscriptionId, TaskCompletionSource emittedOn) + : GenericListener(CheckpointCommitHandler.DiagnosticName), IDisposable { + // The commit DiagnosticListener is process-global, so filter to OUR handler's subscription id + // (CommitEvent.Id) — otherwise a concurrently-running test's handler could be captured. The + // event type is internal, so read Id reflectively. Capture the thread DiagnosticSource.Write + // ran on; it is synchronous, so this is the emitting thread. + protected override void OnEvent(KeyValuePair evt) { + var id = evt.Value?.GetType().GetProperty("Id")?.GetValue(evt.Value) as string; + + if (id == subscriptionId) emittedOn.TrySetResult(Environment.CurrentManagedThreadId); + } + } + + [Test] + public async Task Commit_diagnostic_is_not_emitted_on_the_caller_thread(CancellationToken ct) { + var subscriptionId = $"test-commit-thread-{Guid.NewGuid():N}"; + var emittedOn = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + using var listener = new ThreadCapturingListener(subscriptionId, emittedOn); + + var store = new NoOpCheckpointStore(); + + await using var handler = new CheckpointCommitHandler(subscriptionId, store, TimeSpan.FromMilliseconds(1), batchSize: 1); + + // Call Commit from a DEDICATED thread. Its ManagedThreadId is never a thread-pool id, and the + // worker runs on the pool — so "emitted != caller" is deterministic (comparing two pool threads + // could collide under load). Before the fix Commit emits synchronously on this dedicated thread; + // after it, emission happens on the worker. + var callerThreadId = 0; + var caller = new Thread(() => { + callerThreadId = Environment.CurrentManagedThreadId; + handler.Commit(new CommitPosition(0, 0, DateTime.UtcNow), ct).AsTask().GetAwaiter().GetResult(); + }) { IsBackground = true }; + caller.Start(); + caller.Join(); + + var emittedThreadId = await emittedOn.Task.WaitAsync(TimeSpan.FromSeconds(5), ct); + + emittedThreadId.ShouldNotBe( + callerThreadId, + "the commit diagnostic reads the worker-owned _positions, so it must be emitted on the worker thread, never the ack caller thread" + ); + } +}