add some tracing

Author: timothycoleman

src/KurrentDB.Core.Tests/Services/Replication/ReplicationTestHelper.cs

@@ -22,20 +22,40 @@ public static ClientMessage.WriteEventsCompleted WriteEvent<TLogFormat, TStreamI
 		string streamId) {
 		var resetEvent = new ManualResetEventSlim();
 		ClientMessage.WriteEventsCompleted writeResult = null;
-		node.Node.MainQueue.Publish(new ClientMessage.WriteEvents(Guid.NewGuid(), Guid.NewGuid(),
+		ClientMessage.WriteEvents writeRequest = default;
+		writeRequest = new ClientMessage.WriteEvents(
+			Guid.NewGuid(),
+			Guid.NewGuid(),
 			new CallbackEnvelope(msg => {
+				if (writeRequest.Trace)
+					writeRequest.AddTrace($"Envelope received reply: {msg}");
 				writeResult = (ClientMessage.WriteEventsCompleted)msg;
 				resetEvent.Set();
-			}), false, streamId, -1, events,
-			SystemAccounts.System, new Dictionary<string, string> {
+			}),
+			requireLeader: false, //qq should be leader but this might help make sure?
+			streamId,
+			-1,
+			events,
+			SystemAccounts.System,
+			new Dictionary<string, string> {
 				["uid"] = SystemUsers.Admin,
 				["pwd"] = SystemUsers.DefaultAdminPassword
-			}));
+			}) {
+			Trace = true
+		};
+
+		node.Node.MainQueue.Publish(writeRequest);
+
 		if (!resetEvent.Wait(_timeout)) {
-			Assert.Fail("Timed out waiting for event to be written");
+			var joined = string.Join(Environment.NewLine, writeRequest.GetTraceMessages());
+			Assert.Fail("Timed out waiting for event to be written: " + Environment.NewLine + joined);
 			return null;
 		}
 
+		//qq temp
+//		var joinedTemp = string.Join(Environment.NewLine, writeRequest.GetTraceMessages());
+//		Assert.Fail("Success: " + Environment.NewLine + joinedTemp);
+
 		return writeResult;
 	}
 

src/KurrentDB.Core/Bus/InMemoryBus.MessageHierarchy.cs

@@ -127,6 +127,8 @@ public override async ValueTask InvokeAsync(Message message, CancellationToken t
 
 			// first handler is the parent
 			foreach (var handler in _handlers) {
+				if (message.Trace)
+					message.AddTrace($"InMemoryBus calling handler {handler}");
 				await handler.Invoke(Unsafe.As<T>(message), token);
 			}
 		}

src/KurrentDB.Core/Bus/QueuedHandlerThreadPool.cs

