Fix writer starvation in MemoryJournal

src/contrib/persistence/Akka.Persistence.Query.InMemory.Tests/InMemoryEventsByTagSpec.cs

@@ -14,7 +14,7 @@ namespace Akka.Persistence.Query.InMemory.Tests
     public class InMemoryEventsByTagSpec : EventsByTagSpec
     {
         private static Config Config() => ConfigurationFactory.ParseString(@"
-            akka.loglevel = INFO
+            akka.loglevel = DEBUG
             akka.persistence.journal.inmem {
                 event-adapters {
                   color-tagger  = ""Akka.Persistence.TCK.Query.ColorFruitTagger, Akka.Persistence.TCK""

src/core/Akka.API.Tests/verify/CoreAPISpec.ApprovePersistence.DotNet.verified.txt

@@ -1059,7 +1059,7 @@ namespace Akka.Persistence.Journal
         protected virtual System.Collections.Generic.Dictionary<string, long> DeletedTo { get; }
         protected virtual System.Collections.Generic.List<Akka.Persistence.IPersistentRepresentation> EventLog { get; }
         protected virtual System.Collections.Generic.Dictionary<string, System.Collections.Generic.List<Akka.Persistence.IPersistentRepresentation>> EventsByPersistenceId { get; }
-        protected virtual System.Threading.ReaderWriterLockSlim Lock { get; }
+        protected virtual object Lock { get; }
         public System.Collections.Generic.IDictionary<string, System.Collections.Generic.LinkedList<Akka.Persistence.IPersistentRepresentation>> Add(Akka.Persistence.IPersistentRepresentation persistent) { }
         public System.Collections.Generic.IDictionary<string, System.Collections.Generic.LinkedList<Akka.Persistence.IPersistentRepresentation>> Delete(string pid, long seqNr) { }
         protected override System.Threading.Tasks.Task DeleteMessagesToAsync(string persistenceId, long toSequenceNr, System.Threading.CancellationToken cancellationToken) { }
@@ -1185,7 +1185,7 @@ namespace Akka.Persistence.Journal
         protected override System.Collections.Generic.Dictionary<string, long> DeletedTo { get; }
         protected override System.Collections.Generic.List<Akka.Persistence.IPersistentRepresentation> EventLog { get; }
         protected override System.Collections.Generic.Dictionary<string, System.Collections.Generic.List<Akka.Persistence.IPersistentRepresentation>> EventsByPersistenceId { get; }
-        protected override System.Threading.ReaderWriterLockSlim Lock { get; }
+        protected override object Lock { get; }
     }
     public struct SingleEventSequence : Akka.Persistence.Journal.IEventSequence, System.IEquatable<Akka.Persistence.Journal.IEventSequence>
     {

src/core/Akka.API.Tests/verify/CoreAPISpec.ApprovePersistence.Net.verified.txt

@@ -1058,7 +1058,7 @@ namespace Akka.Persistence.Journal
         protected virtual System.Collections.Generic.Dictionary<string, long> DeletedTo { get; }
         protected virtual System.Collections.Generic.List<Akka.Persistence.IPersistentRepresentation> EventLog { get; }
         protected virtual System.Collections.Generic.Dictionary<string, System.Collections.Generic.List<Akka.Persistence.IPersistentRepresentation>> EventsByPersistenceId { get; }
-        protected virtual System.Threading.ReaderWriterLockSlim Lock { get; }
+        protected virtual object Lock { get; }
         public System.Collections.Generic.IDictionary<string, System.Collections.Generic.LinkedList<Akka.Persistence.IPersistentRepresentation>> Add(Akka.Persistence.IPersistentRepresentation persistent) { }
         public System.Collections.Generic.IDictionary<string, System.Collections.Generic.LinkedList<Akka.Persistence.IPersistentRepresentation>> Delete(string pid, long seqNr) { }
         protected override System.Threading.Tasks.Task DeleteMessagesToAsync(string persistenceId, long toSequenceNr, System.Threading.CancellationToken cancellationToken) { }
@@ -1184,7 +1184,7 @@ namespace Akka.Persistence.Journal
         protected override System.Collections.Generic.Dictionary<string, long> DeletedTo { get; }
         protected override System.Collections.Generic.List<Akka.Persistence.IPersistentRepresentation> EventLog { get; }
         protected override System.Collections.Generic.Dictionary<string, System.Collections.Generic.List<Akka.Persistence.IPersistentRepresentation>> EventsByPersistenceId { get; }
-        protected override System.Threading.ReaderWriterLockSlim Lock { get; }
+        protected override object Lock { get; }
     }
     public struct SingleEventSequence : Akka.Persistence.Journal.IEventSequence, System.IEquatable<Akka.Persistence.Journal.IEventSequence>
     {

src/core/Akka.Persistence.TCK/Query/TestActor.cs

@@ -34,21 +34,35 @@ public TestActor(string persistenceId)
 
         public override string PersistenceId { get; }
 
+        protected override void PreStart()
+        {
+            Log.Debug("[DIAG-ACTOR] TestActor {0} PreStart called", PersistenceId);
+            base.PreStart();
+        }
+
         protected override void OnRecover(object message)
         {
+            Log.Debug("[DIAG-ACTOR] TestActor {0} OnRecover: {1}", PersistenceId, message);
         }
 
         protected override void OnCommand(object message)
         {
+            Log.Debug("[DIAG-ACTOR] TestActor {0} OnCommand received: {1}", PersistenceId, message);
             switch (message)
             {
                 case DeleteCommand delete:
                     DeleteMessages(delete.ToSequenceNr);
                     Become(WhileDeleting(Sender)); // need to wait for delete ACK to return
                     break;
                 case string cmd:
+                    Log.Debug("[DIAG-ACTOR] TestActor {0} calling Persist for: {1}", PersistenceId, cmd);
                     var sender = Sender;
-                    Persist(cmd, e => sender.Tell($"{e}-done"));
+                    Persist(cmd, e =>
+                    {
+                        Log.Debug("[DIAG-ACTOR] TestActor {0} Persist callback executing for: {1}", PersistenceId, e);
+                        sender.Tell($"{e}-done");
+                    });
+                    Log.Debug("[DIAG-ACTOR] TestActor {0} Persist call returned for: {1}", PersistenceId, cmd);
                     break;
             }
         }

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

@@ -391,8 +391,10 @@ private async Task ExecuteBatch(WriteMessages message, int atomicWriteCount, IAc
                 // try in case AsyncWriteMessages throws
                 try
                 {
+                    // NOTE: Not using ConfigureAwait(false) to ensure continuation runs on actor's dispatcher thread
+                    // This ensures proper synchronization context and avoids timing issues with lock contention
                     var writeResult =
-                        await _breaker.WithCircuitBreaker((prepared, awj: this), (state, ct) => state.awj.WriteMessagesAsync(state.prepared, ct)).ConfigureAwait(false);
+                        await _breaker.WithCircuitBreaker((prepared, awj: this), (state, ct) => state.awj.WriteMessagesAsync(state.prepared, ct));
 
                     ProcessResults(writeResult, atomicWriteCount, message, _resequencer, resequencerCounter, self);
                 }