test: Coyote systematic concurrency tests (18 bugs across Core, Kafka, MQTT)

[msallin]

Summary

Adds a Silverback.Tests.Concurrency project that uses Microsoft Coyote to systematically explore task interleavings in Silverback's concurrency-sensitive code. Coyote binary-rewrites Silverback.Core, Silverback.Integration, Silverback.Integration.Kafka, and Silverback.Integration.MQTT, then runs each test across ~100 scheduling strategies to find races deterministically.

No production fixes. All bugs are documented as failing tests.

Running locally requires:

dotnet tool install --global Microsoft.Coyote.CLI --version 1.7.11
dotnet test tests/Silverback.Tests.Concurrency

The MSBuild target sets DOTNET_ROLL_FORWARD=Major because the CLI ships as net8.0.

Test results

Core / Integration

Finding	Test	Status	Bug
—	ConcurrentAbort_ShouldReachAbortedStateExactlyOnce	Passing	Abort idempotency is correct
—	ConcurrentErrorAborts_ShouldSetAbortExceptionExactlyOnce	Passing	Error-abort idempotency is correct
AbortIfIncompleteAsync leak	AbortThenAbortIfIncomplete_ShouldBothReturnAndRemainAborted	Failing	Semaphore leak in AbortIfIncompleteAsync
AbortIfIncompleteAsync leak	ConcurrentAbortMix_ShouldConvergeToSingleAbortedState	Failing	Same leak, mixed abort paths
AbortIfIncompleteAsync leak	SequenceBaseSemaphoreLeakTests (plain xunit)	Failing	Deterministic reproducer, no Coyote needed
H1	SequenceStore_EnumerateWhileAdding_ShouldNotThrow	Failing	Dictionary enumerator version check
C2	Abort_RacingCompleteAsync_ShouldNotReturnAsSilentNoOp	Failing	Abort throws after Complete finishes
#1	LazyStream_GetOrCreateRacingCancel_ShouldNotDoubleSetTcs	Failing	TCS double-set (SetResult + SetCanceled race)
#2	ConsumerChannel_StartReadingDisposingCts_ShouldNotThrow	Failing	ObjectDisposedException on cached token
#4	InMemoryKafkaOffsetStore_EnumerateReturnedCollection	Failing	Live dict.Values escapes lock scope
#5	ParallelSelect_SelectorThrows_ShouldNotLeakSemaphoreToken	Failing	Semaphore leaked when selector throws
H4	CreateStream_RacingPushAsync	Skipped	Coyote rewriter crashes on async enumerator
C1	CompleteCoreAsync_RacingAbortAsync	Skipped	Needs Coyote harness tuning

Kafka

Finding	Test	Status	Bug
C3	SemaphoreCurrentCountRelease_ShouldNotExceedMaxDoP	Failing	Peak active 3 > maxDoP 2
H5	CommitCoreAsync_RevokeRacingCommit_ShouldNotStoreOffset	Failing	Offset stored for revoked partition
H6	OnPartitionsRevoked_CommitBeforeStopCompletes	Failing	Messages consumed after commit
H7	ConsumeLoopHandlerStart_ConcurrentCalls	Passing	Non-volatile bool; not observable on x64
M2	OffsetsTracker_ConcurrentTrackAndRead	Failing	Reader sees partial dual-dict state

MQTT

Finding	Test	Status	Bug
MC1	NextChannelIndex_ConcurrentIncrements_ShouldDistributeEvenly	Failing	Non-atomic increment loses messages
MC2	PublishQueueChannel_RecreationDuringDrain_ShouldNotLoseMessages	Failing	Channel recreated while drainer holds old ref
MC3	PendingReconnect_ConcurrentTryConnect_ShouldReconnectExactlyOnce	Passing	Non-volatile bool; not observable on x64
MC4	ChannelReset_ConcurrentWithWrite_ShouldNotLoseMessages	Passing	Not caught in 100 iterations
MC5	MqttConsumedMessage_TcsReassignment_ShouldNotLoseSignal	Failing	Commit signal sent to stale TCS; channel reader blocks forever

Totals: 15 failing, 5 passing, 2 skipped (22 Coyote tests + 1 plain xunit).

Bugs found

AbortIfIncompleteAsync semaphore leak

Loc: SequenceBase`1.cs:249-265

AbortIfIncompleteAsync acquires _completeSemaphoreSlim then early-returns on !IsPending without hitting the finally that releases it. Dispose() then deadlocks on _completeSemaphoreSlim.Wait(). Introduced in 3db53b5.

Fix: move the !IsPending check inside the try/finally so the finally always releases.

H1 — SequenceStore unsynchronized Dictionary

Loc: SequenceStore.cs:19

Plain Dictionary with zero locking. Cross-thread callers race on Add/Remove/Enumerate.

Fix: replace Dictionary with ConcurrentDictionary or wrap all access in a lock.

C2 — MessageStreamProvider Abort/Complete race

Loc: MessageStreamProvider`1.cs:104-175

