Skip to content

Fix reconnect races that can leave producers permanently disconnected #293

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
blankensteiner merged 2 commits into from
Aug 22, 2026
+414 −48
Merged

Fix reconnect races that can leave producers permanently disconnected #293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8943e37
Deduplicate reconnect scheduling in Producer/Consumer/Reader processes
AmiradelBeyg Aug 14, 2026
ae6f387
Synchronize Process state transitions to prevent lost reconnects
AmiradelBeyg Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 67 additions & 33 deletions src/DotPulsar/Internal/Abstractions/Process.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Func<CancellationToken, Task>> ActionQueue;
private Task? _actionProcessorTask;
protected ChannelState ChannelState;
Expand All @@ -39,7 +41,11 @@ protected Process(Guid correlationId)
public void Start()
{
_actionProcessorTask = ProcessActions(_cancellationTokenSource.Token);
CalculateState();

lock (_stateLock)
{
CalculateState();
}
}

public virtual async ValueTask DisposeAsync()
Expand All @@ -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)
Expand Down
6 changes: 1 addition & 5 deletions src/DotPulsar/Internal/ConsumerProcess.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
6 changes: 1 addition & 5 deletions src/DotPulsar/Internal/ProducerProcess.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =>
Expand Down
6 changes: 1 addition & 5 deletions src/DotPulsar/Internal/ReaderProcess.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Loading
Loading