Skip to content

Commit ae6f387

Browse files
AmiradelBeygAmiradelBeyg
committed
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.
1 parent 8943e37 commit ae6f387

2 files changed

Lines changed: 92 additions & 35 deletions

File tree

‎src/DotPulsar/Internal/Abstractions/Process.cs‎

Lines changed: 46 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ namespace DotPulsar.Internal.Abstractions;
1919
public abstract class Process : IProcess
2020
{
2121
private readonly CancellationTokenSource _cancellationTokenSource;
22+
private readonly object _stateLock = new object();
2223
private int _isReconnecting;
2324
protected readonly AsyncQueue<Func<CancellationToken, Task>> ActionQueue;
2425
private Task? _actionProcessorTask;
@@ -40,7 +41,11 @@ protected Process(Guid correlationId)
4041
public void Start()
4142
{
4243
_actionProcessorTask = ProcessActions(_cancellationTokenSource.Token);
43-
CalculateState();
44+
45+
lock (_stateLock)
46+
{
47+
CalculateState();
48+
}
4449
}
4550

4651
public virtual async ValueTask DisposeAsync()
@@ -52,40 +57,43 @@ public virtual async ValueTask DisposeAsync()
5257

5358
public void Handle(IEvent e)
5459
{
55-
switch (e)
60+
lock (_stateLock)
5661
{
57-
case ExecutorFaulted executorFaulted:
58-
ExecutorState = ExecutorState.Faulted;
59-
Exception = executorFaulted.Exception;
60-
break;
61-
case ChannelActivated _:
62-
ChannelState = ChannelState.Active;
63-
break;
64-
case ChannelClosedByServer _:
65-
ChannelState = ChannelState.ClosedByServer;
66-
break;
67-
case ChannelConnected _:
68-
ChannelState = ChannelState.Connected;
69-
break;
70-
case ChannelDeactivated _:
71-
ChannelState = ChannelState.Inactive;
72-
break;
73-
case SendReceiptWrongOrdering _:
74-
case ChannelDisconnected _:
75-
ChannelState = ChannelState.Disconnected;
76-
break;
77-
case ChannelReachedEndOfTopic _:
78-
ChannelState = ChannelState.ReachedEndOfTopic;
79-
break;
80-
case ChannelUnsubscribed _:
81-
ChannelState = ChannelState.Unsubscribed;
82-
break;
83-
case ProducerWaitingForExclusive _:
84-
ChannelState = ChannelState.WaitingForExclusive;
85-
break;
86-
}
62+
switch (e)
63+
{
64+
case ExecutorFaulted executorFaulted:
65+
ExecutorState = ExecutorState.Faulted;
66+
Exception = executorFaulted.Exception;
67+
break;
68+
case ChannelActivated _:
69+
ChannelState = ChannelState.Active;
70+
break;
71+
case ChannelClosedByServer _:
72+
ChannelState = ChannelState.ClosedByServer;
73+
break;
74+
case ChannelConnected _:
75+
ChannelState = ChannelState.Connected;
76+
break;
77+
case ChannelDeactivated _:
78+
ChannelState = ChannelState.Inactive;
79+
break;
80+
case SendReceiptWrongOrdering _:
81+
case ChannelDisconnected _:
82+
ChannelState = ChannelState.Disconnected;
83+
break;
84+
case ChannelReachedEndOfTopic _:
85+
ChannelState = ChannelState.ReachedEndOfTopic;
86+
break;
87+
case ChannelUnsubscribed _:
88+
ChannelState = ChannelState.Unsubscribed;
89+
break;
90+
case ProducerWaitingForExclusive _:
91+
ChannelState = ChannelState.WaitingForExclusive;
92+
break;
93+
}
8794

88-
CalculateState();
95+
CalculateState();
96+
}
8997
}
9098

9199
protected abstract void CalculateState();
@@ -106,8 +114,11 @@ protected void ScheduleReconnect(IContainsChannel channel)
106114
{
107115
Interlocked.Exchange(ref _isReconnecting, 0);
108116

109-
if (ChannelState is ChannelState.ClosedByServer or ChannelState.Disconnected)
110-
CalculateState();
117+
lock (_stateLock)
118+
{
119+
if (ChannelState is ChannelState.ClosedByServer or ChannelState.Disconnected)
120+
CalculateState();
121+
}
111122
}
112123
});
113124
}

‎tests/DotPulsar.Tests/Internal/ProcessReconnectTests.cs‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,52 @@ public async Task Handle_WhenReplacementDisconnectsDuringReconnect_ReconnectsAga
110110
channel.EstablishCount.ShouldBe(3);
111111
}
112112

113+
[Theory]
114+
[InlineData(ProcessKind.Producer)]
115+
[InlineData(ProcessKind.Consumer)]
116+
[InlineData(ProcessKind.Reader)]
117+
public async Task Handle_WhenEventsArriveConcurrently_ReconnectsAfterFinalDisconnect(ProcessKind processKind)
118+
{
119+
//Arrange
120+
var correlationId = Guid.NewGuid();
121+
var channel = new TrackingChannelContainer();
122+
await using var harness = CreateHarness(processKind, correlationId, channel);
123+
channel.OnEstablished = _ => harness.Process.Handle(new ChannelConnected(correlationId));
124+
125+
harness.Process.Start();
126+
await channel.WaitForEstablishAsync(Current.CancellationToken);
127+
await harness.WaitForConnected(Current.CancellationToken);
128+
129+
//Act
130+
Parallel.For(0, 500, _ =>
131+
{
132+
harness.Process.Handle(new ChannelDisconnected(correlationId));
133+
harness.Process.Handle(new ChannelConnected(correlationId));
134+
});
135+
136+
await WaitForEstablishQuiescenceAsync(channel, Current.CancellationToken);
137+
var establishCountBeforeFinalDisconnect = channel.EstablishCount;
138+
139+
harness.Process.Handle(new ChannelDisconnected(correlationId));
140+
141+
while (channel.EstablishCount <= establishCountBeforeFinalDisconnect)
142+
await channel.WaitForEstablishAsync(Current.CancellationToken);
143+
144+
//Assert
145+
channel.EstablishCount.ShouldBeGreaterThan(establishCountBeforeFinalDisconnect);
146+
}
147+
148+
private static async Task WaitForEstablishQuiescenceAsync(TrackingChannelContainer channel, CancellationToken cancellationToken)
149+
{
150+
while (true)
151+
{
152+
var count = channel.EstablishCount;
153+
await Task.Delay(TimeSpan.FromMilliseconds(250), cancellationToken);
154+
if (channel.EstablishCount == count)
155+
return;
156+
}
157+
}
158+
113159
private static ProcessHarness CreateHarness(
114160
ProcessKind processKind,
115161
Guid correlationId,

0 commit comments

Comments
 (0)