Examine Persistence Actor Dispatcher timings

src/core/Akka.Persistence.TCK.Tests/LocalSnapshotStoreSpec.cs

@@ -46,10 +46,15 @@ public void LocalSnapshotStore_can_snapshot_actors_with_PersistenceId_containing
             ExpectMsg<SaveSnapshotSuccess>();
 
             SnapshotStore.Tell(new LoadSnapshot(pid, SnapshotSelectionCriteria.Latest, long.MaxValue), TestActor);
-            ExpectMsg<LoadSnapshotResult>(res => 
-                res.Snapshot.Snapshot.Equals("sample data") 
-                && res.Snapshot.Metadata.PersistenceId == pid
-                && res.Snapshot.Metadata.SequenceNr == 1);
+            ExpectMsg<LoadSnapshotResult>(IsMessage);
+            bool IsMessage(LoadSnapshotResult res)
+            {
+                var result = res.Snapshot.Snapshot.Equals("sample data") 
+                       && res.Snapshot.Metadata.PersistenceId == pid
+                       && res.Snapshot.Metadata.SequenceNr == 1;
+
+                return result;
+            }
         }
     }
 }

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

@@ -5,6 +5,7 @@
 // </copyright>
 //-----------------------------------------------------------------------
 
+using Akka.Configuration;
 using Akka.Event;
 using Xunit.Abstractions;
 
