attempting to fix racy Akka.Persistence.TestKit.Tests

src/core/Akka.Persistence.TestKit.Tests/Actors/CounterActor.cs

@@ -72,6 +72,16 @@ protected override void OnRecover(object message)
                     return;
             }
         }
+
+        protected override void PostStop()
+        {
+            _log.Info("Shutting down");
+        }
+
+        protected override void PreStart()
+        {
+            _log.Info("Starting up");
+        }
     }
 
     public class CounterActorTests : PersistenceTestKit
@@ -86,9 +96,10 @@ public Task CounterActor_internal_state_will_be_lost_if_underlying_persistence_s
                 var counterProps = Props.Create(() => new CounterActor("test"));
                 var actor = ActorOf(counterProps, "counter");
 
-                Watch(actor);
+                Sys.Log.Info("Messaging actor");
+                await WatchAsync(actor);
                 actor.Tell("inc", TestActor);
-                await ExpectMsgAsync<Terminated>(TimeSpan.FromSeconds(3));
+                await ExpectTerminatedAsync(actor);
 
                 // need to restart actor
                 actor = ActorOf(counterProps, "counter1");

src/core/Akka.Persistence.TestKit.Tests/Bug4762FixSpec.cs

@@ -80,7 +80,6 @@ public Task TestJournal_PersistAll_should_only_count_each_event_exceptions_once(
             return WithJournalWrite(write => write.Pass(), async () =>
             {
                 var actor = ActorOf(() => new TestActor2(probe));
-                Watch(actor);
 
                 var command = new WriteMessage();
                 actor.Tell(command, actor);

src/core/Akka.Persistence.TestKit/Journal/JournalInterceptors.cs

@@ -16,7 +16,7 @@ internal class Noop : IJournalInterceptor
         {
             public static readonly IJournalInterceptor Instance = new Noop();
 
-            public Task InterceptAsync(IPersistentRepresentation message) => Task.FromResult(true);
+            public Task InterceptAsync(IPersistentRepresentation message) => Task.CompletedTask;
         }
 
         internal class Failure : IJournalInterceptor

src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs

@@ -259,20 +259,11 @@ private void HandleReplayMessages(ReplayMessages message)
 
             async Task ExecuteHighestSequenceNr()
             {
-                void CompleteHighSeqNo(long highSeqNo)
-                {
-                    replyTo.Tell(new RecoverySuccess(highSeqNo));
-
-                    if (CanPublish)
-                    {
-                        eventStream.Publish(message);
-                    }
-                }
-
                 try
                 {
                     var highSequenceNr = await _breaker.WithCircuitBreaker((message, readHighestSequenceNrFrom, awj: this), state =>
-                        state.awj.ReadHighestSequenceNrAsync(state.message.PersistenceId, state.readHighestSequenceNrFrom));
+                        state.awj.ReadHighestSequenceNrAsync(state.message.PersistenceId, state.readHighestSequenceNrFrom))
+                        .ConfigureAwait(false);
                     var toSequenceNr = Math.Min(message.ToSequenceNr, highSequenceNr);
                     if (toSequenceNr <= 0L || message.FromSequenceNr > toSequenceNr)
                     {
@@ -293,7 +284,7 @@ await ReplayMessagesAsync(context, message.PersistenceId, message.FromSequenceNr
                                         replyTo.Tell(new ReplayedMessage(adaptedRepresentation), ActorRefs.NoSender);
                                     }
                                 }
-                            });
+                            }).ConfigureAwait(false);
 
                         CompleteHighSeqNo(highSequenceNr);
                     }
@@ -309,6 +300,18 @@ await ReplayMessagesAsync(context, message.PersistenceId, message.FromSequenceNr
                 {
                     replyTo.Tell(new ReplayMessagesFailure(TryUnwrapException(ex)));
                 }
+
+                return;
+
+                void CompleteHighSeqNo(long highSeqNo)
+                {
+                    replyTo.Tell(new RecoverySuccess(highSeqNo));
+
+                    if (CanPublish)
+                    {
+                        eventStream.Publish(message);
+                    }
+                }
             }
 
             // instead of ContinueWith

src/core/Akka.Persistence/Journal/MemoryJournal.cs

@@ -137,7 +137,7 @@ public override Task ReplayMessagesAsync(IActorContext context, string persisten
             var highest = HighestSequenceNr(persistenceId);
             if (highest != 0L && max != 0L)
                 Read(persistenceId, fromSequenceNr, Math.Min(toSequenceNr, highest), max).ForEach(recoveryCallback);
-            return Task.FromResult(new object());
+            return Task.CompletedTask;
         }
 
         /// <summary>
@@ -154,7 +154,7 @@ protected override Task DeleteMessagesToAsync(string persistenceId, long toSeque
                 _meta.AddOrUpdate(persistenceId, highestSeqNr, (_, _) => highestSeqNr);
             for (var snr = 1L; snr <= toSeqNr; snr++)
                 Delete(persistenceId, snr);
-            return Task.FromResult(new object());
+            return Task.CompletedTask;
         }
 
         protected override bool ReceivePluginInternal(object message)