Improved control over aggregate operations timing (#229)

* Remove duplicate tracing on event handling

* Increased request deduplication for Aggregate cluster actors, to support very long aggregates being rehydrated.

* Added the ability to set a default aggregate perform timeout, which will be used if the user does not provide a CancellationToken. Not enabled by default

Author: Magne Helleborg

Source/Aggregates/Actors/AggregateClusterKindFactory.cs

@@ -30,6 +30,6 @@ public static ClusterKind CreateKind(IServiceProvider serviceProvider)
         var idleUnloadTimeout = serviceProvider.GetRequiredService<AggregateUnloadTimeout>()();
 
         return new ClusterKind(aggregateRootType.Id.Value.ToString(),
-            Props.FromProducer(() => new AggregateActor<TAggregate>(providerForTenant, logger, idleUnloadTimeout)).WithClusterRequestDeduplication());
+            Props.FromProducer(() => new AggregateActor<TAggregate>(providerForTenant, logger, idleUnloadTimeout)).WithClusterRequestDeduplication(TimeSpan.FromMinutes(5)));
     }
 }

Source/Aggregates/AggregateRootOperations.cs

@@ -14,6 +14,12 @@
 
 namespace Dolittle.SDK.Aggregates;
 
+/// <summary>
+/// If no CancellationToken is used, this delegate will be called to use the default timeout.
+/// </summary>
+public delegate CancellationToken DefaultAggregatePerformTimeout();
+
+
 /// <summary>
 /// Represents an implementation of <see cref="IAggregateRootOperations{T}"/>.
 /// </summary>
@@ -23,6 +29,7 @@ public class AggregateRootOperations<TAggregate> : IAggregateRootOperations<TAgg
 {
     readonly EventSourceId _eventSourceId;
     readonly IRootContext _context;
+    readonly DefaultAggregatePerformTimeout _defaultTimeout;
     readonly ClusterIdentity _clusterIdentity;
 
     /// <summary>
@@ -31,9 +38,11 @@ public class AggregateRootOperations<TAggregate> : IAggregateRootOperations<TAgg
     /// <param name="eventSourceId">The <see cref="EventSourceId"/> of the aggregate root instance.</param>
     /// <param name="tenantId">The <see cref="TenantId"/> of the current tenant.</param>
     /// <param name="context">The <see cref="IRootContext" />Root context used to communicate with actors</param>
-    public AggregateRootOperations(EventSourceId eventSourceId, TenantId tenantId, IRootContext context)
+    /// <param name="defaultTimeout">The &lt;see cref="DefaultAggregatePerformTimeout" /&gt; Used if no cancellation token is passed.</param>
+    public AggregateRootOperations(EventSourceId eventSourceId, TenantId tenantId, IRootContext context, DefaultAggregatePerformTimeout defaultTimeout)
     {
         _context = context;
+        _defaultTimeout = defaultTimeout;
         _eventSourceId = eventSourceId;
         _clusterIdentity = ClusterIdentityMapper.GetClusterIdentity<TAggregate>(tenantId, eventSourceId);
     }
@@ -51,6 +60,10 @@ public Task Perform(Action<TAggregate> method, CancellationToken cancellationTok
     /// <inheritdoc/>
     public async Task Perform(Func<TAggregate, Task> method, CancellationToken cancellationToken = default)
     {
+        if(cancellationToken == default)
+        {
+            cancellationToken = _defaultTimeout();
+        }
         using var activity = Tracing.ActivitySource.StartActivity($"{typeof(TAggregate).Name}.Perform")
             ?.Tag(_eventSourceId);
 

Source/Events.Handling/EventHandler.cs

@@ -76,8 +76,7 @@ public EventHandler(
     public async Task Handle(object @event, EventType eventType, EventContext context, IServiceProvider serviceProvider, CancellationToken cancellation)
     {
         var time = Stopwatch.StartNew();
-        using var activity = @event is not HandleEventRequest ? context.CommittedExecutionContext.StartChildActivity($"{_activityName}{@event.GetType().Name}")
-            ?.Tag(eventType) : null;
+        using var activity = context.CommittedExecutionContext.StartChildActivity($"{_activityName}{@event.GetType().Name}");
 
         try
         {

Source/SDK/Builders/DolittleClientConfiguration.cs

@@ -42,7 +42,13 @@ public class DolittleClientConfiguration : IConfigurationBuilder
     /// <summary>
     /// How long should the aggregates be kept in memory when not in use.
     /// </summary>
-    public TimeSpan AggregateIdleTimeout { get; private set; } = TimeSpan.FromSeconds(20);
+    public TimeSpan AggregateIdleTimeout { get; private set; } = TimeSpan.FromSeconds(30);
+    
+    /// <summary>
+    /// The default timeout for performing an aggregate operation (potentially including hydration & commits).
+    /// If no CancellationToken is provided, this timeout will be used.
+    /// </summary>
+    public TimeSpan? DefaultAggregatePerformTimeout { get; private set; }
 
     /// <summary>
     /// Gets the event serializer provider.
@@ -195,6 +201,13 @@ public IConfigurationBuilder WithAggregateIdleTimout(TimeSpan timeout)
         AggregateIdleTimeout = timeout;
         return this;
     }
+    
+    /// <inheritdoc />
+    public IConfigurationBuilder WithDefaultAggregatePerformTimeout(TimeSpan timeout)
+    {
+        DefaultAggregatePerformTimeout = timeout;
+        return this;
+    }
 
     /// <inheritdoc />
     public IConfigurationBuilder WithServiceProvider(IServiceProvider serviceProvider)

Source/SDK/Builders/IConfigurationBuilder.cs

@@ -68,6 +68,15 @@ public interface IConfigurationBuilder
     /// <returns></returns>
     public IConfigurationBuilder WithAggregateIdleTimout(TimeSpan timeout);
 
+    /// <summary>
+    /// Sets the default timeout for performing an aggregate operation (potentially including hydration & commits).
+    /// If no CancellationToken is provided, this timeout will be used.
+    /// Not set by default
+    /// </summary>
+    /// <param name="timeout"></param>
+    /// <returns></returns>
+    public IConfigurationBuilder WithDefaultAggregatePerformTimeout(TimeSpan timeout);
+    
     /// <summary>
     /// Configures the root <see cref="IServiceProvider"/> for the <see cref="IDolittleClient"/>.
     /// </summary>

Source/SDK/Proto/ServiceCollectionExtensions.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Generic;
 using System.Collections.Immutable;
+using Dolittle.SDK.Aggregates;
 using Dolittle.SDK.Aggregates.Actors;
 using Dolittle.SDK.Builders;
 using Microsoft.Extensions.DependencyInjection;
@@ -40,6 +41,21 @@ public static IServiceCollection AddProtoInfra(this IServiceCollection services,
                 var timeout = dolittleClientConfiguration.AggregateIdleTimeout;
 
                 return () => timeout;
+            })
+            .AddSingleton<DefaultAggregatePerformTimeout>(sp =>
+            {
+                var dolittleClientConfiguration = sp.GetRequiredService<DolittleClientConfiguration>();
+                var timeout = dolittleClientConfiguration.DefaultAggregatePerformTimeout;
+                if (timeout is not null)
+                {
+                    var seconds = (int)Math.Ceiling(timeout.Value.TotalSeconds);
+                    if (seconds > 0)
+                    {
+                        return () => CancellationTokens.FromSeconds(seconds);
+                    }
+                }
+
+                return () => default;
             });
     }