Add diagnostic logging to MemoryJournal to track lock contention

Added comprehensive telemetry to investigate intermittent test failures in
InMemoryEventsByTagSpec.ReadJournal_live_query_EventsByTag_should_find_events_from_offset_exclusive.

Changes:
- Added Stopwatch-based timing instrumentation to track lock acquisition wait times in WriteMessagesAsync and ReplayTaggedMessagesAsync
- Added DEBUG logging for all journal message types (ReplayTaggedMessages, ReplayAllEvents, SelectCurrentPersistenceIds)
- Logs thread IDs, lock wait times, lock hold times, and event counts
- Enabled DEBUG log level in InMemoryEventsByTagSpec test configuration

Diagnostic output includes:
- "[DIAG] WriteMessagesAsync called on thread X, attempting to acquire lock"
- "[DIAG] Lock acquired after Xms on thread Y"
- "[DIAG] Lock released after holding for Xms"
- "[DIAG] Wrote event for {persistenceId}, seq {seqNr}, total events: {count}"
- "[DIAG] ReplayTaggedMessages lock acquired after Xms"
- "[DIAG] Found X events matching tag 'Y'"

This instrumentation will capture timing data on CI/CD build servers to determine
if the test failures are caused by:
1. Lock contention/fairness issues with Monitor
2. Thread pool starvation
3. Async execution context delays
4. RecoveryPermitter bottlenecks

All diagnostic logging uses [DIAG] prefix for easy filtering. These changes are
temporary for diagnosis and will be reverted once root cause is identified.

Author: Aaronontheweb

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.Persistence/Journal/MemoryJournal.cs

@@ -8,6 +8,7 @@
 using System;
 using System.Collections.Generic;
 using System.Collections.Immutable;
+using System.Diagnostics;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -35,6 +36,9 @@ public class MemoryJournal : AsyncWriteJournal
 
         private readonly object _lock = new();
         private readonly Dictionary<string, long> _deletedTo = new();
+        private ILoggingAdapter _log;
+
+        protected ILoggingAdapter Log => _log ??= Context.GetLogger();
 
         protected virtual List<IPersistentRepresentation> EventLog => _eventLog;
         protected virtual Dictionary<string, List<IPersistentRepresentation>> EventsByPersistenceId => _eventsByPersistenceId;
@@ -43,8 +47,16 @@ public class MemoryJournal : AsyncWriteJournal
 
         protected override Task<IImmutableList<Exception>> WriteMessagesAsync(IEnumerable<AtomicWrite> messages, CancellationToken cancellationToken)
         {
+            var waitStartTime = Stopwatch.GetTimestamp();
+            var threadId = Environment.CurrentManagedThreadId;
+            Log.Debug("[DIAG] WriteMessagesAsync called on thread {0}, attempting to acquire lock", threadId);
+
             lock (Lock)
             {
+                var lockAcquiredTime = Stopwatch.GetTimestamp();
+                var waitTimeMs = (lockAcquiredTime - waitStartTime) * 1000.0 / Stopwatch.Frequency;
+                Log.Debug("[DIAG] Lock acquired after {0:F2}ms on thread {1}", waitTimeMs, threadId);
+
                 foreach (var w in messages)
                 {
                     foreach (var p in (IEnumerable<IPersistentRepresentation>)w.Payload)
@@ -60,8 +72,17 @@ protected override Task<IImmutableList<Exception>> WriteMessagesAsync(IEnumerabl
                             EventsByPersistenceId[persistentRepresentation.PersistenceId] = pidEvents;
                         }
                         pidEvents.Add(persistentRepresentation);
+
+                        Log.Debug("[DIAG] Wrote event for {0}, seq {1}, total events in log: {2}",
+                            persistentRepresentation.PersistenceId,
+                            persistentRepresentation.SequenceNr,
+                            EventLog.Count);
                     }
                 }
+
+                var lockReleasedTime = Stopwatch.GetTimestamp();
+                var holdTimeMs = (lockReleasedTime - lockAcquiredTime) * 1000.0 / Stopwatch.Frequency;
+                Log.Debug("[DIAG] Lock released after holding for {0:F2}ms on thread {1}", holdTimeMs, threadId);
             }
 
             return Task.FromResult<IImmutableList<Exception>>(null);
