Add caching to workflow runtime and workflow management stores

Directory.Packages.props

@@ -82,6 +82,7 @@
         <PackageVersion Include="Quartz.Serialization.Json" Version="3.8.0"/>
         <PackageVersion Include="Refit" Version="7.0.0"/>
         <PackageVersion Include="Refit.HttpClientFactory" Version="7.0.0"/>
+        <PackageVersion Include="Scrutor" Version="4.2.2"/>
         <PackageVersion Include="ShortGuid" Version="2.0.1"/>
         <PackageVersion Include="StackExchange.Redis" Version="2.7.4"/>
         <PackageVersion Include="System.CommandLine" Version="2.0.0-beta4.22272.1"/>

Elsa.sln

@@ -332,6 +332,12 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Workflows.Runtime.Unit
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Server.LoadBalancer", "src\bundles\Elsa.Server.LoadBalancer\Elsa.Server.LoadBalancer.csproj", "{7C1E6DD9-C7DE-4686-AA73-D67A15D005A1}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Caching", "src\modules\Elsa.Caching\Elsa.Caching.csproj", "{A1FE2CA0-8EF3-4EDB-9055-DD282E374E16}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "caching", "caching", "{EB3A7401-0DE3-476F-9E6F-057F1F4590FB}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Caching.Distributed.MassTransit", "src\modules\Elsa.Caching.Distributed.MassTransit\Elsa.Caching.Distributed.MassTransit.csproj", "{BF4CDE28-F0CC-482D-8392-C1C9BB4C368F}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -804,6 +810,14 @@ Global
 		{7C1E6DD9-C7DE-4686-AA73-D67A15D005A1}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{7C1E6DD9-C7DE-4686-AA73-D67A15D005A1}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{7C1E6DD9-C7DE-4686-AA73-D67A15D005A1}.Release|Any CPU.Build.0 = Release|Any CPU
+		{A1FE2CA0-8EF3-4EDB-9055-DD282E374E16}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{A1FE2CA0-8EF3-4EDB-9055-DD282E374E16}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{A1FE2CA0-8EF3-4EDB-9055-DD282E374E16}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{A1FE2CA0-8EF3-4EDB-9055-DD282E374E16}.Release|Any CPU.Build.0 = Release|Any CPU
+		{BF4CDE28-F0CC-482D-8392-C1C9BB4C368F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{BF4CDE28-F0CC-482D-8392-C1C9BB4C368F}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{BF4CDE28-F0CC-482D-8392-C1C9BB4C368F}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{BF4CDE28-F0CC-482D-8392-C1C9BB4C368F}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -945,6 +959,9 @@ Global
 		{086A7E2A-6CE7-41DA-927B-C033638060B1} = {56C2FFB8-EA54-45B5-A095-4A78142EB4B5}
 		{B0946844-DE0E-4C59-9391-C0EB649DBF4E} = {18453B51-25EB-4317-A4B3-B10518252E92}
 		{7C1E6DD9-C7DE-4686-AA73-D67A15D005A1} = {F06B9573-DF68-4606-866C-A7546A10A05A}
+		{EB3A7401-0DE3-476F-9E6F-057F1F4590FB} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79}
+		{A1FE2CA0-8EF3-4EDB-9055-DD282E374E16} = {EB3A7401-0DE3-476F-9E6F-057F1F4590FB}
+		{BF4CDE28-F0CC-482D-8392-C1C9BB4C368F} = {EB3A7401-0DE3-476F-9E6F-057F1F4590FB}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {D4B5CEAA-7D70-4FCB-A68E-B03FBE5E0E5E}

samples/aspnet/Elsa.Samples.AspNet.ChildWorkflows/Workflows/ParentWorkflow.cs

