Merge pull request #1413 from Cratis/feautre/tcs-and-logging

Feautre/tcs and logging
Cratis · Sep 19, 2024 · 7f48560 · 7f48560
2 parents 6a57a0d + e04889c
commit 7f48560
Show file tree

Hide file tree

Showing 13 changed files with 30 additions and 30 deletions.
diff --git a/Source/Clients/DotNET/ChronicleConnection.cs b/Source/Clients/DotNET/ChronicleConnection.cs
@@ -98,10 +98,8 @@ async Task Connect()
 
         _channel = CreateGrpcChannel();
         _connectionService = _channel.CreateGrpcService<IConnectionService>();
-
         _lastKeepAlive = DateTimeOffset.UtcNow;
-
-        _connectTcs = new TaskCompletionSource();
+        _connectTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
 
         _keepAliveSubscription = _connectionService.Connect(new()
         {

diff --git a/Source/Clients/DotNET/Connections/ConnectionLifecycle.cs b/Source/Clients/DotNET/Connections/ConnectionLifecycle.cs
@@ -56,9 +56,7 @@ public async Task Connected()
     public async Task Disconnected()
     {
         IsConnected = false;
-
         logger.Disconnected();
-
         var tasks = OnDisconnected.GetInvocationList().Select(_ => Task.Run(async () =>
         {
             try

diff --git a/Source/Clients/DotNET/Connections/IConnectionLifecycle.cs b/Source/Clients/DotNET/Connections/IConnectionLifecycle.cs
@@ -19,7 +19,7 @@ public interface IConnectionLifecycle
     event Disconnected OnDisconnected;
 
     /// <summary>
-    /// Gets whether or not the client is connected.
+    /// Gets whether, or not the client is connected.
     /// </summary>
     bool IsConnected { get; }
 

diff --git a/Source/Clients/DotNET/Reactions/Reactors.cs b/Source/Clients/DotNET/Reactions/Reactors.cs
@@ -171,14 +171,14 @@ async Task ObserverMethod(ISubject<ReactorMessage> messages, ReactorHandler hand
 
                 BaseIdentityProvider.SetCurrentIdentity(Identity.System with { OnBehalfOf = context.CausedBy });
                 var eventType = _eventTypes.GetClrTypeFor(metadata.Type.Id);
-
                 var content = await _eventSerializer.Deserialize(eventType, JsonNode.Parse(@event.Content)!.AsObject());
 
                 await handler.OnNext(metadata, context, content);
                 lastSuccessfullyObservedEvent = @event.Metadata.SequenceNumber;
             }
             catch (Exception ex)
             {
+                _logger.ErrorWhileHandlingEvent(ex, @event.Metadata.Type.Id, handler.Id);
                 exceptionMessages = ex.GetAllMessages();
                 exceptionStackTrace = ex.StackTrace ?? string.Empty;
                 state = ObservationState.Failed;

diff --git a/Source/Clients/DotNET/Reactions/ReactorsLogMessages.cs b/Source/Clients/DotNET/Reactions/ReactorsLogMessages.cs
@@ -15,5 +15,8 @@ internal static partial class ReactorsLogMessages
     internal static partial void DiscoverAllReactors(this ILogger<Reactors> logger);
 
     [LoggerMessage(LogLevel.Trace, "Event of type {EventTypeId} was received for Reactor {ReactorId}")]
-    internal static partial void EventReceived(this ILogger<Reactors> logger, EventTypeId eventTypeId, ReactorId ReactorId);
+    internal static partial void EventReceived(this ILogger<Reactors> logger, EventTypeId eventTypeId, ReactorId reactorId);
+
+    [LoggerMessage(LogLevel.Warning, "An error occurred while handling event of type {EventTypeId} was for Reactor {ReactorId}")]
+    internal static partial void ErrorWhileHandlingEvent(this ILogger<Reactors> logger, Exception ex, EventTypeId eventTypeId, ReactorId reactorId);
 }
diff --git a/Source/Clients/DotNET/Reducers/ReducerInvoker.cs b/Source/Clients/DotNET/Reducers/ReducerInvoker.cs
@@ -62,9 +62,8 @@ public Task<ReduceResult> Invoke(IEnumerable<EventAndContext> eventsAndContexts,
 
             try
             {
-                if (_methodsByEventType.ContainsKey(eventType))
+                if (_methodsByEventType.TryGetValue(eventType, out var method))
                 {
-                    var method = _methodsByEventType[eventType];
                     var parameters = method.GetParameters();
 
                     if (parameters.Length == 3)

diff --git a/Source/Clients/DotNET/Reducers/Reducers.cs b/Source/Clients/DotNET/Reducers/Reducers.cs
@@ -124,7 +124,7 @@ void RegisterReducer(IReducerHandler handler)
             handler.Id,
             handler.EventSequenceId);
 
-        var registration = new RegisterReducer()
+        var registration = new RegisterReducer
         {
             ConnectionId = eventStore.Connection.Lifecycle.ConnectionId,
             EventStoreName = eventStore.Name,
@@ -134,12 +134,12 @@ void RegisterReducer(IReducerHandler handler)
                 ReducerId = handler.Id,
                 EventSequenceId = handler.EventSequenceId,
                 EventTypes = handler.EventTypes.Select(et => new EventTypeWithKeyExpression { EventType = et.ToContract(), Key = "$eventSourceId" }).ToArray(),
-                Model = new Contracts.Models.ModelDefinition()
+                Model = new Contracts.Models.ModelDefinition
                 {
                     Name = modelNameResolver.GetNameFor(handler.ReadModelType),
                     Schema = jsonSchemaGenerator.Generate(handler.ReadModelType).ToJson()
                 },
-                Sink = new SinkDefinition()
+                Sink = new SinkDefinition
                 {
                     TypeId = WellKnownSinkTypes.MongoDB
                 }
@@ -168,27 +168,25 @@ async Task ObserverMethod(ISubject<ReducerMessage> messages, IReducerHandler han
         {
             var metadata = @event.Metadata.ToClient();
             var context = @event.Context.ToClient();
-            var eventType = eventTypes.GetClrTypeFor(metadata.Type.Id);
             var contentAsExpando = JsonSerializer.Deserialize<ExpandoObject>(@event.Content)!;
             return new AppendedEvent(
                 metadata,
                 context,
                 contentAsExpando);
-        });
+        }).ToList();
 
         try
         {
             BaseIdentityProvider.SetCurrentIdentity(Identity.System);
             var initialState = operation.InitialState is null ? null : JsonSerializer.Deserialize(operation.InitialState, handler.ReadModelType, jsonSerializerOptions);
-            var reduceResult = await handler.OnNext(
-                appendedEvents,
-                initialState);
+            var reduceResult = await handler.OnNext(appendedEvents, initialState);
 
             modelState = JsonSerializer.Serialize(reduceResult.ModelState, jsonSerializerOptions);
-            lastSuccessfullyObservedEvent = appendedEvents.Last().Metadata.SequenceNumber;
+            lastSuccessfullyObservedEvent = appendedEvents[^1].Metadata.SequenceNumber;
         }
         catch (Exception ex)
         {
+            logger.ErrorWhileHandlingEvents(ex, appendedEvents[0].Context.SequenceNumber, appendedEvents[^1].Context.SequenceNumber, handler.Id);
             exceptionMessages = ex.GetAllMessages();
             exceptionStackTrace = ex.StackTrace ?? string.Empty;
             state = ObservationState.Failed;

diff --git a/Source/Clients/DotNET/Reducers/ReducersLogMessages.cs b/Source/Clients/DotNET/Reducers/ReducersLogMessages.cs
@@ -1,6 +1,7 @@
 // Copyright (c) Cratis. All rights reserved.
 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
 
+using Cratis.Chronicle.Events;
 using Cratis.Chronicle.EventSequences;
 using Microsoft.Extensions.Logging;
 
@@ -13,4 +14,7 @@ internal static partial class ReducersLogMessages
 
     [LoggerMessage(LogLevel.Trace, "Registering reducer with id '{ReducerId}', for event sequence '{EventSequenceId}'")]
     internal static partial void RegisterReducer(this ILogger<Reducers> logger, ReducerId reducerId, EventSequenceId eventSequenceId);
+
+    [LoggerMessage(LogLevel.Warning, "An error occurred while handling events with sequence number {StartSequenceNumber} to {EndSequenceNumber} was for Reducer {ReducerId}")]
+    internal static partial void ErrorWhileHandlingEvents(this ILogger<Reducers> logger, Exception ex, EventSequenceNumber startSequenceNumber, EventSequenceNumber endSequenceNumber, ReducerId reducerId);
 }
diff --git a/Source/Kernel/Grains/Jobs/Job.cs b/Source/Kernel/Grains/Jobs/Job.cs
@@ -109,13 +109,13 @@ public async Task Start(TRequest request)
         await WriteStateAsync();
 
         var grainId = this.GetGrainId();
-        var tcs = new TaskCompletionSource<IImmutableList<JobStepDetails>>();
+        var tcs = new TaskCompletionSource<IImmutableList<JobStepDetails>>(TaskCreationOptions.RunContinuationsAsynchronously);
 
         PrepareAllSteps(request, tcs);
 
 #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
         tcs.Task.ContinueWith(
-            async (Task<IImmutableList<JobStepDetails>> jobStepsTask) =>
+            async jobStepsTask =>
             {
                 var jobSteps = await jobStepsTask;
                 if (jobSteps.Count == 0)

diff --git a/Source/Kernel/Grains/Observation/Reactions/Clients/ReactorObserverSubscriber.cs b/Source/Kernel/Grains/Observation/Reactions/Clients/ReactorObserverSubscriber.cs
@@ -64,7 +64,7 @@ public async Task<ObserverSubscriberResult> OnNext(IEnumerable<AppendedEvent> ev
         {
             throw new MissingStateForReactorSubscriber(_observerId);
         }
-        var tcs = new TaskCompletionSource<ObserverSubscriberResult>();
+        var tcs = new TaskCompletionSource<ObserverSubscriberResult>(TaskCreationOptions.RunContinuationsAsynchronously);
         try
         {
             reactorMediator.OnNext(

diff --git a/Source/Kernel/Grains/Observation/Reducers/Clients/ReducerObserverSubscriber.cs b/Source/Kernel/Grains/Observation/Reducers/Clients/ReducerObserverSubscriber.cs
@@ -69,7 +69,7 @@ public async Task<ObserverSubscriberResult> OnNext(IEnumerable<AppendedEvent> ev
             throw new MissingStateForReducerSubscriber(_observerId);
         }
 
-        var tcs = new TaskCompletionSource<ObserverSubscriberResult>();
+        var tcs = new TaskCompletionSource<ObserverSubscriberResult>(TaskCreationOptions.RunContinuationsAsynchronously);
         try
         {
             var firstEvent = events.First();
@@ -79,7 +79,7 @@ public async Task<ObserverSubscriberResult> OnNext(IEnumerable<AppendedEvent> ev
 
             await (_pipeline?.Handle(reducerContext, async (events, initialState) =>
             {
-                var reducerSubscriberResultTCS = new TaskCompletionSource<ReducerSubscriberResult>();
+                var reducerSubscriberResultTCS = new TaskCompletionSource<ReducerSubscriberResult>(TaskCreationOptions.RunContinuationsAsynchronously);
 
                 reducerMediator.OnNext(
                     _observerId,

diff --git a/Source/Kernel/Services/Observation/Reactions/Reactors.cs b/Source/Kernel/Services/Observation/Reactions/Reactors.cs
@@ -29,7 +29,7 @@ public class Reactors(
     /// <inheritdoc/>
     public IObservable<EventsToObserve> Observe(IObservable<ReactorMessage> messages, CallContext context = default)
     {
-        var registrationTcs = new TaskCompletionSource<RegisterReactor>();
+        var registrationTcs = new TaskCompletionSource<RegisterReactor>(TaskCreationOptions.RunContinuationsAsynchronously);
         TaskCompletionSource<ObserverSubscriberResult>? observationResultTcs = null;
         TaskCompletionSource<IEnumerable<AppendedEvent>>? eventsTcs;
         IReactor clientObserver = null!;
@@ -83,7 +83,7 @@ public IObservable<EventsToObserve> Observe(IObservable<ReactorMessage> messages
                 connectionId = registration.ConnectionId;
                 observerId = registration.ObserverId;
 
-                eventsTcs = new TaskCompletionSource<IEnumerable<AppendedEvent>>();
+                eventsTcs = new TaskCompletionSource<IEnumerable<AppendedEvent>>(TaskCreationOptions.RunContinuationsAsynchronously);
                 reactorMediator.Subscribe(
                     registration.ObserverId,
                     registration.ConnectionId,
@@ -103,7 +103,7 @@ public IObservable<EventsToObserve> Observe(IObservable<ReactorMessage> messages
                         break;
                     }
 
-                    eventsTcs = new TaskCompletionSource<IEnumerable<AppendedEvent>>();
+                    eventsTcs = new TaskCompletionSource<IEnumerable<AppendedEvent>>(TaskCreationOptions.RunContinuationsAsynchronously);
                 }
             }
             catch (OperationCanceledException)

diff --git a/Source/Kernel/Services/Observation/Reducers/Reducers.cs b/Source/Kernel/Services/Observation/Reducers/Reducers.cs
@@ -36,7 +36,7 @@ public class Reducers(
     /// <inheritdoc/>
     public IObservable<ReduceOperationMessage> Observe(IObservable<ReducerMessage> messages, CallContext context = default)
     {
-        var registrationTcs = new TaskCompletionSource<RegisterReducer>();
+        var registrationTcs = new TaskCompletionSource<RegisterReducer>(TaskCreationOptions.RunContinuationsAsynchronously);
         TaskCompletionSource<ReducerSubscriberResult>? reducerResultTcs = null;
         TaskCompletionSource<ReduceOperation>? reduceOperationTcs;
 
@@ -103,7 +103,7 @@ public IObservable<ReduceOperationMessage> Observe(IObservable<ReducerMessage> m
                 connectionId = registration.ConnectionId;
                 observerId = registration.Reducer.ReducerId;
 
-                reduceOperationTcs = new TaskCompletionSource<ReduceOperation>();
+                reduceOperationTcs = new TaskCompletionSource<ReduceOperation>(TaskCreationOptions.RunContinuationsAsynchronously);
                 reducerMediator.Subscribe(
                     registration.Reducer.ReducerId,
                     registration.ConnectionId,
@@ -129,7 +129,7 @@ public IObservable<ReduceOperationMessage> Observe(IObservable<ReducerMessage> m
                         break;
                     }
 
-                    reduceOperationTcs = new TaskCompletionSource<ReduceOperation>();
+                    reduceOperationTcs = new TaskCompletionSource<ReduceOperation>(TaskCreationOptions.RunContinuationsAsynchronously);
                 }
             }
             catch (OperationCanceledException)