@@ -135,9 +135,13 @@ async void IThreadPoolWorkItem.Execute() {
 				while (!_lifetimeToken.IsCancellationRequested && _queue.TryDequeue(out var item)) {
 					var start = _tracker.RecordMessageDequeued(item.EnqueuedAt);
 					var msg = item.Message;
+					//qq maybe build this into the tracker?
+					if (msg.Trace)
+						msg.AddTrace($"Dequeued from {Name}");
 #if DEBUG
 					_queueStats.Dequeued(msg);
 #endif
+
 					try {
 						var queueCnt = _queue.Count;
 						_queueStats.ProcessingStarted(msg.GetType(), queueCnt);
@@ -206,6 +210,10 @@ public void Publish(Message message) {
 		_queueStats.Enqueued();
 #endif
 		_queue.Enqueue(new(_tracker.Now, message));
+		//qq maybe build this into the tracker?
+		if (message.Trace)
+			message.AddTrace($"Enqueued into {Name}");
+
 		if (!_lifetimeToken.IsCancellationRequested && Interlocked.CompareExchange(ref _isRunning, 1, 0) == 0) {
 			ThreadPool.UnsafeQueueUserWorkItem(this, preferLocal: false);
 		}

src/KurrentDB.Core/Messaging/Message.cs

@@ -2,6 +2,7 @@
 // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
 
 using System;
+using System.Collections.Generic;
 using System.Threading;
 using Newtonsoft.Json;
 
@@ -26,4 +27,40 @@ public DerivedMessageAttribute(object messageGroup) {
 public abstract partial class Message(CancellationToken token = default) {
 	[JsonIgnore]
 	public CancellationToken CancellationToken => token;
+
+	[JsonIgnore]
+	public bool Trace { get; init; }
+
+	public string[] GetTraceMessages() {
+		var list = new List<string>();
+		foreach (var message in GetTrace()) {
+			if (message is string m) {
+				list.Add(m);
+			} else if (message is Message msg) {
+				var msgName = msg.GetType().Name;
+				foreach (var y in msg.GetTraceMessages()) {
+					list.Add($"  {msgName}: {y}");
+				}
+			}
+		}
+		return list.ToArray();
+	}
+
+	public object[] GetTrace() {
+		lock (this) {
+			return _traceMessages.ToArray();
+		}
+	}
+
+	private List<object> _traceMessages;
+
+	public void AddTrace(object message) {
+		if (!Trace)
+			return;
+
+		lock (this) {
+			_traceMessages ??= [];
+			_traceMessages.Add(message);
+		}
+	}
 }

src/KurrentDB.Core/Services/RequestManager/Managers/RequestManagerBase.cs

@@ -42,6 +42,7 @@ public abstract class RequestManagerBase :
 	protected long FailureCurrentVersion = -1;
 	protected long TransactionId;
 
+	private readonly Message _request;
 	protected readonly CommitSource CommitSource;
 	protected long LastEventPosition;
 	protected bool Registered;
@@ -70,7 +71,8 @@ protected RequestManagerBase(
 			CommitSource commitSource,
 			int prepareCount = 0,
 			long transactionId = -1,
-			bool waitForCommit = false) {
+			bool waitForCommit = false,
+			Message request = default) {
 		Ensure.NotEmptyGuid(internalCorrId, nameof(internalCorrId));
 		Ensure.NotEmptyGuid(clientCorrId, nameof(clientCorrId));
 		Ensure.NotNull(publisher, nameof(publisher));
@@ -87,6 +89,7 @@ protected RequestManagerBase(
 		CommitSource = commitSource;
 		_prepareCount = prepareCount;
 		TransactionId = transactionId;
+		_request = request;
 		_commitReceived = !waitForCommit; //if not waiting for commit flag as true
 		_allPreparesWritten = _prepareCount == 0; //if not waiting for prepares flag as true
 		if (prepareCount == 0 && waitForCommit == false) {
@@ -102,11 +105,18 @@ protected RequestManagerBase(
 	protected abstract Message ClientFailMsg { get; }
 	public void Start() {
 		NextTimeoutTime = DateTime.UtcNow + Timeout;
-		Publisher.Publish(WriteRequestMsg);
+		var x = WriteRequestMsg;
+		if (_request.Trace)
+			_request.AddTrace(x);
+		Publisher.Publish(x);
 	}
 
 	public void Handle(StorageMessage.PrepareAck message) {
 		if (Interlocked.Read(ref _complete) == 1 || _allPreparesWritten) { return; }
+
+		if (_request.Trace)
+			_request.AddTrace($"RequestManager Received {message.GetType().Name}");
+
 		NextTimeoutTime = DateTime.UtcNow + Timeout;
 		if (message.Flags.HasAnyOf(PrepareFlags.TransactionBegin)) {
 			TransactionId = message.LogPosition;
@@ -125,6 +135,10 @@ public void Handle(StorageMessage.PrepareAck message) {
 	}
 	public virtual void Handle(StorageMessage.CommitIndexed message) {
 		if (Interlocked.Read(ref _complete) == 1 || _commitReceived) { return; }
+
+		if (_request.Trace)
+			_request.AddTrace($"RequestManager Received {message.GetType().Name}");
+
 		NextTimeoutTime = DateTime.UtcNow + Timeout;
 		_commitReceived = true;
 		_allEventsWritten = _commitReceived && _allPreparesWritten;
@@ -153,6 +167,8 @@ protected virtual void Committed() {
 		Publisher.Publish(new StorageMessage.RequestCompleted(InternalCorrId, true));
 	}
 	public void Handle(StorageMessage.RequestManagerTimerTick message) {
+		if (_request.Trace)
+			_request.AddTrace($"RequestManager Received {message.GetType().Name}");
 		if (_allEventsWritten) { AllEventsWritten(); }
 		if (Interlocked.Read(ref _complete) == 1 || message.UtcNow < NextTimeoutTime)
 			return;
@@ -161,16 +177,25 @@ public void Handle(StorageMessage.RequestManagerTimerTick message) {
 		CompleteFailedRequest(result, msg);
 	}
 	public void Handle(StorageMessage.InvalidTransaction message) {
+		if (_request.Trace)
+			_request.AddTrace($"RequestManager Received {message.GetType().Name}");
 		CompleteFailedRequest(OperationResult.InvalidTransaction, "Invalid transaction.");
 	}
 	public void Handle(StorageMessage.WrongExpectedVersion message) {
+		if (_request.Trace)
+			_request.AddTrace($"RequestManager Received {message.GetType().Name}");
 		FailureCurrentVersion = message.CurrentVersion;
 		CompleteFailedRequest(OperationResult.WrongExpectedVersion, "Wrong expected version.", message.CurrentVersion);
 	}
 	public void Handle(StorageMessage.StreamDeleted message) {
+		if (_request.Trace)
+			_request.AddTrace($"RequestManager Received {message.GetType().Name}");
 		CompleteFailedRequest(OperationResult.StreamDeleted, "Stream is deleted.");
 	}
 	public void Handle(StorageMessage.AlreadyCommitted message) {
+		if (_request.Trace)
+			_request.AddTrace($"RequestManager Received {message.GetType().Name}");
+
 		if (Interlocked.Read(ref _complete) == 1 || _allEventsWritten) { return; }
 		Log.Debug("IDEMPOTENT WRITE TO STREAM ClientCorrelationID {clientCorrelationId}, {message}.", ClientCorrId,
 			message);
@@ -193,6 +218,8 @@ private void CompleteFailedRequest(OperationResult result, string error, long cu
 		if (Interlocked.CompareExchange(ref _complete, 1, 0) == 1) { return; }
 		Result = result;
 		FailureMessage = error;
+		if (_request.Trace)
+			_request.AddTrace($"RequestManager Replying with RequestCompleted {result}");
 		Publisher.Publish(new StorageMessage.RequestCompleted(InternalCorrId, false, currentVersion));
 		_clientResponseEnvelope.ReplyWith(ClientFailMsg);
 	}

src/KurrentDB.Core/Services/RequestManager/Managers/WriteEvents.cs

@@ -14,7 +14,8 @@ public class WriteEvents : RequestManagerBase {
 	private readonly string _streamId;
 	private readonly Event[] _events;
 	private readonly CancellationToken _cancellationToken;
-	public WriteEvents(IPublisher publisher,
+	public WriteEvents(
+		IPublisher publisher,
 		TimeSpan timeout,
 		IEnvelope clientResponseEnvelope,
 		Guid internalCorrId,
@@ -23,6 +24,7 @@ public WriteEvents(IPublisher publisher,
 		long expectedVersion,
 		Event[] events,
 		CommitSource commitSource,
+		ClientMessage.WriteEvents request = default,
 		CancellationToken cancellationToken = default)
 		: base(
 				 publisher,
@@ -33,7 +35,8 @@ public WriteEvents(IPublisher publisher,
 				 expectedVersion,
 				 commitSource,
 				 prepareCount: 0,
-				 waitForCommit: true) {
+				 waitForCommit: true,
+				 request: request) {
 		_streamId = streamId;
 		_events = events;
 		_cancellationToken = cancellationToken;
@@ -46,7 +49,7 @@ public WriteEvents(IPublisher publisher,
 				_streamId,
 				ExpectedVersion,
 				_events,
-				_cancellationToken);
+				_cancellationToken) { Trace = true };
 
 
 	protected override Message ClientSuccessMsg =>

src/KurrentDB.Core/Services/RequestManager/RequestManagementService.cs

@@ -69,6 +69,7 @@ public void Handle(ClientMessage.WriteEvents message) {
 							message.ExpectedVersion,
 							message.Events,
 							_commitSource,
+							message,
 							message.CancellationToken);
 		_currentRequests.Add(message.InternalCorrId, manager);
 		_currentTimedRequests.Add(message.InternalCorrId, Stopwatch.StartNew());

src/KurrentDB.Core/Services/VNode/VNodeFSM.cs

@@ -100,8 +100,11 @@ private static void EnsureActionType(Type expectedMessageType, [DisallowNull] Mu
 	private struct HandlersBuffer {
 		private Handler _handler;
 
-		public readonly ValueTask InvokeAsync(VNodeState index, Message message, CancellationToken token)
-			=> Unsafe.Add(ref Unsafe.AsRef(in _handler), (int)index).InvokeAsync(message, token);
+		public readonly ValueTask InvokeAsync(VNodeState index, Message message, CancellationToken token) {
+			if (message.Trace)
+				message.AddTrace($"FSM handling in state {index}");
+			return Unsafe.Add(ref Unsafe.AsRef(in _handler), (int)index).InvokeAsync(message, token);
+		}
 	}
 
 	[StructLayout(LayoutKind.Auto)]