@@ -9,6 +9,7 @@ public class ParentWorkflow : WorkflowBase
 {
     protected override void Build(IWorkflowBuilder builder)
     {
+        builder.WithDefinitionId("CustomId");
         var childOutput = builder.WithVariable<IDictionary<string, object>>();
 
         builder.Root = new Sequence

src/bundles/Elsa.Server.Web/Activities/SlowActivity.cs

@@ -23,7 +23,7 @@ public class SlowActivity : CodeActivity, IActivityPropertyDefaultValueProvider
     protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
     {
         var delay = Delay.Get(context);
-        await Task.Delay(delay);
+        await Task.Delay(delay, context.CancellationToken);
     }
 
     object IActivityPropertyDefaultValueProvider.GetDefaultValue(PropertyInfo property)

src/bundles/Elsa.Server.Web/Elsa.Server.Web.csproj

@@ -8,6 +8,7 @@
     </PropertyGroup>
 
     <ItemGroup>
+        <ProjectReference Include="..\..\modules\Elsa.Caching.Distributed.MassTransit\Elsa.Caching.Distributed.MassTransit.csproj" />
         <ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.PostgreSql\Elsa.EntityFrameworkCore.PostgreSql.csproj"/>
         <ProjectReference Include="..\..\modules\Elsa.MassTransit.AzureServiceBus\Elsa.MassTransit.AzureServiceBus.csproj"/>
         <ProjectReference Include="..\Elsa\Elsa.csproj"/>

src/bundles/Elsa.Server.Web/Enums/DistributedCachingTransport.cs

@@ -0,0 +1,11 @@
+namespace Elsa.Server.Web;
+
+/// <summary>
+/// Represents the transport options for distributed caching.
+/// </summary>
+public enum DistributedCachingTransport
+{
+    Memory,
+    Redis,
+    MassTransit
+}

src/bundles/Elsa.Server.Web/Program.cs

@@ -19,7 +19,7 @@
 using Elsa.MongoDb.Modules.Runtime;
 using Elsa.Server.Web;
 using Elsa.Workflows.Management.Compression;
-using Elsa.Workflows.Management.Services;
+using Elsa.Workflows.Management.Stores;
 using Elsa.Workflows.Runtime.Stores;
 using Medallion.Threading.FileSystem;
 using Medallion.Threading.Postgres;
@@ -39,9 +39,12 @@
 const bool useQuartz = true;
 const bool useMassTransit = true;
 const bool useZipCompression = true;
-const MassTransitBroker useMassTransitBroker = MassTransitBroker.Memory;
 const bool runEFCoreMigrations = true;
-const bool useMemoryStores = false;
+const bool useMemoryStores = true;
+const bool useCachingStores = true;
+const bool useDistributedCaching = true;
+const DistributedCachingTransport distributedCachingTransport = DistributedCachingTransport.MassTransit;
+const MassTransitBroker useMassTransitBroker = MassTransitBroker.Memory;
 
 var builder = WebApplication.CreateBuilder(args);
 var services = builder.Services;
@@ -149,6 +152,9 @@
 
                 if (useMassTransit)
                     management.UseMassTransitDispatcher();
+
+                if (useCachingStores)
+                    management.UseCachingStores();
             })
             .UseWorkflowRuntime(runtime =>
             {
@@ -183,9 +189,7 @@
                 }
 
                 if (useMassTransit)
-                {
                     runtime.UseMassTransitDispatcher();
-                }
 
                 runtime.WorkflowInboxCleanupOptions = options => configuration.GetSection("Runtime:WorkflowInboxCleanup").Bind(options);
                 runtime.WorkflowDispatcherOptions = options => configuration.GetSection("Runtime:WorkflowDispatcher").Bind(options);
@@ -197,6 +201,9 @@
                     runtime.WorkflowInboxStore = sp => sp.GetRequiredService<MemoryWorkflowInboxMessageStore>();
                 }
 
+                if (useCachingStores)
+                    runtime.UseCachingStores();
+
                 runtime.DistributedLockProvider = _ =>
                 {
                     switch (distributedLockProviderName)
@@ -330,6 +337,14 @@
             });
         }
 