@@ -72,23 +73,42 @@ 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
     {
-        public CounterActorTests(ITestOutputHelper output) : base(output:output){}
+        // create a Config that enables debug mode on the TestJournal
+        private static readonly Config Config =
+            ConfigurationFactory.ParseString("""
+                                             akka.persistence.journal.test.debug = on
+                                             akka.persistence.snapshot-store.test.debug = on
+                                             """);
+
+        public CounterActorTests(ITestOutputHelper output) : base(Config, output:output){}
 
         [Fact]
         public Task CounterActor_internal_state_will_be_lost_if_underlying_persistence_store_is_not_available()
         {
             return WithJournalWrite(write => write.Fail(), async () => 
             {
-                var counterProps = Props.Create(() => new CounterActor("test"));
+                var counterProps = Props.Create(() => new CounterActor("test"))
+                    .WithDispatcher("akka.actor.internal-dispatcher");
                 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

@@ -11,7 +11,9 @@
 using System.Text;
 using System.Threading.Tasks;
 using Akka.Actor;
+using Akka.Configuration;
 using Akka.Event;
+using FluentAssertions.Extensions;
 using Xunit;
 using Xunit.Abstractions;
 
@@ -22,16 +24,26 @@ namespace Akka.Persistence.TestKit.Tests
     /// </summary>
     public class Bug4762FixSpec : PersistenceTestKit
     {
-        public Bug4762FixSpec(ITestOutputHelper outputHelper) : base(output: outputHelper)
+        // create a Config that enables debug mode on the TestJournal
+        private static readonly Config Config =
+            ConfigurationFactory.ParseString("""
+                                             akka.loglevel = DEBUG
+                                             akka.persistence.journal.test.debug = on
+                                             akka.persistence.journal.test.replay-filter.debug = on
+                                             akka.persistence.snapshot-store.test.debug = on
+                                             """);
+
+        public Bug4762FixSpec(ITestOutputHelper outputHelper) : base(Config, output: outputHelper)
         {
-
         }
-        
+
         private class WriteMessage
-        { }
+        {
+        }
 
         private class TestEvent
-        { }
+        {
+        }
 
         private class TestActor2 : UntypedPersistentActor
         {
@@ -48,24 +60,27 @@ public TestActor2(IActorRef probe)
             protected override void OnCommand(object message)
             {
                 _log.Info("Received command {0}", message);
-                
+
                 switch (message)
                 {
                     case WriteMessage _:
                         var event1 = new TestEvent();
                         var event2 = new TestEvent();
                         var events = new List<TestEvent> { event1, event2 };
-                        PersistAll(events, _ =>
-                        {
-                            _probe.Tell(Done.Instance);
-                        });
+                        PersistAll(events, _ => { _probe.Tell(Done.Instance); });
                         break;
 
                     default:
                         return;
                 }
             }
 
+            protected override void PreStart()
+            {
+                _log.Info("Starting up and beginning recovery");
+                base.PreStart();
+            }
+
             protected override void OnRecover(object message)
             {
                 _log.Info("Received recover {0}", message);
@@ -79,17 +94,18 @@ public Task TestJournal_PersistAll_should_only_count_each_event_exceptions_once(
             var probe = CreateTestProbe();
             return WithJournalWrite(write => write.Pass(), async () =>
             {
-                var actor = ActorOf(() => new TestActor2(probe));
-                Watch(actor);
+                var actor = Sys.ActorOf(
+                    Props.Create(() => new TestActor2(probe))
+                        .WithDispatcher("akka.actor.internal-dispatcher"), "test-actor");
 
                 var command = new WriteMessage();
                 actor.Tell(command, actor);
 
-                await probe.ExpectMsgAsync<RecoveryCompleted>();
+                await probe.ExpectMsgAsync<RecoveryCompleted>(10.Seconds());
                 await probe.ExpectMsgAsync<Done>();
                 await probe.ExpectMsgAsync<Done>();
                 await probe.ExpectNoMsgAsync(3000);
             });
         }
     }
-}
+}

src/core/Akka.Persistence.TestKit.Xunit2/PersistenceTestKit.cs

@@ -7,6 +7,7 @@
 
 using Akka.Actor.Setup;
 using Akka.Configuration;
+using Akka.Event;
 
 namespace Akka.Persistence.TestKit
 {
@@ -111,6 +112,11 @@ public async Task WithJournalRecovery(Func<JournalRecoveryBehavior, Task> behavi
                 await behaviorSelector(Journal.OnRecovery);
                 await execution();
             }
+            catch (Exception ex)
+            {
+                Log.Error(ex, "Error during execution of WithJournalRecovery");
+                throw;
+            }
             finally
             {
                 await Journal.OnRecovery.Pass();
@@ -136,6 +142,11 @@ public async Task WithJournalWrite(Func<JournalWriteBehavior, Task> behaviorSele
                 await behaviorSelector(Journal.OnWrite);
                 await execution();
             }
+            catch (Exception ex)
+            {
+                Log.Error(ex, "Error during execution of WithJournalWrite");
+                throw;
+            }
             finally
             {
                 await Journal.OnWrite.Pass();
@@ -157,7 +168,7 @@ public Task WithJournalRecovery(Func<JournalRecoveryBehavior, Task> behaviorSele
                 if (execution == null) throw new ArgumentNullException(nameof(execution));
 
                 execution();
-                return Task.FromResult(new object());
+                return Task.CompletedTask;
             });
 
         /// <summary>
@@ -175,7 +186,7 @@ public Task WithJournalWrite(Func<JournalWriteBehavior, Task> behaviorSelector,
                 if (execution == null) throw new ArgumentNullException(nameof(execution));
 
                 execution();
-                return Task.FromResult(new object());
+                return Task.CompletedTask;
             });
 
         /// <summary>
@@ -197,6 +208,11 @@ public async Task WithSnapshotSave(Func<SnapshotStoreSaveBehavior, Task> behavio
                 await behaviorSelector(Snapshots.OnSave);
                 await execution();
             }
+            catch (Exception ex)
+            {
+                Log.Error(ex, "Error during execution of WithSnapshotSave");
+                throw;
+            }
             finally
             {
                 await Snapshots.OnSave.Pass();
@@ -222,6 +238,11 @@ public async Task WithSnapshotLoad(Func<SnapshotStoreLoadBehavior, Task> behavio
                 await behaviorSelector(Snapshots.OnLoad);
                 await execution();
             }
+            catch (Exception ex)
+            {
+                Log.Error(ex, "Error during execution of WithSnapshotLoad");
+                throw;
+            }
             finally
             {
                 await Snapshots.OnLoad.Pass();
@@ -247,6 +268,11 @@ public async Task WithSnapshotDelete(Func<SnapshotStoreDeleteBehavior, Task> beh
                 await behaviorSelector(Snapshots.OnDelete);
                 await execution();
             }
+            catch (Exception ex)
+            {
+                Log.Error(ex, "Error during execution of WithSnapshotDelete");
+                throw;
+            }
             finally
             {
                 await Snapshots.OnDelete.Pass();
@@ -268,7 +294,7 @@ public Task WithSnapshotSave(Func<SnapshotStoreSaveBehavior, Task> behaviorSelec
                 if (execution == null) throw new ArgumentNullException(nameof(execution));
 
                 execution();
-                return Task.FromResult(true);
+                return Task.CompletedTask;
             });
 
         /// <summary>
@@ -286,7 +312,7 @@ public Task WithSnapshotLoad(Func<SnapshotStoreLoadBehavior, Task> behaviorSelec
                 if (execution == null) throw new ArgumentNullException(nameof(execution));
 
                 execution();
-                return Task.FromResult(true);
+                return Task.CompletedTask;
             });
 
         /// <summary>
@@ -304,7 +330,7 @@ public Task WithSnapshotDelete(Func<SnapshotStoreDeleteBehavior, Task> behaviorS
                 if (execution == null) throw new ArgumentNullException(nameof(execution));
 
                 execution();
-                return Task.FromResult(true);
+                return Task.CompletedTask;
             });
 
         /// <summary>
@@ -348,7 +374,7 @@ private static Config GetConfig(Config customConfig)
         {
             var defaultConfig = ConfigurationFactory.FromResource<TestJournal>("Akka.Persistence.TestKit.config.conf");
             if (customConfig == Config.Empty) return defaultConfig;
-            else return defaultConfig.SafeWithFallback(customConfig);
+            else return customConfig.WithFallback(defaultConfig);
         }        
     }
 }

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.TestKit/Journal/JournalRecoveryBehaviorSetter.cs

@@ -26,7 +26,7 @@ internal JournalRecoveryBehaviorSetter(IActorRef journal)
         public Task SetInterceptorAsync(IJournalInterceptor interceptor)
             => _journal.Ask<TestJournal.Ack>(
                 new TestJournal.UseRecoveryInterceptor(interceptor),
-                TimeSpan.FromSeconds(3)
+                TimeSpan.FromSeconds(0.5)
             );
     }
 }

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

@@ -26,7 +26,7 @@ internal JournalWriteBehaviorSetter(IActorRef journal)
         public Task SetInterceptorAsync(IJournalInterceptor interceptor)
             => _journal.Ask<TestJournal.Ack>(
                 new TestJournal.UseWriteInterceptor(interceptor),
-                TimeSpan.FromSeconds(3)
+                TimeSpan.FromSeconds(0.5)
             );
     }
 }