@@ -219,16 +240,21 @@ protected override bool ReceivePluginInternal(object message)
             switch (message)
             {
                 case SelectCurrentPersistenceIds request:
+                    Log.Debug("[DIAG] Received SelectCurrentPersistenceIds from {0}", request.ReplyTo);
                     SelectAllPersistenceIdsAsync(request.Offset)
                         .PipeTo(request.ReplyTo, success: result => new CurrentPersistenceIds(result.Item1, result.LastOrdering));
                     return true;
 
                 case ReplayTaggedMessages replay:
+                    Log.Debug("[DIAG] Received ReplayTaggedMessages for tag '{0}', fromOffset={1}, max={2}, toOffset={3}, replyTo={4}",
+                        replay.Tag, replay.FromOffset, replay.Max, replay.ToOffset, replay.ReplyTo);
                     ReplayTaggedMessagesAsync(replay)
                         .PipeTo(replay.ReplyTo, success: h => new ReplayTaggedMessagesSuccess(h), failure: e => new ReplayMessagesFailure(e));
                     return true;
 
                 case ReplayAllEvents replay:
+                    Log.Debug("[DIAG] Received ReplayAllEvents fromOffset={0}, max={1}, replyTo={2}",
+                        replay.FromOffset, replay.Max, replay.ReplyTo);
                     ReplayAllEventsAsync(replay)
                         .PipeTo(replay.ReplyTo, success: h => new EventReplaySuccess(h),
                             failure: e => new EventReplayFailure(e));
@@ -258,11 +284,19 @@ protected override bool ReceivePluginInternal(object message)
         /// </summary>
         private Task<int> ReplayTaggedMessagesAsync(ReplayTaggedMessages replay)
         {
+            var waitStartTime = Stopwatch.GetTimestamp();
+            var threadId = Environment.CurrentManagedThreadId;
+            Log.Debug("[DIAG] ReplayTaggedMessagesAsync starting for tag '{0}' on thread {1}, attempting to acquire lock", replay.Tag, threadId);
+
             IPersistentRepresentation[] snapshot;
             int count;
 
             lock (Lock)
             {
+                var lockAcquiredTime = Stopwatch.GetTimestamp();
+                var waitTimeMs = (lockAcquiredTime - waitStartTime) * 1000.0 / Stopwatch.Frequency;
+                Log.Debug("[DIAG] ReplayTaggedMessages lock acquired after {0:F2}ms on thread {1}", waitTimeMs, threadId);
+
                 // Scan for events with matching tag
                 snapshot = EventLog
                     .Where(e => e.Payload is Tagged tagged && tagged.Tags.Contains(replay.Tag))
@@ -271,16 +305,22 @@ private Task<int> ReplayTaggedMessagesAsync(ReplayTaggedMessages replay)
                     .ToArray();
 
                 count = EventLog.Count(e => e.Payload is Tagged tagged && tagged.Tags.Contains(replay.Tag));
+
+                Log.Debug("[DIAG] Found {0} events matching tag '{1}', total tagged events: {2}, EventLog size: {3}",
+                    snapshot.Length, replay.Tag, count, EventLog.Count);
             }
 
             // Send messages outside the lock to avoid potential deadlocks
             var index = 0;
             foreach (var persistence in snapshot)
             {
                 replay.ReplyTo.Tell(new ReplayedTaggedMessage(persistence, replay.Tag, replay.FromOffset + index), ActorRefs.NoSender);
+                Log.Debug("[DIAG] Sent ReplayedTaggedMessage for {0}, seq {1}, offset {2}",
+                    persistence.PersistenceId, persistence.SequenceNr, replay.FromOffset + index);
                 index++;
             }
 
+            Log.Debug("[DIAG] ReplayTaggedMessagesAsync completed, returning highestSequenceNr={0}", count - 1);
             return Task.FromResult(count - 1);
         }