diff --git a/src/DotPulsar/Internal/Abstractions/Process.cs b/src/DotPulsar/Internal/Abstractions/Process.cs index 0f6ed44ae..587639be7 100644 --- a/src/DotPulsar/Internal/Abstractions/Process.cs +++ b/src/DotPulsar/Internal/Abstractions/Process.cs @@ -19,6 +19,8 @@ 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; protected ChannelState ChannelState; @@ -39,7 +41,11 @@ protected Process(Guid correlationId) public void Start() { _actionProcessorTask = ProcessActions(_cancellationTokenSource.Token); - CalculateState(); + + lock (_stateLock) + { + CalculateState(); + } } public virtual async ValueTask DisposeAsync() @@ -51,44 +57,72 @@ 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(); + 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); + + lock (_stateLock) + { + 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..8a04cb2ca --- /dev/null +++ b/tests/DotPulsar.Tests/Internal/ProcessReconnectTests.cs @@ -0,0 +1,344 @@ +/* + * 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); + } + + [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, + 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(); + } +}