+        if (useDistributedCaching)
+        {
+            elsa.UseDistributedCache(distributedCaching =>
+            {
+                if (distributedCachingTransport == DistributedCachingTransport.MassTransit) distributedCaching.UseMassTransit();
+            });
+        }
+
         elsa.InstallDropIns(options => options.DropInRootDirectory = Path.Combine(Directory.GetCurrentDirectory(), "App_Data", "DropIns"));
         elsa.AddSwagger();
         elsa.AddFastEndpointsAssembly<Program>();

src/bundles/Elsa.ServerAndStudio.Web/Program.cs

@@ -46,7 +46,7 @@
                 identity.UseConfigurationBasedRoleProvider(options => identitySection.Bind(options));
             })
             .UseDefaultAuthentication()
-            .UseInstanceManagement(x => x.HeartbeatOptions = settings => heartbeatSection.Bind(settings))
+            .UseApplicationCluster(x => x.HeartbeatOptions = settings => heartbeatSection.Bind(settings))
             .UseWorkflowManagement(management =>
             {
                 if (useMassTransit)

src/common/Elsa.Testing.Shared/ServiceProviderExtensions.cs

@@ -140,7 +140,7 @@ public static async Task<RunWorkflowResult> RunActivityAsync(this IServiceProvid
     public static async Task<WorkflowDefinition> GetWorkflowDefinitionAsync(this IServiceProvider services, string workflowDefinitionId, VersionOptions versionOptions, CancellationToken cancellationToken = default)
     {
         var workflowDefinitionService = services.GetRequiredService<IWorkflowDefinitionService>();
-        var workflowDefinition = await workflowDefinitionService.FindAsync(workflowDefinitionId, versionOptions);
+        var workflowDefinition = await workflowDefinitionService.FindWorkflowDefinitionAsync(workflowDefinitionId, versionOptions);
         return workflowDefinition!;
     }
 }

src/modules/Elsa.Alterations/AlterationHandlers/MigrateHandler.cs

@@ -21,7 +21,7 @@ protected override async ValueTask HandleAsync(AlterationContext context, Migrat
         var definitionId = context.Workflow.Identity.DefinitionId;
         var targetVersion = alteration.TargetVersion;
         var cancellationToken = context.CancellationToken;
-        var targetWorkflowDefinition = await workflowDefinitionService.FindAsync(definitionId, VersionOptions.SpecificVersion(targetVersion), cancellationToken);
+        var targetWorkflowDefinition = await workflowDefinitionService.FindWorkflowDefinitionAsync(definitionId, VersionOptions.SpecificVersion(targetVersion), cancellationToken);
 
         if (targetWorkflowDefinition == null)
         {

src/modules/Elsa.Alterations/Services/DefaultAlterationRunner.cs

@@ -1,10 +1,8 @@
-using Elsa.Alterations.Core.Contexts;
 using Elsa.Alterations.Core.Contracts;
 using Elsa.Alterations.Core.Models;
 using Elsa.Alterations.Core.Results;
 using Elsa.Alterations.Middleware.Workflows;
 using Elsa.Common.Contracts;
-using Elsa.Extensions;
 using Elsa.Workflows;
 using Elsa.Workflows.Contracts;
 using Elsa.Workflows.Management.Contracts;
@@ -75,18 +73,15 @@ public async Task<RunAlterationsResult> RunAsync(string workflowInstanceId, IEnu
         }
 
         // Load workflow definition.
-        var workflowDefinition = await _workflowDefinitionService.FindAsync(workflowState.DefinitionVersionId, cancellationToken);
+        var workflow = await _workflowDefinitionService.FindWorkflowAsync(workflowState.DefinitionVersionId, cancellationToken);
 
         // If the workflow definition is not found, log an error and continue.
-        if (workflowDefinition == null)
+        if (workflow == null)
         {
             log.Add($"Workflow definition with ID '{workflowState.DefinitionVersionId}' not found.", LogLevel.Error);
             return result;
         }
-
-        // Materialize workflow.
-        var workflow = await _workflowDefinitionService.MaterializeWorkflowAsync(workflowDefinition, cancellationToken);
-
+
         // Create workflow execution context.
         var workflowExecutionContext = await WorkflowExecutionContext.CreateAsync(_serviceProvider, workflow, workflowState, cancellationTokens: cancellationToken);
         workflowExecutionContext.TransientProperties.Add(RunAlterationsMiddleware.AlterationsPropertyKey, alterations);

src/modules/Elsa.Caching.Distributed.MassTransit/Consumers/TriggerChangeTokenSignalConsumer.cs

@@ -0,0 +1,21 @@
+using Elsa.Caching.Contracts;
+using Elsa.Caching.Distributed.MassTransit.Messages;
+using JetBrains.Annotations;
+using MassTransit;
+
+namespace Elsa.Caching.Distributed.MassTransit.Consumers;
+
+/// <summary>
+/// Consumes <see cref="TriggerChangeTokenSignal"/> messages and triggers the change token signal.
+/// </summary>
+[UsedImplicitly]
+public class TriggerChangeTokenSignalConsumer(IDistributedChangeTokenSignaler changeTokenSignaler) : IConsumer<TriggerChangeTokenSignal>
+{
+    /// <inheritdoc />
+    public async Task Consume(ConsumeContext<TriggerChangeTokenSignal> context)
+    {
+        var message = context.Message;
+        var cancellationToken = context.CancellationToken;
+        await changeTokenSignaler.TriggerTokenLocalAsync(message.Key, cancellationToken);
+    }
+}

src/modules/Elsa.Caching.Distributed.MassTransit/Elsa.Caching.Distributed.MassTransit.csproj

@@ -0,0 +1,16 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <Description>
+            Provides distributed caching services leveraging MassTransit for transport of signals.
+        </Description>
+        <PackageTags>elsa module distributed caching mass-transit</PackageTags>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\..\common\Elsa.Features\Elsa.Features.csproj" />
+        <ProjectReference Include="..\Elsa.Caching\Elsa.Caching.csproj" />
+        <ProjectReference Include="..\Elsa.MassTransit\Elsa.MassTransit.csproj" />
+    </ItemGroup>
+
+</Project>

src/modules/Elsa.Caching.Distributed.MassTransit/Extensions/ModuleExtensions.cs

@@ -0,0 +1,28 @@
+using Elsa.Caching.Distributed.MassTransit.Features;
+using Elsa.Caching.Features;
+
+// ReSharper disable once CheckNamespace
+namespace Elsa.Extensions;
+
+/// <summary>
+/// Provides methods to install and configure the distributed caching feature with MassTransit.
+/// </summary>
+public static class ModuleExtensions
+{
+    /// <summary>
+    /// Configures the distributed caching feature to use MassTransit.
+    /// </summary>
+    public static MassTransitDistributedCacheFeature UseMassTransit(this DistributedCacheFeature distributedCacheFeature, Action<MassTransitDistributedCacheFeature>? configure = default)
+    {
+        return distributedCacheFeature.Module.Configure(configure);
+    }
+
+    /// <summary>
+    /// Configures the memory caching feature with the distributed caching feature that uses MassTransit.
+    /// </summary>
+    public static MemoryCacheFeature UseMassTransit(this MemoryCacheFeature memoryCacheFeature, Action<MassTransitDistributedCacheFeature>? configure = default)
+    {
+        memoryCacheFeature.Module.Configure(configure);
+        return memoryCacheFeature;
+    }
+}

src/modules/Elsa.Caching.Distributed.MassTransit/Features/MassTransitDistributedCacheFeature.cs

@@ -0,0 +1,32 @@
+using Elsa.Caching.Distributed.MassTransit.Consumers;
+using Elsa.Caching.Distributed.MassTransit.Services;
+using Elsa.Caching.Features;
+using Elsa.Extensions;
+using Elsa.Features.Abstractions;
+using Elsa.Features.Attributes;
+using Elsa.Features.Services;
+using Elsa.MassTransit.Features;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Elsa.Caching.Distributed.MassTransit.Features;
+
+/// <summary>
+/// Configures distributed cache management with MassTransit.
+/// </summary>
+[DependsOn(typeof(DistributedCacheFeature))]
+[DependsOn(typeof(MassTransitFeature))]
+public class MassTransitDistributedCacheFeature(IModule module) : FeatureBase(module)
+{
+    /// <inheritdoc />
+    public override void Configure()
+    {
+        Module.AddMassTransitConsumer<TriggerChangeTokenSignalConsumer>("elsa-trigger-change-token-signal", true);
+        Module.Use<DistributedCacheFeature>(feature => feature.WithChangeTokenSignalPublisher(sp => sp.GetRequiredService<MassTransitChangeTokenSignalPublisher>()));
+    }
+
+    /// <inheritdoc />
+    public override void Apply()
+    {
+        Services.AddSingleton<MassTransitChangeTokenSignalPublisher>();
+    }
+}

src/modules/Elsa.Caching.Distributed.MassTransit/FodyWeavers.xml

@@ -0,0 +1,3 @@
+<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd">
+  <ConfigureAwait />
+</Weavers>

src/modules/Elsa.Caching.Distributed.MassTransit/Messages/TriggerChangeTokenSignal.cs

@@ -0,0 +1,7 @@
+namespace Elsa.Caching.Distributed.MassTransit.Messages;
+
+/// <summary>
+/// Represents a message containing a signal to trigger a change token.
+/// </summary>
+/// <param name="Key">The key of the change token to trigger.</param>
+public record TriggerChangeTokenSignal(string Key);

src/modules/Elsa.Caching.Distributed.MassTransit/Services/MassTransitChangeTokenSignalPublisher.cs

@@ -0,0 +1,18 @@
+using Elsa.Caching.Contracts;
+using Elsa.Caching.Distributed.MassTransit.Messages;
+using MassTransit;
+
+namespace Elsa.Caching.Distributed.MassTransit.Services;
+
+/// <summary>
+/// Represents a service that publishes change token signals using MassTransit.
+/// </summary>
+public class MassTransitChangeTokenSignalPublisher(IBus bus) : IChangeTokenSignalPublisher
+{
+    /// <inheritdoc />
+    public async ValueTask PublishAsync(string key, CancellationToken cancellationToken = default)
+    {
+        var message = new TriggerChangeTokenSignal(key);
+        await bus.Publish(message, cancellationToken);
+    }
+}

src/modules/Elsa.Caching/Contracts/IChangeTokenSignalPublisher.cs

@@ -0,0 +1,12 @@
+namespace Elsa.Caching.Contracts;
+
+/// <summary>
+/// Represents a service that can publish change token signals.
+/// </summary>
+public interface IChangeTokenSignalPublisher
+{
+    /// <summary>
+    /// Publishes a change token signal for the specified key.
+    /// </summary>
+    ValueTask PublishAsync(string key, CancellationToken cancellationToken = default);
+}

src/modules/Elsa.Caching/Contracts/IChangeTokenSignaler.cs

@@ -0,0 +1,19 @@
+using Microsoft.Extensions.Primitives;
+
+namespace Elsa.Caching.Contracts;
+
+/// <summary>
+/// Provides change tokens for memory caches, allowing code to evict cache entries by triggering a signal.
+/// </summary>
+public interface IChangeTokenSignaler
+{
+    /// <summary>
+    /// Gets a change token for the specified key.
+    /// </summary>
+    IChangeToken GetToken(string key);
+
+    /// <summary>
+    /// Triggers the change token for the specified key.
+    /// </summary>
+    ValueTask TriggerTokenAsync(string key, CancellationToken cancellationToken = default);
+}