Fixed possible racing conditions in Journal and SnapshotStore #2735
Conversation
| .ContinueWith(t => | ||
| { | ||
| if (!t.IsFaulted && !t.IsCanceled) | ||
| if (t.IsCompleted) |
There was a problem hiding this comment.
!t.IsFaulted && !t.IsCanceled is the same as t.IsCompleted
| @@ -253,43 +261,42 @@ private void HandleReplayMessages(ReplayMessages message) | |||
| // Send replayed messages and replay result to persistentActor directly. No need | |||
| // to resequence replayed messages relative to written and looped messages. | |||
| // not possible to use circuit breaker here | |||
| ReplayMessagesAsync(context, message.PersistenceId, message.FromSequenceNr, toSequenceNr, | |||
| message.Max, p => | |||
| ReplayMessagesAsync(context, message.PersistenceId, message.FromSequenceNr, toSequenceNr, message.Max, p => | |||
There was a problem hiding this comment.
Do we need the parameter context?
There was a problem hiding this comment.
If the context is being handled asynchronously, then we do need to close over it... however, the context is volatile - so I'm not sure why we'd want to snapshot it. What data are we using from it?
There was a problem hiding this comment.
@cconstantin @Horusiath As I see - we dont use it in our persistence plugins
| { | ||
| var msg = (LoadSnapshot) message; | ||
| var senderPersistentActor = Sender; // Sender is PersistentActor | ||
| var self = Self; //Self MUST BE CLOSED OVER here, or the code below will be subject to race conditions |
There was a problem hiding this comment.
These variables should be closed over here, in order to prevent racing conditions
| .ContinueWith(t => | ||
| { | ||
| if (_publish) eventStream.Publish(message); | ||
| if (t.IsCompleted && _publish) |
There was a problem hiding this comment.
Added an additional check
| { | ||
| return (!t.IsFaulted && !t.IsCanceled) | ||
| ? (object) new DeleteMessagesSuccess(message.ToSequenceNr) | ||
| _breaker.WithCircuitBreaker(() => DeleteMessagesToAsync(message.PersistenceId, message.ToSequenceNr)) |
There was a problem hiding this comment.
Circuit Breaker was added here
There was a problem hiding this comment.
Looks good to me, same as JVM source
| { | ||
| return (!t.IsFaulted && !t.IsCanceled) | ||
| ? (object) new DeleteMessagesSuccess(message.ToSequenceNr) | ||
| _breaker.WithCircuitBreaker(() => DeleteMessagesToAsync(message.PersistenceId, message.ToSequenceNr)) |
There was a problem hiding this comment.
Looks good to me, same as JVM source
| @@ -253,43 +261,42 @@ private void HandleReplayMessages(ReplayMessages message) | |||
| // Send replayed messages and replay result to persistentActor directly. No need | |||
| // to resequence replayed messages relative to written and looped messages. | |||
| // not possible to use circuit breaker here | |||
| ReplayMessagesAsync(context, message.PersistenceId, message.FromSequenceNr, toSequenceNr, | |||
| message.Max, p => | |||
| ReplayMessagesAsync(context, message.PersistenceId, message.FromSequenceNr, toSequenceNr, message.Max, p => | |||
There was a problem hiding this comment.
If the context is being handled asynchronously, then we do need to close over it... however, the context is volatile - so I'm not sure why we'd want to snapshot it. What data are we using from it?
| .PipeTo(replyTo) | ||
| .ContinueWith(t => | ||
| { | ||
| if (!t.IsFaulted && CanPublish) context.System.EventStream.Publish(message); | ||
| if (t.IsCompleted && CanPublish) eventStream.Publish(message); |
There was a problem hiding this comment.
Good idea to close over the EventStream 👍
No description provided.