From 8943e37e3ecbfff803ef2e3e582f811b5dedf632 Mon Sep 17 00:00:00 2001 From: AmiradelBeyg Date: Fri, 14 Aug 2026 16:31:46 -0400 Subject: [PATCH 1/2] Deduplicate reconnect scheduling in Producer/Consumer/Reader processes Every ChannelDisconnected event enqueued a new CloseChannel + EstablishNewChannel action. When a second disconnected event arrived while a reconnect was already in flight (e.g. a failed in-flight send racing the connection-level disconnect), the stale action would later close the freshly re-established channel, deregistering a healthy producer/consumer from the broker and forcing another reconnect cycle. ScheduleReconnect now runs at most one reconnect action at a time and re-evaluates the channel state after the action completes, so a disconnect arriving mid-reconnect is never lost and a replacement channel is never closed by a stale action. --- .../Internal/Abstractions/Process.cs | 23 ++ src/DotPulsar/Internal/ConsumerProcess.cs | 6 +- src/DotPulsar/Internal/ProducerProcess.cs | 6 +- src/DotPulsar/Internal/ReaderProcess.cs | 6 +- .../Internal/ProcessReconnectTests.cs | 298 ++++++++++++++++++ 5 files changed, 324 insertions(+), 15 deletions(-) create mode 100644 tests/DotPulsar.Tests/Internal/ProcessReconnectTests.cs diff --git a/src/DotPulsar/Internal/Abstractions/Process.cs b/src/DotPulsar/Internal/Abstractions/Process.cs index 0f6ed44ae..495d00efb 100644 --- a/src/DotPulsar/Internal/Abstractions/Process.cs +++ b/src/DotPulsar/Internal/Abstractions/Process.cs @@ -19,6 +19,7 @@ namespace DotPulsar.Internal.Abstractions; public abstract class Process : IProcess { private readonly CancellationTokenSource _cancellationTokenSource; + private int _isReconnecting; protected readonly AsyncQueue> ActionQueue; private Task? _actionProcessorTask; protected ChannelState ChannelState; @@ -89,6 +90,28 @@ public void Handle(IEvent e) protected abstract void CalculateState(); + protected void ScheduleReconnect(IContainsChannel channel) + { + if (Interlocked.CompareExchange(ref _isReconnecting, 1, 0) != 0) + return; + + ActionQueue.Enqueue(async cancellationToken => + { + try + { + await channel.CloseChannel(cancellationToken).ConfigureAwait(false); + await channel.EstablishNewChannel(cancellationToken).ConfigureAwait(false); + } + finally + { + Interlocked.Exchange(ref _isReconnecting, 0); + + if (ChannelState is ChannelState.ClosedByServer or ChannelState.Disconnected) + CalculateState(); + } + }); + } + private async Task ProcessActions(CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) diff --git a/src/DotPulsar/Internal/ConsumerProcess.cs b/src/DotPulsar/Internal/ConsumerProcess.cs index 18fc913d2..346751fbf 100644 --- a/src/DotPulsar/Internal/ConsumerProcess.cs +++ b/src/DotPulsar/Internal/ConsumerProcess.cs @@ -63,11 +63,7 @@ protected override void CalculateState() case ChannelState.ClosedByServer: case ChannelState.Disconnected: _stateManager.SetState(ConsumerState.Disconnected); - ActionQueue.Enqueue(async x => - { - await _subConsumer.CloseChannel(x).ConfigureAwait(false); - await _subConsumer.EstablishNewChannel(x).ConfigureAwait(false); - }); + ScheduleReconnect(_subConsumer); return; case ChannelState.Connected: if (!_isFailoverSubscription) diff --git a/src/DotPulsar/Internal/ProducerProcess.cs b/src/DotPulsar/Internal/ProducerProcess.cs index d6f407207..ce57eb0f1 100644 --- a/src/DotPulsar/Internal/ProducerProcess.cs +++ b/src/DotPulsar/Internal/ProducerProcess.cs @@ -56,11 +56,7 @@ protected override void CalculateState() case ChannelState.ClosedByServer: case ChannelState.Disconnected: _stateManager.SetState(ProducerState.Disconnected); - ActionQueue.Enqueue(async x => - { - await _subProducer.CloseChannel(x).ConfigureAwait(false); - await _subProducer.EstablishNewChannel(x).ConfigureAwait(false); - }); + ScheduleReconnect(_subProducer); return; case ChannelState.Connected: ActionQueue.Enqueue(x => diff --git a/src/DotPulsar/Internal/ReaderProcess.cs b/src/DotPulsar/Internal/ReaderProcess.cs index 9456dc232..e7fe8c00c 100644 --- a/src/DotPulsar/Internal/ReaderProcess.cs +++ b/src/DotPulsar/Internal/ReaderProcess.cs @@ -54,11 +54,7 @@ protected override void CalculateState() case ChannelState.ClosedByServer: case ChannelState.Disconnected: _stateManager.SetState(ReaderState.Disconnected); - ActionQueue.Enqueue(async x => - { - await _subReader.CloseChannel(x).ConfigureAwait(false); - await _subReader.EstablishNewChannel(x).ConfigureAwait(false); - }); + ScheduleReconnect(_subReader); return; case ChannelState.Connected: _stateManager.SetState(ReaderState.Connected); diff --git a/tests/DotPulsar.Tests/Internal/ProcessReconnectTests.cs b/tests/DotPulsar.Tests/Internal/ProcessReconnectTests.cs new file mode 100644 index 000000000..d1e16fae7 --- /dev/null +++ b/tests/DotPulsar.Tests/Internal/ProcessReconnectTests.cs @@ -0,0 +1,298 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace DotPulsar.Tests.Internal; + +using DotPulsar.Internal; +using DotPulsar.Internal.Abstractions; +using DotPulsar.Internal.Events; + +[Trait("Category", "Unit")] +public sealed class ProcessReconnectTests +{ + [Theory] + [InlineData(ProcessKind.Producer)] + [InlineData(ProcessKind.Consumer)] + [InlineData(ProcessKind.Reader)] + public async Task Start_WhenInitiallyDisconnected_EstablishesInitialChannel(ProcessKind processKind) + { + //Arrange + var correlationId = Guid.NewGuid(); + var channel = new TrackingChannelContainer(); + await using var harness = CreateHarness(processKind, correlationId, channel); + channel.OnEstablished = _ => harness.Process.Handle(new ChannelConnected(correlationId)); + + //Act + harness.Process.Start(); + await channel.WaitForEstablishAsync(Current.CancellationToken); + await harness.WaitForConnected(Current.CancellationToken); + + //Assert + channel.CurrentGeneration.ShouldBe(1); + channel.EstablishCount.ShouldBe(1); + } + + [Theory] + [InlineData(ProcessKind.Producer)] + [InlineData(ProcessKind.Consumer)] + [InlineData(ProcessKind.Reader)] + public async Task Handle_WhenChannelDisconnectedTwice_DoesNotCloseReplacementChannel(ProcessKind processKind) + { + //Arrange + var correlationId = Guid.NewGuid(); + var channel = new TrackingChannelContainer(); + await using var harness = CreateHarness(processKind, correlationId, channel); + channel.OnEstablished = _ => harness.Process.Handle(new ChannelConnected(correlationId)); + + harness.Process.Start(); + await channel.WaitForEstablishAsync(Current.CancellationToken); + await harness.WaitForConnected(Current.CancellationToken); + + //Act + var closeBlocker = channel.BlockNextClose(); + harness.Process.Handle(new ChannelDisconnected(correlationId)); + await closeBlocker.WaitForCloseStarted(Current.CancellationToken); + harness.Process.Handle(new ChannelDisconnected(correlationId)); + closeBlocker.Release(); + + await channel.WaitForEstablishAsync(Current.CancellationToken); + var replacementGeneration = channel.CurrentGeneration; + var replacementClosed = await Task.WhenAny( + channel.ReplacementClosed, + Task.Delay(TimeSpan.FromMilliseconds(250), Current.CancellationToken)) == channel.ReplacementClosed; + + //Assert + replacementGeneration.ShouldBe(2); + replacementClosed.ShouldBeFalse(); + channel.EstablishCount.ShouldBe(2); + } + + [Theory] + [InlineData(ProcessKind.Producer)] + [InlineData(ProcessKind.Consumer)] + [InlineData(ProcessKind.Reader)] + public async Task Handle_WhenReplacementDisconnectsDuringReconnect_ReconnectsAgain(ProcessKind processKind) + { + //Arrange + var correlationId = Guid.NewGuid(); + var channel = new TrackingChannelContainer(); + await using var harness = CreateHarness(processKind, correlationId, channel); + channel.OnEstablished = generation => + { + harness.Process.Handle(new ChannelConnected(correlationId)); + if (generation == 2) + harness.Process.Handle(new ChannelDisconnected(correlationId)); + }; + + harness.Process.Start(); + await channel.WaitForEstablishAsync(Current.CancellationToken); + await harness.WaitForConnected(Current.CancellationToken); + + //Act + harness.Process.Handle(new ChannelDisconnected(correlationId)); + + await channel.WaitForEstablishAsync(Current.CancellationToken); + await channel.WaitForEstablishAsync(Current.CancellationToken); + + //Assert + channel.CurrentGeneration.ShouldBe(3); + channel.EstablishCount.ShouldBe(3); + } + + private static ProcessHarness CreateHarness( + ProcessKind processKind, + Guid correlationId, + TrackingChannelContainer channel) + { + switch (processKind) + { + case ProcessKind.Producer: + { + var stateManager = new StateManager( + ProducerState.Disconnected, + ProducerState.Closed, + ProducerState.Faulted, + ProducerState.Fenced); + return new ProcessHarness( + new ProducerProcess(correlationId, stateManager, channel), + channel, + cancellationToken => stateManager + .OnStateChangeTo(ProducerState.Connected, cancellationToken) + .AsTask()); + } + case ProcessKind.Consumer: + { + var stateManager = new StateManager( + ConsumerState.Disconnected, + ConsumerState.Closed, + ConsumerState.ReachedEndOfTopic, + ConsumerState.Faulted); + return new ProcessHarness( + new ConsumerProcess(correlationId, stateManager, channel, false), + channel, + cancellationToken => stateManager + .OnStateChangeTo(ConsumerState.Active, cancellationToken) + .AsTask()); + } + case ProcessKind.Reader: + { + var stateManager = new StateManager( + ReaderState.Disconnected, + ReaderState.Closed, + ReaderState.ReachedEndOfTopic, + ReaderState.Faulted); + return new ProcessHarness( + new ReaderProcess(correlationId, stateManager, channel), + channel, + cancellationToken => stateManager + .OnStateChangeTo(ReaderState.Connected, cancellationToken) + .AsTask()); + } + default: + throw new ArgumentOutOfRangeException(nameof(processKind), processKind, null); + } + } + + public enum ProcessKind + { + Producer, + Consumer, + Reader + } + + private sealed class ProcessHarness : IAsyncDisposable + { + private readonly TrackingChannelContainer _channel; + private readonly Func _waitForConnected; + + public ProcessHarness( + IProcess process, + TrackingChannelContainer channel, + Func waitForConnected) + { + Process = process; + _channel = channel; + _waitForConnected = waitForConnected; + } + + public IProcess Process { get; } + + public Task WaitForConnected(CancellationToken cancellationToken) + => _waitForConnected(cancellationToken); + + public async ValueTask DisposeAsync() + { + await Process.DisposeAsync(); + await _channel.DisposeAsync(); + } + } + + private sealed class TrackingChannelContainer : IContainsChannel + { + private readonly Lock _lock = new(); + private readonly SemaphoreSlim _established = new(0); + private readonly TaskCompletionSource _replacementClosed = new(TaskCreationOptions.RunContinuationsAsynchronously); + private CloseBlocker? _nextCloseBlocker; + private int _currentGeneration; + private int _establishCount; + + public Action? OnEstablished { get; set; } + + public int CurrentGeneration + { + get + { + lock (_lock) + return _currentGeneration; + } + } + + public int EstablishCount + { + get + { + lock (_lock) + return _establishCount; + } + } + + public Task ReplacementClosed => _replacementClosed.Task; + + public Task EstablishNewChannel(CancellationToken cancellationToken) + { + int generation; + lock (_lock) + { + _currentGeneration++; + _establishCount++; + generation = _currentGeneration; + } + + _established.Release(); + OnEstablished?.Invoke(generation); + return Task.CompletedTask; + } + + public async ValueTask CloseChannel(CancellationToken cancellationToken) + { + CloseBlocker? closeBlocker; + int generation; + lock (_lock) + { + generation = _currentGeneration; + closeBlocker = _nextCloseBlocker; + _nextCloseBlocker = null; + } + + if (closeBlocker is not null) + { + closeBlocker.CloseStarted.TrySetResult(); + await closeBlocker.Continue.Task.WaitAsync(cancellationToken); + } + + if (generation == 2) + _replacementClosed.TrySetResult(); + } + + public ValueTask ChannelFaulted(Exception exception) => ValueTask.CompletedTask; + + public ValueTask DisposeAsync() + { + _established.Dispose(); + return ValueTask.CompletedTask; + } + + public async Task WaitForEstablishAsync(CancellationToken cancellationToken) + => await _established.WaitAsync(cancellationToken); + + public CloseBlocker BlockNextClose() + { + var closeBlocker = new CloseBlocker(); + lock (_lock) + _nextCloseBlocker = closeBlocker; + return closeBlocker; + } + } + + private sealed class CloseBlocker + { + public TaskCompletionSource CloseStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource Continue { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public async Task WaitForCloseStarted(CancellationToken cancellationToken) + => await CloseStarted.Task.WaitAsync(cancellationToken); + + public void Release() => Continue.TrySetResult(); + } +} From ae6f38734349f2bd39074e1043b9bc76f8eb0d9a Mon Sep 17 00:00:00 2001 From: AmiradelBeyg Date: Fri, 14 Aug 2026 16:35:13 -0400 Subject: [PATCH 2/2] Synchronize Process state transitions to prevent lost reconnects Handle is invoked synchronously on the caller's thread, so events race: ChannelConnected arrives from the connection response continuation while ChannelDisconnected arrives from the message dispatcher when an in-flight send fails. The switch writes ChannelState based on the event, but CalculateState reads the shared field afterwards - the two steps are not atomic. A stale Connected write can land between a Disconnected write and its CalculateState call, making both threads read Connected: the disconnect is handled but no reconnect is ever scheduled. Since the dispatcher stops itself after reporting the failure and the channel is already deregistered, no further event arrives and the producer stays disconnected forever while its state reports Connected. Guard the state update and the resulting decision with a per-process lock so every event's decision is made against that event's own state: every handled disconnect now schedules a reconnect. The lock is on the lifecycle-event path only (a handful of events per channel transition) and never executes on the message send path. --- .../Internal/Abstractions/Process.cs | 81 +++++++++++-------- .../Internal/ProcessReconnectTests.cs | 46 +++++++++++ 2 files changed, 92 insertions(+), 35 deletions(-) diff --git a/src/DotPulsar/Internal/Abstractions/Process.cs b/src/DotPulsar/Internal/Abstractions/Process.cs index 495d00efb..587639be7 100644 --- a/src/DotPulsar/Internal/Abstractions/Process.cs +++ b/src/DotPulsar/Internal/Abstractions/Process.cs @@ -19,6 +19,7 @@ namespace DotPulsar.Internal.Abstractions; public abstract class Process : IProcess { private readonly CancellationTokenSource _cancellationTokenSource; + private readonly object _stateLock = new object(); private int _isReconnecting; protected readonly AsyncQueue> ActionQueue; private Task? _actionProcessorTask; @@ -40,7 +41,11 @@ protected Process(Guid correlationId) public void Start() { _actionProcessorTask = ProcessActions(_cancellationTokenSource.Token); - CalculateState(); + + lock (_stateLock) + { + CalculateState(); + } } public virtual async ValueTask DisposeAsync() @@ -52,40 +57,43 @@ public virtual async ValueTask DisposeAsync() public void Handle(IEvent e) { - switch (e) + lock (_stateLock) { - case ExecutorFaulted executorFaulted: - ExecutorState = ExecutorState.Faulted; - Exception = executorFaulted.Exception; - break; - case ChannelActivated _: - ChannelState = ChannelState.Active; - break; - case ChannelClosedByServer _: - ChannelState = ChannelState.ClosedByServer; - break; - case ChannelConnected _: - ChannelState = ChannelState.Connected; - break; - case ChannelDeactivated _: - ChannelState = ChannelState.Inactive; - break; - case SendReceiptWrongOrdering _: - case ChannelDisconnected _: - ChannelState = ChannelState.Disconnected; - break; - case ChannelReachedEndOfTopic _: - ChannelState = ChannelState.ReachedEndOfTopic; - break; - case ChannelUnsubscribed _: - ChannelState = ChannelState.Unsubscribed; - break; - case ProducerWaitingForExclusive _: - ChannelState = ChannelState.WaitingForExclusive; - break; - } + switch (e) + { + case ExecutorFaulted executorFaulted: + ExecutorState = ExecutorState.Faulted; + Exception = executorFaulted.Exception; + break; + case ChannelActivated _: + ChannelState = ChannelState.Active; + break; + case ChannelClosedByServer _: + ChannelState = ChannelState.ClosedByServer; + break; + case ChannelConnected _: + ChannelState = ChannelState.Connected; + break; + case ChannelDeactivated _: + ChannelState = ChannelState.Inactive; + break; + case SendReceiptWrongOrdering _: + case ChannelDisconnected _: + ChannelState = ChannelState.Disconnected; + break; + case ChannelReachedEndOfTopic _: + ChannelState = ChannelState.ReachedEndOfTopic; + break; + case ChannelUnsubscribed _: + ChannelState = ChannelState.Unsubscribed; + break; + case ProducerWaitingForExclusive _: + ChannelState = ChannelState.WaitingForExclusive; + break; + } - CalculateState(); + CalculateState(); + } } protected abstract void CalculateState(); @@ -106,8 +114,11 @@ protected void ScheduleReconnect(IContainsChannel channel) { Interlocked.Exchange(ref _isReconnecting, 0); - if (ChannelState is ChannelState.ClosedByServer or ChannelState.Disconnected) - CalculateState(); + lock (_stateLock) + { + if (ChannelState is ChannelState.ClosedByServer or ChannelState.Disconnected) + CalculateState(); + } } }); } diff --git a/tests/DotPulsar.Tests/Internal/ProcessReconnectTests.cs b/tests/DotPulsar.Tests/Internal/ProcessReconnectTests.cs index d1e16fae7..8a04cb2ca 100644 --- a/tests/DotPulsar.Tests/Internal/ProcessReconnectTests.cs +++ b/tests/DotPulsar.Tests/Internal/ProcessReconnectTests.cs @@ -110,6 +110,52 @@ public async Task Handle_WhenReplacementDisconnectsDuringReconnect_ReconnectsAga channel.EstablishCount.ShouldBe(3); } + [Theory] + [InlineData(ProcessKind.Producer)] + [InlineData(ProcessKind.Consumer)] + [InlineData(ProcessKind.Reader)] + public async Task Handle_WhenEventsArriveConcurrently_ReconnectsAfterFinalDisconnect(ProcessKind processKind) + { + //Arrange + var correlationId = Guid.NewGuid(); + var channel = new TrackingChannelContainer(); + await using var harness = CreateHarness(processKind, correlationId, channel); + channel.OnEstablished = _ => harness.Process.Handle(new ChannelConnected(correlationId)); + + harness.Process.Start(); + await channel.WaitForEstablishAsync(Current.CancellationToken); + await harness.WaitForConnected(Current.CancellationToken); + + //Act + Parallel.For(0, 500, _ => + { + harness.Process.Handle(new ChannelDisconnected(correlationId)); + harness.Process.Handle(new ChannelConnected(correlationId)); + }); + + await WaitForEstablishQuiescenceAsync(channel, Current.CancellationToken); + var establishCountBeforeFinalDisconnect = channel.EstablishCount; + + harness.Process.Handle(new ChannelDisconnected(correlationId)); + + while (channel.EstablishCount <= establishCountBeforeFinalDisconnect) + await channel.WaitForEstablishAsync(Current.CancellationToken); + + //Assert + channel.EstablishCount.ShouldBeGreaterThan(establishCountBeforeFinalDisconnect); + } + + private static async Task WaitForEstablishQuiescenceAsync(TrackingChannelContainer channel, CancellationToken cancellationToken) + { + while (true) + { + var count = channel.EstablishCount; + await Task.Delay(TimeSpan.FromMilliseconds(250), cancellationToken); + if (channel.EstablishCount == count) + return; + } + } + private static ProcessHarness CreateHarness( ProcessKind processKind, Guid correlationId,