fix(subscriptions): emit commit diagnostic on the worker thread to stop a _positions data race#552
Open
alexeyzimarev wants to merge 1 commit into
Open
fix(subscriptions): emit commit diagnostic on the worker thread to stop a _positions data race#552alexeyzimarev wants to merge 1 commit into
alexeyzimarev wants to merge 1 commit into
Conversation
…op 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 <noreply@anthropic.com>
Contributor
PR Summary by QodoFix checkpoint commit diagnostics to avoid cross-thread _positions race
AI Description
Diagram
High-Level Assessment
Files changed (2)
|
Contributor
Code Review by Qodo
1. Blocking GetResult() in test
|
![qodo-free-for-open-source-projects[bot]](https://avatars.githubusercontent.com/in/1068000?s=60&v=4){kind=small}
Comment on lines
+51
to
+54
| var caller = new Thread(() => { | ||
| callerThreadId = Environment.CurrentManagedThreadId; | ||
| handler.Commit(new CommitPosition(0, 0, DateTime.UtcNow), ct).AsTask().GetAwaiter().GetResult(); | ||
| }) { IsBackground = true }; |
Contributor
There was a problem hiding this comment.
1. Blocking getresult() in test 📘 Rule violation ☼ Reliability
The new concurrency test blocks on handler.Commit(...).AsTask().GetAwaiter().GetResult() which violates the repo guidance to avoid blocking waits in async flows. Even though it’s test code, this can hide deadlocks/hangs and conflicts with the async/.NoContext() convention.
Agent Prompt
## Issue description
`CheckpointCommitHandlerConcurrencyTests` blocks on an async operation via `GetAwaiter().GetResult()`, which conflicts with the repository’s async-first convention.
## Issue Context
This code is in a test, but blocking waits can still create brittle hangs/deadlock-like behavior and violates the compliance guidance to avoid sync-over-async.
## Fix Focus Areas
- src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerConcurrencyTests.cs[51-54]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
+72
to
+74
| if (Diagnostic.IsEnabled(CommitOperation)) | ||
| Diagnostic.Write(CommitOperation, new CommitEvent(_subscriptionId, _positions.Max, _positions.Min)); | ||
|
|
Contributor
There was a problem hiding this comment.
2. Metrics can go stale 🐞 Bug ◔ Observability
SubscriptionMetrics derives gap/time and pending-checkpoints from CommitEvent diagnostics, but diagnostics now emit only from the single commit-worker Process loop. If the worker is blocked awaiting CommitInternal (e.g., slow checkpoint store), ACKs can keep calling Commit and enqueueing positions while no new CommitEvent is emitted, causing metrics to lag/freeze and misrepresent backlog and processing lag.
Agent Prompt
### Issue description
`SubscriptionMetrics` uses `CheckpointCommitMetrics`, which only updates on `CheckpointCommitHandler`'s `CommitOperation` diagnostic events. This PR moved `Diagnostic.Write("Commit", ...)` from `Commit()` (ACK caller thread) to the worker-thread `Process()` loop. Because the worker is single-threaded and awaits checkpoint-store I/O in `CommitInternal`, diagnostics (and thus metrics) can stop updating while ACKs continue enqueueing positions.
### Issue Context
This change fixes a real data race, but it also changes telemetry semantics: metrics now reflect *worker ingestion/progress*, not *ACK/processing progress*, and can appear frozen during slow checkpoint commits.
### Fix Focus Areas
- src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs[64-96]
- src/Core/src/Eventuous.Subscriptions/Diagnostics/CheckpointCommitMetrics.cs[12-37]
- src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionMetrics.cs[44-63]
### Suggested fix approach
Choose one (or combine):
1) **Keep ACK-path diagnostic emission without `_positions` access**
- Have the worker maintain a thread-safe snapshot (e.g., boxed `CommitPosition?` or an atomic `ulong firstPendingSequence`) updated inside `Process()` / after `RemoveWhere`.
- In `Commit()`, emit the diagnostic with `CommitPosition = position` and `FirstPending = snapshot` (no `SortedSet` reads on caller thread).
2) **Split diagnostics**
- Emit a lightweight “ACK received” diagnostic from `Commit()` (no `_positions`), and a separate “pending range” diagnostic from `Process()`.
- Update `CheckpointCommitMetrics` to use the ACK diagnostic for gap/time, and the worker diagnostic for pending-checkpoints.
3) **If the new semantics are intended**, update metric descriptions/documentation so `gap` and `time lag` explicitly reflect commit-worker progress rather than last processed/acked event.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
CheckpointCommitHandler.Commitbuilds the commit diagnostic on the ACK caller's thread and reads_positions.Minthere:_positions(CommitPositionSequence : SortedSet<CommitPosition>) is not thread-safe and is otherwise confined to the single commit-worker thread (Process→UnionWith/RemoveWhere). The caller-thread_positions.Minread therefore races the worker's mutations.SortedSet.Minwalks.Leftpointers thatRemoveWheretransiently nulls during red-black-tree rebalancing, so the read can throwNullReferenceExceptionout 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), i.e. when the app callsAddEventuousSubscriptions()— which is why it hit a production app but never the test suite. #540 guarded the nullCheckpointCommitHandleron this path but not this cross-thread_positionsaccess.Reproduced in isolation: one thread doing
UnionWith+RemoveWhereon aSortedSetwhile others read.MinthrowsNullReferenceExceptionwithin ~1s. (Insert-onlyUnionWithdoes not — it's the removal-driven rebalancing nearMinthat does it, matching the worker's commit-trim.)Fix
Move the commit-diagnostic emission from
Commit(ack caller thread) into the worker'sProcessloop, where_positionsis owned and read safely. It readsMax(latest received ≈ "last processed", used by the gap metric) andMin(FirstPending, used by the pending-checkpoints gauge).Commitno longer touches_positions.Telemetry is preserved — both
SubscriptionMetricsconsumers (checkpoint queue length/ pending gauge viaRecord(), andGapCount/GapTimeviaGetLastCommitPosition/GetLastTimestamp) are still fed equivalent values. Emission is now per processed batch on the worker, so it also fires during a commit stall, keeping the backlog gauge fresh.Test
Adds
CheckpointCommitHandlerConcurrencyTests: a deterministic regression test asserting the commit diagnostic is not emitted on the caller thread.Commitis invoked from a dedicatedThread(whoseManagedThreadIdis never a thread-pool id, so the comparison can't collide under load). RED before the fix (emitted synchronously inCommit), GREEN after (emitted on the worker). The fullEventuous.Tests.Subscriptionssuite (32 tests) stays green.🤖 Generated with Claude Code