Abort() and CompleteAsync() share an early-return guard using plain bools. When Abort runs after Complete finishes, it throws InvalidOperationException.

Fix: replace the re-entry guard with a proper wait-then-observe pattern.

#1 — LazyMessageStreamEnumerable TaskCompletionSource double-set

Loc: LazyMessageStreamEnumerable`1.cs:49-59

GetOrCreateStream() calls _taskCompletionSource.SetResult(stream). Cancel() calls _taskCompletionSource.SetCanceled(). No synchronization. The second Set throws InvalidOperationException in a fire-and-forget context, silently lost.

Fix: use TrySetResult / TrySetCanceled instead of Set, and add a lock or Interlocked flag around the create-vs-cancel decision.

#2 — ConsumerChannel CancellationTokenSource dispose-while-active

Loc: ConsumerChannel`1.cs:86-102

StartReading() disposes and recreates _readCancellationTokenSource without synchronization. Concurrent readers that cached the old token get ObjectDisposedException.

Fix: don't dispose the old CTS immediately; let it be GC'd, or swap under a lock and ensure readers always re-fetch the token.

#4 — InMemoryKafkaOffsetStore live collection escapes lock scope

Loc: InMemoryKafkaOffsetStore.cs:20-28

GetStoredOffsets() returns offsetsByTopicPartition.Values (a live Dictionary.ValueCollection view) after releasing the lock. StoreOffsetsAsync mutates the dictionary under lock, but the caller enumerates the live view outside the lock.

Fix: return a snapshot: return [.. offsetsByTopicPartition.Values]; inside the lock.

#5 — EnumerableSelectExtensions semaphore leak on throw

Loc: EnumerableSelectExtensions.cs:42-61

ParallelSelectAsync releases the semaphore in the try block, not the finally. When the selector throws, the catch cancels the CTS but does NOT release the semaphore. The token is permanently leaked.

Fix: move semaphore.Release() to a finally block:

await semaphore.WaitAsync(cancellationTokenSource.Token).ConfigureAwait(false);
try { result = await selector(value).ConfigureAwait(false); }
finally { semaphore.Release(); }

C3 — ConsumerChannelsManager semaphore ownership (reported starvation root cause)

Loc: ConsumerChannelsManager.cs:150-165

Conditionally acquires _messagesLimiterSemaphoreSlim based on _channels.Count > MaxDegreeOfParallelism, then releases based on CurrentCount < MaxDegreeOfParallelism. After a partition revoke drops channel count, a handler skips the acquire but releases in its finally. Over-releases a token it never owned.

Fix: capture bool acquired locally and release iff true.

H5 — Kafka commit TOCTOU on revoked partitions

Loc: KafkaConsumer.cs:251-279

CommitCoreAsync double-checks IsNotRevoked but _revokedPartitions is mutated concurrently by the rebalance handler.

Fix: snapshot the revoked set before the loop.

H6 — OnPartitionsRevoked commits after fire-and-forget stop

Loc: KafkaConsumer.cs:152-174

StopAsync().FireAndForget() does not await; Client.Commit() runs before the consume loop actually stops.

Fix: await StopAsync() before calling Client.Commit().

M2 — OffsetsTracker dual AddOrUpdate non-atomicity

Loc: OffsetsTracker.cs:27-42

TrackOffset performs two sequential AddOrUpdate on _commitOffsets and _rollbackOffsets. A concurrent reader sees partial state.

Fix: take a lock around both updates, or merge into a single structure.

MC1 — MQTT _nextChannelIndex non-atomic increment

Loc: ConsumerChannelsManager.cs:84 (MQTT)

OnMessageReceivedAsync uses _channels[_nextChannelIndex++]. MQTTnet fires callbacks on arbitrary threads.

Fix: use Interlocked.Increment(ref _nextChannelIndex) % _channels.Length.

MC2 — MQTT publish queue channel recreation race

Loc: MqttClientWrapper.cs:113-114

ConnectCoreAsync recreates _publishQueueChannel while ProcessPublishQueueAsync still drains the old one.

Fix: signal the drainer to switch channels, or use a long-lived channel.

MC5 — MQTT ConsumedApplicationMessage TCS reassignment race

Loc: ConsumerChannelsManager.cs:73 + MqttConsumer.cs:164,171

After awaiting the TCS, the channel reader reassigns a new one. A concurrent CommitCoreAsync/RollbackCoreAsync calls SetResult on the OLD reference. The new TCS is never signaled; the channel reader blocks forever.

Fix: use a thread-safe signal mechanism (e.g. a Channel or AsyncAutoResetEvent) instead of reassigning a TCS field.

Test plan

• dotnet build Silverback.sln -- no errors
• dotnet test tests/Silverback.Tests.Concurrency -- 15 failing, 5 passing, 2 skipped
• dotnet test tests/Silverback.Integration.Tests --filter SequenceBaseSemaphoreLeakTests -- 1 failing
• Review each bug description and decide fix priority

[chatgpt-codex-connector]

You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard.