configure Redis cache and distributed locks

This commit is contained in:
Behzad Alizadeh
2026-10-10 16:50:37 +03:30
parent 3f75fec42b
commit 2676bf6ffd
155 changed files with 1784 additions and 229 deletions
+1
View File
@@ -151,6 +151,7 @@
<PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="$(MicrosoftVersion)" /> <PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="$(MicrosoftVersion)" />
<PackageVersion Include="Microsoft.Extensions.Configuration" Version="$(MicrosoftVersion)" /> <PackageVersion Include="Microsoft.Extensions.Configuration" Version="$(MicrosoftVersion)" />
<PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="$(MicrosoftVersion)" /> <PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="$(MicrosoftVersion)" />
<PackageVersion Include="Microsoft.Extensions.Configuration.Binder" Version="$(MicrosoftVersion)" />
<PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="$(MicrosoftVersion)" /> <PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="$(MicrosoftVersion)" />
<PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="$(MicrosoftVersion)" /> <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="$(MicrosoftVersion)" />
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="$(MicrosoftVersion)" /> <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="$(MicrosoftVersion)" />
+21
View File
@@ -419,6 +419,12 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Caspian.WorkflowFramework",
EndProject EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Caspian.WorkflowFramework.Sample", "src\apps\Caspian.WorkflowFramework.Sample\Caspian.WorkflowFramework.Sample.csproj", "{C260D75F-19EB-4982-A98B-D8E126776910}" Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Caspian.WorkflowFramework.Sample", "src\apps\Caspian.WorkflowFramework.Sample\Caspian.WorkflowFramework.Sample.csproj", "{C260D75F-19EB-4982-A98B-D8E126776910}"
EndProject EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Caching.Redis", "src\modules\Elsa.Caching.Redis\Elsa.Caching.Redis.csproj", "{CEB64AF4-A63C-4204-828A-53E6096095A0}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Caching.UnitTests", "test\unit\Elsa.Caching.UnitTests\Elsa.Caching.UnitTests.csproj", "{86197750-6B7F-4F65-8456-E30E82F53F1C}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Caching.ProcessTests", "test\integration\Elsa.Caching.ProcessTests\Elsa.Caching.ProcessTests.csproj", "{0368A4B4-FA10-489D-B390-F141A6225720}"
EndProject
Global Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU Debug|Any CPU = Debug|Any CPU
@@ -875,6 +881,18 @@ Global
{C260D75F-19EB-4982-A98B-D8E126776910}.Debug|Any CPU.Build.0 = Debug|Any CPU {C260D75F-19EB-4982-A98B-D8E126776910}.Debug|Any CPU.Build.0 = Debug|Any CPU
{C260D75F-19EB-4982-A98B-D8E126776910}.Release|Any CPU.ActiveCfg = Release|Any CPU {C260D75F-19EB-4982-A98B-D8E126776910}.Release|Any CPU.ActiveCfg = Release|Any CPU
{C260D75F-19EB-4982-A98B-D8E126776910}.Release|Any CPU.Build.0 = Release|Any CPU {C260D75F-19EB-4982-A98B-D8E126776910}.Release|Any CPU.Build.0 = Release|Any CPU
{CEB64AF4-A63C-4204-828A-53E6096095A0}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{CEB64AF4-A63C-4204-828A-53E6096095A0}.Debug|Any CPU.Build.0 = Debug|Any CPU
{CEB64AF4-A63C-4204-828A-53E6096095A0}.Release|Any CPU.ActiveCfg = Release|Any CPU
{CEB64AF4-A63C-4204-828A-53E6096095A0}.Release|Any CPU.Build.0 = Release|Any CPU
{86197750-6B7F-4F65-8456-E30E82F53F1C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{86197750-6B7F-4F65-8456-E30E82F53F1C}.Debug|Any CPU.Build.0 = Debug|Any CPU
{86197750-6B7F-4F65-8456-E30E82F53F1C}.Release|Any CPU.ActiveCfg = Release|Any CPU
{86197750-6B7F-4F65-8456-E30E82F53F1C}.Release|Any CPU.Build.0 = Release|Any CPU
{0368A4B4-FA10-489D-B390-F141A6225720}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{0368A4B4-FA10-489D-B390-F141A6225720}.Debug|Any CPU.Build.0 = Debug|Any CPU
{0368A4B4-FA10-489D-B390-F141A6225720}.Release|Any CPU.ActiveCfg = Release|Any CPU
{0368A4B4-FA10-489D-B390-F141A6225720}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection EndGlobalSection
GlobalSection(SolutionProperties) = preSolution GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE HideSolutionNode = FALSE
@@ -1030,6 +1048,9 @@ Global
{C583AF05-D517-4B7F-8955-6B61500ED3D8} = {7FD1FD1E-5778-4065-AAA5-1F878129EF77} {C583AF05-D517-4B7F-8955-6B61500ED3D8} = {7FD1FD1E-5778-4065-AAA5-1F878129EF77}
{A0F413B7-FA0F-472D-8BC2-6193DA440072} = {D92BEAB2-60D6-4BB4-885A-6BA681C6CCF1} {A0F413B7-FA0F-472D-8BC2-6193DA440072} = {D92BEAB2-60D6-4BB4-885A-6BA681C6CCF1}
{C260D75F-19EB-4982-A98B-D8E126776910} = {D92BEAB2-60D6-4BB4-885A-6BA681C6CCF1} {C260D75F-19EB-4982-A98B-D8E126776910} = {D92BEAB2-60D6-4BB4-885A-6BA681C6CCF1}
{CEB64AF4-A63C-4204-828A-53E6096095A0} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79}
{86197750-6B7F-4F65-8456-E30E82F53F1C} = {18453B51-25EB-4317-A4B3-B10518252E92}
{0368A4B4-FA10-489D-B390-F141A6225720} = {1B8D5897-902E-4632-8698-E89CAF3DDF54}
EndGlobalSection EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {D4B5CEAA-7D70-4FCB-A68E-B03FBE5E0E5E} SolutionGuid = {D4B5CEAA-7D70-4FCB-A68E-B03FBE5E0E5E}
BIN
View File
Binary file not shown.
+31
View File
@@ -0,0 +1,31 @@
param(
[Parameter(Mandatory = $true)]
[string] $RedisConnection
)
$cacheRepoRoot = [IO.Path]::GetFullPath((Join-Path $PSScriptRoot '..'))
$cachePreviousConnection = $env:REDIS_TEST_CONNECTION
$cacheValidationRoot = Join-Path $cacheRepoRoot 'artifacts/cache-validation'
New-Item -ItemType Directory -Path $cacheValidationRoot -Force | Out-Null
$env:REDIS_TEST_CONNECTION = $RedisConnection
try {
Push-Location $cacheRepoRoot
dotnet test test/unit/Elsa.Caching.UnitTests/Elsa.Caching.UnitTests.csproj -m:1 -p:BuildInParallel=false -p:GeneratePackageOnBuild=false
if ($LASTEXITCODE -ne 0) { throw 'Workflow cache tests failed.' }
dotnet build test/integration/Elsa.Caching.ProcessTests/Elsa.Caching.ProcessTests.csproj -m:1 -p:BuildInParallel=false -p:GeneratePackageOnBuild=false
if ($LASTEXITCODE -ne 0) { throw 'Workflow cache process test build failed.' }
$cacheTestRun = [Guid]::NewGuid().ToString('N')
$cacheDatabasePath = Join-Path $cacheValidationRoot ($cacheTestRun + '.db')
$cacheKeyPrefix = 'workflow-process-tests:' + $cacheTestRun + ':'
$cacheProcessProgram = Join-Path $cacheRepoRoot 'test/integration/Elsa.Caching.ProcessTests/bin/Debug/net9.0/Elsa.Caching.ProcessTests.dll'
foreach ($cacheProcessMode in @('write', 'read', 'invalidate', 'read-updated', 'pending', 'bypass')) {
dotnet $cacheProcessProgram $cacheDatabasePath $cacheKeyPrefix $cacheProcessMode
if ($LASTEXITCODE -ne 0) { throw "Independent process test failed: $cacheProcessMode" }
}
}
finally {
Pop-Location
$env:REDIS_TEST_CONNECTION = $cachePreviousConnection
}
@@ -0,0 +1,22 @@
namespace Elsa.Server.Web.Configuration;
/// <summary>Host connection settings, separate from the workflow cache policy.</summary>
public class RedisSettings
{
public bool AllowAdmin { get; set; }
public bool Ssl { get; set; }
public int ConnectTimeout { get; set; } = 5000;
public int ConnectRetry { get; set; } = 2;
public int SyncTimeout { get; set; } = 5000;
public int Database { get; set; } = 1;
public string? ServiceName { get; set; }
public string? User { get; set; }
public string? Password { get; set; }
public List<RedisHostSettings> Hosts { get; set; } = new();
}
public class RedisHostSettings
{
public string Host { get; set; } = string.Empty;
public int Port { get; set; } = 6379;
}
@@ -4,6 +4,8 @@
<ProjectReference Include="..\..\modules\Elsa.Agents.Persistence.EntityFrameworkCore.MySql\Elsa.Agents.Persistence.EntityFrameworkCore.MySql.csproj" Condition=" '$(TargetFramework)' != 'net9.0' " /> <ProjectReference Include="..\..\modules\Elsa.Agents.Persistence.EntityFrameworkCore.MySql\Elsa.Agents.Persistence.EntityFrameworkCore.MySql.csproj" Condition=" '$(TargetFramework)' != 'net9.0' " />
<ProjectReference Include="..\..\modules\Elsa.Caching.Distributed.MassTransit\Elsa.Caching.Distributed.MassTransit.csproj" /> <ProjectReference Include="..\..\modules\Elsa.Caching.Distributed.MassTransit\Elsa.Caching.Distributed.MassTransit.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Caching.Distributed.ProtoActor\Elsa.Caching.Distributed.ProtoActor.csproj" /> <ProjectReference Include="..\..\modules\Elsa.Caching.Distributed.ProtoActor\Elsa.Caching.Distributed.ProtoActor.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Caching.Redis\Elsa.Caching.Redis.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Caching\Elsa.Caching.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.MySql\Elsa.EntityFrameworkCore.MySql.csproj" Condition=" '$(TargetFramework)' != 'net9.0' " /> <ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.MySql\Elsa.EntityFrameworkCore.MySql.csproj" Condition=" '$(TargetFramework)' != 'net9.0' " />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.Oracle\Elsa.EntityFrameworkCore.Oracle.csproj" /> <ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.Oracle\Elsa.EntityFrameworkCore.Oracle.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.PostgreSql\Elsa.EntityFrameworkCore.PostgreSql.csproj" /> <ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.PostgreSql\Elsa.EntityFrameworkCore.PostgreSql.csproj" />
@@ -0,0 +1,43 @@
using Elsa.Server.Web.Configuration;
using Microsoft.Extensions.DependencyInjection.Extensions;
using StackExchange.Redis;
namespace Elsa.Server.Web.Extensions;
public static class RedisServiceCollectionExtensions
{
public static IServiceCollection AddWorkflowRedis(this IServiceCollection services, IConfiguration configuration)
{
services.TryAddSingleton<IConnectionMultiplexer>(_ => ConnectionMultiplexer.Connect(
CreateConfiguration(configuration.GetSection("Redis").Get<RedisSettings>() ?? new())));
return services;
}
public static ConfigurationOptions CreateConfiguration(RedisSettings settings)
{
var options = new ConfigurationOptions
{
AllowAdmin = settings.AllowAdmin,
Ssl = settings.Ssl,
AbortOnConnectFail = false,
ConnectTimeout = settings.ConnectTimeout,
ConnectRetry = settings.ConnectRetry,
SyncTimeout = settings.SyncTimeout,
DefaultDatabase = settings.Database,
User = string.IsNullOrWhiteSpace(settings.User) ? null : settings.User,
Password = string.IsNullOrWhiteSpace(settings.Password) ? null : settings.Password
};
foreach (var host in settings.Hosts.Where(x => !string.IsNullOrWhiteSpace(x.Host)))
{
if (host.Port is < 1 or > 65535) throw new InvalidOperationException("Redis host port must be between 1 and 65535.");
options.EndPoints.Add(host.Host, host.Port);
}
if (options.EndPoints.Count == 0) throw new InvalidOperationException("Redis:Hosts must contain a Redis or Sentinel endpoint.");
if (!string.IsNullOrWhiteSpace(settings.ServiceName))
{
options.ServiceName = settings.ServiceName;
options.TieBreaker = string.Empty;
}
return options;
}
}
@@ -0,0 +1,62 @@
using Elsa.Common.DistributedHosting.DistributedLocks;
using Elsa.Workflows.Runtime.Features;
using Medallion.Threading;
using Medallion.Threading.FileSystem;
using Medallion.Threading.Postgres;
using Medallion.Threading.Redis;
using StackExchange.Redis;
namespace Elsa.Server.Web.Extensions;
public static class WorkflowRuntimeFeatureExtensions
{
public static WorkflowRuntimeFeature UseConfiguredDistributedLocking(this WorkflowRuntimeFeature runtime, IConfiguration configuration)
{
var legacySettings = configuration.GetSection("Runtime:DistributedLocking");
var settings = configuration.GetSection("Workflow:Runtime:DistributedLocking");
runtime.DistributedLockingOptions = options =>
{
legacySettings.Bind(options);
settings.Bind(options);
};
var providerName = settings["Provider"] ?? legacySettings["Provider"];
runtime.DistributedLockProvider = sp =>
{
switch (providerName?.ToUpperInvariant())
{
case "REDIS":
{
var database = settings.GetValue<int?>("Database")
?? legacySettings.GetValue<int?>("Database")
?? configuration.GetValue<int?>("Redis:Database") ?? -1;
var prefix = settings["KeyPrefix"] ?? legacySettings["KeyPrefix"] ?? "nextbank:workflow:locks:";
if (database < -1) throw new InvalidOperationException("Workflow lock database must be -1 or a non-negative number.");
if (string.IsNullOrWhiteSpace(prefix)) throw new InvalidOperationException("Workflow lock KeyPrefix must not be empty.");
var connection = sp.GetRequiredService<IConnectionMultiplexer>();
return new PrefixedDistributedLockProvider(new RedisDistributedSynchronizationProvider(connection.GetDatabase(database)), prefix);
}
case "POSTGRES":
return new PostgresDistributedSynchronizationProvider(configuration.GetConnectionString("PostgreSql")!, options =>
{
options.KeepaliveCadence(TimeSpan.FromMinutes(5));
options.UseMultiplexing();
});
case "FILE":
return new FileDistributedSynchronizationProvider(new(Path.Combine(Directory.GetCurrentDirectory(), "App_Data", "locks")));
case null:
case "NOOP":
return new NoopDistributedSynchronizationProvider();
default:
throw new InvalidOperationException($"Unsupported workflow lock provider '{providerName}'. Configure Redis, File, Postgres or Noop.");
}
};
return runtime;
}
private sealed class PrefixedDistributedLockProvider(IDistributedLockProvider provider, string prefix) : IDistributedLockProvider
{
public IDistributedLock CreateLock(string name) => provider.CreateLock(prefix + name);
}
}
+5 -34
View File
@@ -3,7 +3,6 @@ using Elsa.Alterations.Extensions;
using Elsa.Alterations.MassTransit.Extensions; using Elsa.Alterations.MassTransit.Extensions;
using Elsa.Caching.Options; using Elsa.Caching.Options;
using Elsa.Common.Codecs; using Elsa.Common.Codecs;
using Elsa.Common.DistributedHosting.DistributedLocks;
using Elsa.Common.RecurringTasks; using Elsa.Common.RecurringTasks;
using Elsa.Common.Serialization; using Elsa.Common.Serialization;
using Elsa.Dapper.Extensions; using Elsa.Dapper.Extensions;
@@ -64,9 +63,6 @@ using Hangfire.PostgreSql.Factories;
using Hangfire.SqlServer; using Hangfire.SqlServer;
using Hangfire.Storage.SQLite; using Hangfire.Storage.SQLite;
using JetBrains.Annotations; using JetBrains.Annotations;
using Medallion.Threading.FileSystem;
using Medallion.Threading.Postgres;
using Medallion.Threading.Redis;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using OpenTelemetry.Metrics; using OpenTelemetry.Metrics;
using OpenTelemetry.Resources; using OpenTelemetry.Resources;
@@ -76,7 +72,6 @@ using Proto.Persistence.Sqlite;
using Proto.Persistence.SqlServer; using Proto.Persistence.SqlServer;
using Proto.Remote; using Proto.Remote;
using Proto.Remote.GrpcNet; using Proto.Remote.GrpcNet;
using StackExchange.Redis;
// ReSharper disable RedundantAssignment // ReSharper disable RedundantAssignment
const PersistenceProvider persistenceProvider = PersistenceProvider.EntityFrameworkCore; const PersistenceProvider persistenceProvider = PersistenceProvider.EntityFrameworkCore;
@@ -119,8 +114,6 @@ var cockroachDbConnectionString = configuration.GetConnectionString("CockroachDb
var mongoDbConnectionString = configuration.GetConnectionString("MongoDb")!; var mongoDbConnectionString = configuration.GetConnectionString("MongoDb")!;
var azureServiceBusConnectionString = configuration.GetConnectionString("AzureServiceBus")!; var azureServiceBusConnectionString = configuration.GetConnectionString("AzureServiceBus")!;
var rabbitMqConnectionString = configuration.GetConnectionString("RabbitMq")!; var rabbitMqConnectionString = configuration.GetConnectionString("RabbitMq")!;
var redisConnectionString = configuration.GetConnectionString("Redis")!;
var distributedLockProviderName = configuration.GetSection("Runtime:DistributedLocking")["Provider"];
var appRole = Enum.Parse<ApplicationRole>(configuration["AppRole"] ?? "Default"); var appRole = Enum.Parse<ApplicationRole>(configuration["AppRole"] ?? "Default");
var sqlDatabaseProvider = Enum.Parse<SqlDatabaseProvider>(configuration["DatabaseProvider"] ?? "Sqlite"); var sqlDatabaseProvider = Enum.Parse<SqlDatabaseProvider>(configuration["DatabaseProvider"] ?? "Sqlite");
@@ -164,10 +157,13 @@ if (useManualOtelInstrumentation)
} }
// Add Elsa services. // Add Elsa services.
services.AddWorkflowRedis(configuration);
services.AddScoped<IPropertyUIHandler, CustomCheckListOptionsProvider>(); services.AddScoped<IPropertyUIHandler, CustomCheckListOptionsProvider>();
services services
.AddElsa(elsa => .AddElsa(elsa =>
{ {
elsa.UseCaching(cache => cache.RegisterRedis());
if (persistenceProvider == PersistenceProvider.MongoDb) if (persistenceProvider == PersistenceProvider.MongoDb)
elsa.UseMongoDb(mongoDbConnectionString); elsa.UseMongoDb(mongoDbConnectionString);
@@ -324,7 +320,7 @@ services
management.UseMassTransitDispatcher(); management.UseMassTransitDispatcher();
if (useCaching) if (useCaching)
management.UseCache(); management.UseCache(configuration.GetSection("Workflow:Cache"));
management.SetDefaultLogPersistenceMode(LogPersistenceMode.Inherit); management.SetDefaultLogPersistenceMode(LogPersistenceMode.Inherit);
management.UseReadOnlyMode(useReadOnlyMode); management.UseReadOnlyMode(useReadOnlyMode);
@@ -394,31 +390,7 @@ services
if (useCaching) if (useCaching)
runtime.UseCache(); runtime.UseCache();
runtime.DistributedLockingOptions = options => configuration.GetSection("Runtime:DistributedLocking").Bind(options); runtime.UseConfiguredDistributedLocking(configuration);
runtime.DistributedLockProvider = _ =>
{
switch (distributedLockProviderName)
{
case "Postgres":
return new PostgresDistributedSynchronizationProvider(postgresConnectionString, options =>
{
options.KeepaliveCadence(TimeSpan.FromMinutes(5));
options.UseMultiplexing();
});
case "Redis":
{
var connectionMultiplexer = ConnectionMultiplexer.Connect(redisConnectionString);
var database = connectionMultiplexer.GetDatabase();
return new RedisDistributedSynchronizationProvider(database);
}
case "File":
return new FileDistributedSynchronizationProvider(new(Path.Combine(Directory.GetCurrentDirectory(), "App_Data", "locks")));
case "Noop":
default:
return new NoopDistributedSynchronizationProvider();
}
};
}) })
.UseEnvironments(environments => environments.EnvironmentsOptions = options => configuration.GetSection("Environments").Bind(options)) .UseEnvironments(environments => environments.EnvironmentsOptions = options => configuration.GetSection("Environments").Bind(options))
.UseScheduling(scheduling => .UseScheduling(scheduling =>
@@ -758,7 +730,6 @@ services.Configure<RecurringTaskOptions>(options =>
services.Configure<RuntimeOptions>(options => { options.InactivityThreshold = TimeSpan.FromSeconds(15); }); services.Configure<RuntimeOptions>(options => { options.InactivityThreshold = TimeSpan.FromSeconds(15); });
services.Configure<BookmarkQueuePurgeOptions>(options => options.Ttl = TimeSpan.FromSeconds(10)); services.Configure<BookmarkQueuePurgeOptions>(options => options.Ttl = TimeSpan.FromSeconds(10));
services.Configure<CachingOptions>(options => options.CacheDuration = TimeSpan.FromDays(1));
services.Configure<IncidentOptions>(options => options.DefaultIncidentStrategy = typeof(ContinueWithIncidentsStrategy)); services.Configure<IncidentOptions>(options => options.DefaultIncidentStrategy = typeof(ContinueWithIncidentsStrategy));
services.AddHealthChecks(); services.AddHealthChecks();
services.AddControllers(); services.AddControllers();
@@ -12,7 +12,7 @@
"commandName": "Project", "commandName": "Project",
"dotnetRunMessages": true, "dotnetRunMessages": true,
"launchBrowser": true, "launchBrowser": true,
"applicationUrl": "https://localhost:5002;http://localhost:5072", "applicationUrl": "https://localhost:5001;http://localhost:5072",
"environmentVariables": { "environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development" "ASPNETCORE_ENVIRONMENT": "Development"
} }
@@ -0,0 +1,31 @@
{
"Redis": {
"AllowAdmin": true,
"Ssl": false,
"ConnectTimeout": 5000,
"ConnectRetry": 2,
"Database": 1,
"ServiceName": "",
"Hosts": [
{ "Host": "localhost", "Port": 6379 }
]
},
"Workflow": {
"Cache": {
"Enabled": true,
"Provider": "Redis",
"Duration": "1.00:00:00",
"Redis": {
"Database": 1,
"KeyPrefix": "nextbank:workflow:cache:"
}
},
"Runtime": {
"DistributedLocking": {
"Provider": "Redis",
"Database": 1,
"KeyPrefix": "nextbank:workflow:locks:"
}
}
}
}
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Api.Client</PackageId> <PackageId>Caspian.Elsa.Api.Client</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides HTTP clients for the Elsa Workflows REST API. Provides HTTP clients for the Elsa Workflows REST API.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Api.Common</PackageId> <PackageId>Caspian.Elsa.Api.Common</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides common features to modules that expose API endpoints. Provides common features to modules that expose API endpoints.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.DropIns.Core</PackageId> <PackageId>Caspian.Elsa.DropIns.Core</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Reference this package when developing drop-ins for Elsa Workflows. Reference this package when developing drop-ins for Elsa Workflows.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.DropIns</PackageId> <PackageId>Caspian.Elsa.DropIns</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides drop-in features for Elsa workflows. Provides drop-in features for Elsa workflows.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Features</PackageId> <PackageId>Caspian.Elsa.Features</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides utilities to turn your class library into a module that provides features. Provides utilities to turn your class library into a module that provides features.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Mediator</PackageId> <PackageId>Caspian.Elsa.Mediator</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides a mediator service inspired by MediatR. Provides a mediator service inspired by MediatR.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Testing.Shared.Component</PackageId> <PackageId>Caspian.Elsa.Testing.Shared.Component</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides common utilities that are helpful when writing component tests against Elsa Workflows functionality. Provides common utilities that are helpful when writing component tests against Elsa Workflows functionality.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Testing.Shared.Integration</PackageId> <PackageId>Caspian.Elsa.Testing.Shared.Integration</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides common utilities that are helpful when writing integration tests against Elsa Workflows functionality. Provides common utilities that are helpful when writing integration tests against Elsa Workflows functionality.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Testing.Shared</PackageId> <PackageId>Caspian.Elsa.Testing.Shared</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides common utilities that are helpful when writing tests against Elsa Workflows functionality. Provides common utilities that are helpful when writing tests against Elsa Workflows functionality.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Alterations.Core</PackageId> <PackageId>Caspian.Elsa.Alterations.Core</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides core interfaces, models and services that support the alteration engine. Provides core interfaces, models and services that support the alteration engine.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Alterations.MassTransit</PackageId> <PackageId>Caspian.Elsa.Alterations.MassTransit</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides alteration background job runners using MassTransit. Provides alteration background job runners using MassTransit.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Alterations</PackageId> <PackageId>Caspian.Elsa.Alterations</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides alterations for workflow instances. Provides alterations for workflow instances.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.CSharp</PackageId> <PackageId>Caspian.Elsa.CSharp</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides a CSharp expression provider. Provides a CSharp expression provider.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Caching.Distributed.MassTransit</PackageId> <PackageId>Caspian.Elsa.Caching.Distributed.MassTransit</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides distributed caching services leveraging MassTransit for transport of signals. Provides distributed caching services leveraging MassTransit for transport of signals.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Caching.Distributed.ProtoActor</PackageId> <PackageId>Caspian.Elsa.Caching.Distributed.ProtoActor</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides distributed caching services leveraging Proto.Actor for transport of signals. Provides distributed caching services leveraging Proto.Actor for transport of signals.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Caching.Distributed</PackageId> <PackageId>Caspian.Elsa.Caching.Distributed</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides distributed caching services and utility functions common to Elsa modules. Provides distributed caching services and utility functions common to Elsa modules.
@@ -0,0 +1,21 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<PackageId>Caspian.Elsa.Caching.Redis</PackageId>
<Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description>
Bundles the most commonly-used packages when building an Elsa workflows application.
</Description>
<RootNamespace>Elsa</RootNamespace>
<PackageTags>elsa bundle</PackageTags>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\Elsa.Caching\Elsa.Caching.csproj" />
<ProjectReference Include="..\Elsa.Workflows.Core\Elsa.Workflows.Core.csproj" />
<PackageReference Include="StackExchange.Redis" />
<PackageReference Include="Microsoft.Extensions.Configuration.Binder" />
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" />
</ItemGroup>
</Project>
@@ -0,0 +1,31 @@
using Elsa.Caching.Features;
using Elsa.Caching.Redis.Features;
using Elsa.Caching.Redis.Options;
using Elsa.Caching.Redis.Services;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Extensions;
public static class RedisCacheFeatureExtensions
{
public static CacheFeature RegisterRedis(this CacheFeature cache)
=> cache.RegisterProvider("Redis", section => cache.UseRedis(options => { if (section != null) section.Bind(options); }));
public static CacheFeature UseRedis(this CacheFeature cache, Action<RedisCacheOptions>? configure = null)
{
cache.Module.Configure<RedisCacheFeature>(redis =>
{
// Bind first; explicit property assignments then override individual settings.
cache.Configuration?.GetSection("Redis").Bind(redis.Options);
configure?.Invoke(redis.Options);
});
return cache.SelectProvider("Redis", sp => sp.GetRequiredService<RedisCacheProvider>());
}
public static T UseRedis<T>(this T feature, Action<RedisCacheOptions>? configure = null) where T : CacheConsumerFeature
{
feature.Cache.UseRedis(configure);
return feature;
}
}
@@ -0,0 +1,33 @@
using Elsa.Caching.Features;
using Elsa.Caching.Redis.Options;
using Elsa.Caching.Redis.Services;
using Elsa.Features.Abstractions;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Microsoft.Extensions.DependencyInjection;
using StackExchange.Redis;
namespace Elsa.Caching.Redis.Features;
[DependsOn(typeof(CacheFeature))]
public class RedisCacheFeature(IModule module) : FeatureBase(module)
{
public RedisCacheOptions Options { get; } = new();
public override void Apply()
{
if (Options.Database < -1) throw new InvalidOperationException("Redis cache database must be -1 or a non-negative number.");
if (string.IsNullOrWhiteSpace(Options.KeyPrefix)) throw new InvalidOperationException("Redis cache KeyPrefix must not be empty.");
Services.AddSingleton(Options);
if (!Options.HasExplicitConnectionFactory && !string.IsNullOrWhiteSpace(Options.ConnectionString))
{
Services.AddSingleton<OwnedRedisCacheConnection>(_ => new(ConnectionMultiplexer.Connect(Options.ConnectionString)));
Options.ConnectionFactory = sp => sp.GetRequiredService<OwnedRedisCacheConnection>().Connection;
}
Services.AddSingleton<RedisCacheProvider>();
}
private sealed class OwnedRedisCacheConnection(IConnectionMultiplexer connection) : IDisposable
{
public IConnectionMultiplexer Connection => connection;
public void Dispose() => connection.Dispose();
}
}
@@ -0,0 +1,3 @@
<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd">
<ConfigureAwait />
</Weavers>
@@ -0,0 +1,20 @@
using StackExchange.Redis;
namespace Elsa.Caching.Redis.Options;
public class RedisCacheOptions
{
/// <summary>Uses the host's singleton connection by default. The cache does not dispose host connections.</summary>
private Func<IServiceProvider, IConnectionMultiplexer> _connectionFactory
= sp => (IConnectionMultiplexer)(sp.GetService(typeof(IConnectionMultiplexer))
?? throw new InvalidOperationException("Register IConnectionMultiplexer or configure Redis ConnectionFactory/ConnectionString."));
internal bool HasExplicitConnectionFactory { get; private set; }
public Func<IServiceProvider, IConnectionMultiplexer> ConnectionFactory
{
get => _connectionFactory;
set { _connectionFactory = value ?? throw new ArgumentNullException(nameof(value)); HasExplicitConnectionFactory = true; }
}
public string? ConnectionString { get; set; }
public int Database { get; set; } = -1;
public string KeyPrefix { get; set; } = "nextbank:workflow:cache:";
}
@@ -0,0 +1,80 @@
using System.Text.Json;
using Elsa.Caching.Models;
using Elsa.Caching.Redis.Options;
using Elsa.Workflows;
using Microsoft.Extensions.Logging;
using StackExchange.Redis;
namespace Elsa.Caching.Redis.Services;
/// <summary>Stores serializable payloads only. Redis failures bypass caching; never fall back to local payloads.</summary>
public class RedisCacheProvider(IServiceProvider services, RedisCacheOptions options, IPayloadSerializer serializer,
ILogger<RedisCacheProvider> logger) : ICacheProvider
{
public bool IsShared => true;
private IDatabase Database => options.ConnectionFactory(services).GetDatabase(options.Database);
private string Key(string key) => options.KeyPrefix + key;
private const string ReadScript = """
local payload = redis.call('HGET', KEYS[1], 'payload')
if not payload then return false end
local slide = tonumber(redis.call('HGET', KEYS[1], 'slide')) or 0
local deadline = tonumber(redis.call('HGET', KEYS[1], 'deadline')) or 0
if slide > 0 then
local ttl = slide
if deadline > 0 then
local t = redis.call('TIME')
ttl = math.min(ttl, deadline - (tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000)))
end
if ttl <= 0 then redis.call('DEL', KEYS[1]); return false end
redis.call('PEXPIRE', KEYS[1], ttl)
end
return payload
""";
private const string WriteScript = """
local absolute = tonumber(ARGV[2])
local slide = tonumber(ARGV[3])
local ttl = absolute
if ttl <= 0 or (slide > 0 and slide < ttl) then ttl = slide end
local deadline = 0
if absolute > 0 then
local t = redis.call('TIME')
deadline = tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000) + absolute
end
redis.call('HSET', KEYS[1], 'payload', ARGV[1], 'slide', slide, 'deadline', deadline)
redis.call('PEXPIRE', KEYS[1], ttl)
return 1
""";
public async Task<CacheResult<T>> GetAsync<T>(string key, CancellationToken cancellationToken = default)
{
cancellationToken.ThrowIfCancellationRequested();
try
{
var data = await Database.ScriptEvaluateAsync(ReadScript, new RedisKey[] { Key(key) }).WaitAsync(cancellationToken);
if (data.IsNull) return new(false, default);
return new(true, JsonSerializer.Deserialize<T>((string)data!, serializer.GetOptions()));
}
catch (RedisException ex) { logger.LogWarning(ex, "Redis workflow cache read failed; using the persistent store."); return new(false, default); }
catch (JsonException ex) { logger.LogWarning(ex, "Invalid Redis workflow cache payload; rebuilding the entry."); return new(false, default); }
}
public async Task SetAsync<T>(string key, T? value, CacheEntryOptions entryOptions, CancellationToken cancellationToken = default)
{
var absolute = Milliseconds(entryOptions.AbsoluteExpirationRelativeToNow);
var sliding = Milliseconds(entryOptions.SlidingExpiration);
if (absolute <= 0 && sliding <= 0) throw new ArgumentException("A Redis cache entry must have a positive lifetime.");
var data = JsonSerializer.Serialize(value, serializer.GetOptions());
try
{
await Database.ScriptEvaluateAsync(WriteScript, new RedisKey[] { Key(key) }, new RedisValue[] { data, absolute, sliding }).WaitAsync(cancellationToken);
}
catch (RedisException ex) { logger.LogWarning(ex, "Redis workflow cache write failed; the SQL revision remains authoritative."); }
}
public async Task RemoveAsync(string key, CancellationToken cancellationToken = default)
{
try { await Database.KeyDeleteAsync(Key(key)).WaitAsync(cancellationToken); }
catch (RedisException ex) { logger.LogWarning(ex, "Redis workflow cache removal failed."); }
}
private static long Milliseconds(TimeSpan? duration) => duration.HasValue ? Math.Max(1, (long)duration.Value.TotalMilliseconds) : 0;
}
@@ -0,0 +1,15 @@
namespace Elsa.Caching;
/// <summary>
/// Authoritative cache revision and durable write fences. A pending mutation disables cache reads and writes.
/// Shared providers require a durable store shared by every node, independently of the cache payload storage.
/// </summary>
public interface ICacheConsistencyStore
{
bool IsDurable { get; }
Task<CacheConsistencyState> ReadAsync(CancellationToken cancellationToken = default);
Task<string> BeginMutationAsync(CancellationToken cancellationToken = default, string scope = "workflow");
Task CompleteMutationAsync(string mutationId, CancellationToken cancellationToken = default);
}
public record CacheConsistencyState(string Revision, bool HasPendingMutations, string? RouteRevision = null, bool HasPendingRouteMutations = false);
@@ -10,6 +10,21 @@ namespace Elsa.Caching;
/// </summary> /// </summary>
public interface ICacheManager public interface ICacheManager
{ {
bool IsShared => false;
bool IsEnabled => true;
Task<string> BeginMutationAsync(CancellationToken cancellationToken = default) => Task.FromResult(string.Empty);
Task<string> BeginRouteMutationAsync(CancellationToken cancellationToken = default) => BeginMutationAsync(cancellationToken);
Task CompleteMutationAsync(string id, CancellationToken cancellationToken = default) => Task.CompletedTask;
Task<CacheConsistencyState> GetConsistencyStateAsync(CancellationToken cancellationToken = default)
=> Task.FromResult(new CacheConsistencyState("0", false));
Task<T?> GetOrCreateAsync<T>(string key, Func<Task<T>> factory, Elsa.Caching.Models.CacheEntryOptions options, CancellationToken cancellationToken = default)
=> GetOrCreateAsync(key, entry =>
{
entry.AbsoluteExpirationRelativeToNow = options.AbsoluteExpirationRelativeToNow;
entry.SlidingExpiration = options.SlidingExpiration;
foreach (var tag in options.Tags) entry.ExpirationTokens.Add(GetToken(tag));
return factory();
});
/// <summary> /// <summary>
/// Provides options for configuring caching. /// Provides options for configuring caching.
/// </summary> /// </summary>
@@ -0,0 +1,14 @@
using Elsa.Caching.Models;
namespace Elsa.Caching;
/// <summary>Storage for cache entries. Implementations must distinguish a miss from a cached null.</summary>
public interface ICacheProvider
{
bool IsShared { get; }
Task<CacheResult<T>> GetAsync<T>(string key, CancellationToken cancellationToken = default);
Task SetAsync<T>(string key, T? value, CacheEntryOptions options, CancellationToken cancellationToken = default);
Task RemoveAsync(string key, CancellationToken cancellationToken = default);
}
public record CacheResult<T>(bool Found, T? Value);
+4 -2
View File
@@ -1,10 +1,10 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<PackageId>Caspian.Elsa.Caching</PackageId> <PackageId>Caspian.Elsa.Caching</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides caching services and utility functions common to Elsa modules. Provides caching services and utility functions common to Elsa modules.
@@ -19,6 +19,8 @@
<ItemGroup> <ItemGroup>
<PackageReference Include="Microsoft.Extensions.Caching.Memory" /> <PackageReference Include="Microsoft.Extensions.Caching.Memory" />
<PackageReference Include="Scrutor" /> <PackageReference Include="Scrutor" />
<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" />
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" />
</ItemGroup> </ItemGroup>
</Project> </Project>
@@ -0,0 +1,13 @@
using Elsa.Caching.Features;
using Elsa.Features.Services;
namespace Elsa.Extensions;
public static class CacheFeatureExtensions
{
public static IModule UseCaching(this IModule module, Action<CacheFeature>? configure = null)
{
module.Configure(configure);
return module;
}
}
@@ -0,0 +1,34 @@
using Elsa.Caching;
namespace Elsa.Extensions;
public static class CacheMutationExtensions
{
public static async Task MutateRoutesAsync(this ICacheManager cache, Func<Task> operation, CancellationToken cancellationToken = default)
{
var id = await cache.BeginRouteMutationAsync(cancellationToken);
await operation();
await cache.CompleteMutationAsync(id, CancellationToken.None);
}
public static async Task<T> MutateRoutesAsync<T>(this ICacheManager cache, Func<Task<T>> operation, CancellationToken cancellationToken = default)
{
var id = await cache.BeginRouteMutationAsync(cancellationToken);
var value = await operation();
await cache.CompleteMutationAsync(id, CancellationToken.None);
return value;
}
public static async Task MutateAsync(this ICacheManager cache, Func<Task> operation, CancellationToken cancellationToken = default)
{
var id = await cache.BeginMutationAsync(cancellationToken);
await operation();
// Once SQL has committed, request cancellation must not leave a fence unfinished.
await cache.CompleteMutationAsync(id, CancellationToken.None);
}
public static async Task<T> MutateAsync<T>(this ICacheManager cache, Func<Task<T>> operation, CancellationToken cancellationToken = default)
{
var id = await cache.BeginMutationAsync(cancellationToken);
var value = await operation();
await cache.CompleteMutationAsync(id, CancellationToken.None);
return value;
}
}
@@ -0,0 +1,14 @@
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
namespace Elsa.Caching.Features;
/// <summary>Common configuration entry point for workflow cache consumers.</summary>
public abstract class CacheConsumerFeature(IModule module) : FeatureBase(module)
{
public CacheFeature Cache => Module.Configure<CacheFeature>();
public bool Enabled { get => Cache.Enabled; set => Cache.Enabled = value; }
public TimeSpan Duration { get => Cache.Duration; set => Cache.Duration = value; }
public void UseMemory() => Cache.UseMemory();
public void UseConfiguration(Microsoft.Extensions.Configuration.IConfigurationSection section) => Cache.UseConfiguration(section);
}
@@ -0,0 +1,80 @@
using Elsa.Caching.Services;
using Elsa.Caching.Options;
using Elsa.Features.Abstractions;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Caching.Features;
/// <summary>Configures one provider for the entire Elsa module.</summary>
[DependsOn(typeof(MemoryCacheFeature))]
public class CacheFeature(IModule module) : FeatureBase(module)
{
private readonly Dictionary<string, Action<IConfigurationSection?>> _providers = new(StringComparer.OrdinalIgnoreCase);
private string? _explicitProvider;
private bool _selectingConfiguration;
public bool Enabled { get; set; } = true;
public TimeSpan Duration { get; set; } = TimeSpan.FromMinutes(1);
public Func<IServiceProvider, ICacheProvider> Provider { get; private set; } = sp => sp.GetRequiredService<MemoryCacheProvider>();
public Func<IServiceProvider, ICacheConsistencyStore> ConsistencyStore { get; set; } = sp => sp.GetRequiredService<MemoryCacheConsistencyStore>();
public CacheFeature RegisterProvider(string name, Action<IConfigurationSection?> configure)
{
_providers[name] = configure;
return this;
}
public CacheFeature SelectProvider(string name, Func<IServiceProvider, ICacheProvider> factory)
{
if (_explicitProvider != null && !string.Equals(_explicitProvider, name, StringComparison.OrdinalIgnoreCase))
throw new InvalidOperationException($"Conflicting Elsa cache providers: '{_explicitProvider}' and '{name}'. Select one provider per module.");
if (!_selectingConfiguration) _explicitProvider = name;
Provider = factory;
return this;
}
public CacheFeature UseMemory() => SelectProvider("Memory", sp => sp.GetRequiredService<MemoryCacheProvider>());
public CacheFeature UseConfiguration(IConfigurationSection section)
{
if (section["Enabled"] is { } enabledValue)
Enabled = bool.TryParse(enabledValue, out var enabled) ? enabled
: throw new InvalidOperationException("Workflow cache Enabled must be true or false.");
if (section["Duration"] is { } durationValue)
Duration = TimeSpan.TryParse(durationValue, out var duration) ? duration
: throw new InvalidOperationException("Workflow cache Duration must be a valid TimeSpan.");
Configuration = section;
var name = section["Provider"] ?? "Memory";
if (_explicitProvider == null)
{
_selectingConfiguration = true;
try
{
if (name.Equals("Memory", StringComparison.OrdinalIgnoreCase)) UseMemory();
else if (_providers.TryGetValue(name, out var select)) select(section.GetSection(name));
else throw new InvalidOperationException($"Elsa cache provider '{name}' is not registered.");
}
finally { _selectingConfiguration = false; }
}
Configuration = section;
return this;
}
public IConfigurationSection? Configuration { get; private set; }
public override void Apply()
{
if (Duration <= TimeSpan.Zero) throw new InvalidOperationException("Elsa cache duration must be positive.");
Services.Configure<CachingOptions>(options => options.CacheDuration = Duration);
Services.AddSingleton(Provider);
Services.AddSingleton(ConsistencyStore);
Services.AddSingleton(this);
Services.AddSingleton<ICacheManager>(sp =>
{
if (sp.GetRequiredService<ICacheProvider>().IsShared && !sp.GetRequiredService<ICacheConsistencyStore>().IsDurable)
throw new InvalidOperationException("Shared Elsa caching requires a durable ICacheConsistencyStore. Configure the EF Core consistency store.");
return ActivatorUtilities.CreateInstance<ProviderCacheManager>(sp);
});
}
}
@@ -3,6 +3,7 @@ using Elsa.Caching.Services;
using Elsa.Features.Abstractions; using Elsa.Features.Abstractions;
using Elsa.Features.Services; using Elsa.Features.Services;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
namespace Elsa.Caching.Features; namespace Elsa.Caching.Features;
@@ -23,9 +24,12 @@ public class MemoryCacheFeature(IModule module) : FeatureBase(module)
Services Services
.AddMemoryCache() .AddMemoryCache()
.AddSingleton<ICacheManager, CacheManager>()
.AddSingleton<IChangeTokenSignalInvoker, ChangeTokenSignalInvoker>() .AddSingleton<IChangeTokenSignalInvoker, ChangeTokenSignalInvoker>()
.AddSingleton<IChangeTokenSignaler, ChangeTokenSignaler>() .AddSingleton<IChangeTokenSignaler, ChangeTokenSignaler>()
; ;
// Keep local script/template caches independent of the workflow provider.
Services.TryAddSingleton<MemoryCacheProvider>();
Services.TryAddSingleton<MemoryCacheConsistencyStore>();
Services.TryAddSingleton<ICacheManager, CacheManager>();
} }
} }
@@ -0,0 +1,9 @@
namespace Elsa.Caching.Models;
/// <summary>Provider-independent lifetime and invalidation dependencies.</summary>
public class CacheEntryOptions
{
public TimeSpan? AbsoluteExpirationRelativeToNow { get; set; }
public TimeSpan? SlidingExpiration { get; set; }
public ICollection<string> Tags { get; } = new List<string>();
}
@@ -0,0 +1,34 @@
namespace Elsa.Caching.Services;
public class MemoryCacheConsistencyStore : ICacheConsistencyStore
{
private readonly object _sync = new();
private readonly Dictionary<string, string> _pending = new();
private long _revision;
private long _routeRevision;
public bool IsDurable => false;
public Task<CacheConsistencyState> ReadAsync(CancellationToken cancellationToken = default)
{
lock (_sync) return Task.FromResult(new CacheConsistencyState(_revision.ToString(), _pending.Values.Any(x => x == "workflow"), _routeRevision.ToString(), _pending.Count != 0));
}
public Task<string> BeginMutationAsync(CancellationToken cancellationToken = default, string scope = "workflow")
{
lock (_sync)
{
var id = Guid.NewGuid().ToString("N");
_pending.Add(id, scope);
if (scope == "workflow") _revision++;
_routeRevision++;
return Task.FromResult(id);
}
}
public Task CompleteMutationAsync(string mutationId, CancellationToken cancellationToken = default)
{
lock (_sync)
{
if (_pending.Remove(mutationId, out var scope) && scope == "workflow") _revision++;
_routeRevision++;
}
return Task.CompletedTask;
}
}
@@ -0,0 +1,32 @@
using Elsa.Caching.Models;
using Microsoft.Extensions.Caching.Memory;
namespace Elsa.Caching.Services;
public class MemoryCacheProvider(IMemoryCache cache, IChangeTokenSignaler signaler) : ICacheProvider
{
public bool IsShared => false;
public Task<CacheResult<T>> GetAsync<T>(string key, CancellationToken cancellationToken = default)
{
var found = cache.TryGetValue(key, out T? value);
return Task.FromResult(new CacheResult<T>(found, value));
}
public Task SetAsync<T>(string key, T? value, CacheEntryOptions options, CancellationToken cancellationToken = default)
{
var memoryOptions = new MemoryCacheEntryOptions
{
AbsoluteExpirationRelativeToNow = options.AbsoluteExpirationRelativeToNow,
SlidingExpiration = options.SlidingExpiration
};
foreach (var tag in options.Tags) memoryOptions.AddExpirationToken(signaler.GetToken(tag));
cache.Set(key, value, memoryOptions);
return Task.CompletedTask;
}
public Task RemoveAsync(string key, CancellationToken cancellationToken = default)
{
cache.Remove(key);
return Task.CompletedTask;
}
}
@@ -0,0 +1,122 @@
using System.Security.Cryptography;
using System.Text;
using System.Transactions;
using Elsa.Caching.Features;
using Elsa.Caching.Models;
using Elsa.Caching.Options;
using Microsoft.Extensions.Caching.Memory;
using Microsoft.Extensions.Options;
using Microsoft.Extensions.Primitives;
using Microsoft.Extensions.Logging;
namespace Elsa.Caching.Services;
/// <summary>Coordinates cache payloads using an authoritative revision, independent of the provider.</summary>
public class ProviderCacheManager(ICacheProvider provider, ICacheConsistencyStore consistency,
IChangeTokenSignaler signaler, IOptions<CachingOptions> options, CacheFeature feature, IMemoryCache memoryCache,
ILogger<ProviderCacheManager> logger) : ICacheManager
{
public IOptions<CachingOptions> CachingOptions => options;
public bool IsShared => provider.IsShared;
public bool IsEnabled => feature.Enabled;
public IChangeToken GetToken(string key) => signaler.GetToken(key);
public Task<CacheConsistencyState> GetConsistencyStateAsync(CancellationToken cancellationToken = default) => consistency.ReadAsync(cancellationToken);
public async Task<string> BeginMutationAsync(CancellationToken cancellationToken = default)
=> await BeginMutationCoreAsync("workflow", cancellationToken);
public Task<string> BeginRouteMutationAsync(CancellationToken cancellationToken = default)
=> BeginMutationCoreAsync("routes", cancellationToken);
public async Task CompleteMutationAsync(string id, CancellationToken cancellationToken = default)
{
// Begin registered the transaction-completed callback. Do not close a fence before the outer commit.
if (Transaction.Current != null) return;
await consistency.CompleteMutationAsync(id, cancellationToken);
}
private async Task<string> BeginMutationCoreAsync(string scope, CancellationToken cancellationToken)
{
var transaction = Transaction.Current;
string id;
using (new TransactionScope(TransactionScopeOption.Suppress, TransactionScopeAsyncFlowOption.Enabled))
id = await consistency.BeginMutationAsync(cancellationToken, scope);
if (transaction != null)
{
void CompleteAfterTransaction(TransactionStatus status)
{
if (status is not (TransactionStatus.Committed or TransactionStatus.Aborted)) return;
try
{
using var suppressed = new TransactionScope(TransactionScopeOption.Suppress, TransactionScopeAsyncFlowOption.Enabled);
consistency.CompleteMutationAsync(id, CancellationToken.None).GetAwaiter().GetResult();
}
catch (Exception ex) { logger.LogError(ex, "Workflow cache mutation {MutationId} remains pending after transaction completion.", id); }
}
transaction.TransactionCompleted += (_, args) => CompleteAfterTransaction(args.Transaction!.TransactionInformation.Status);
if (transaction.TransactionInformation.Status != TransactionStatus.Active)
CompleteAfterTransaction(transaction.TransactionInformation.Status);
}
return id;
}
public async ValueTask TriggerTokenAsync(string key, CancellationToken cancellationToken = default)
{
var id = await BeginMutationAsync(cancellationToken);
await CompleteMutationAsync(id, cancellationToken);
await signaler.TriggerTokenAsync(key, cancellationToken);
}
public async Task<TItem?> GetOrCreateAsync<TItem>(object key, Func<ICacheEntry, Task<TItem>> factory)
{
if (Transaction.Current != null) { using var transactionEntry = new CompatibilityCacheEntry(key); return await factory(transactionEntry); }
if (!feature.Enabled) { using var uncached = new CompatibilityCacheEntry(key); return await factory(uncached); }
if (!IsShared) return await memoryCache.GetOrCreateAsync(key, factory);
// Compatibility entry for older extensions. New consumers use CacheEntryOptions.
using var entry = new CompatibilityCacheEntry(key);
return await GetOrCreateAsync(key.ToString()!, () => factory(entry), entry.Options);
}
public async Task<T?> GetOrCreateAsync<T>(string key, Func<Task<T>> factory, CacheEntryOptions entryOptions, CancellationToken cancellationToken = default)
{
if (!feature.Enabled || Transaction.Current != null) return await factory();
if (IsShared && !consistency.IsDurable)
throw new InvalidOperationException("Shared Elsa caching requires a durable ICacheConsistencyStore. Configure the EF Core consistency store.");
var before = await consistency.ReadAsync(cancellationToken);
if (before.HasPendingMutations) return await factory();
var storageKey = CreateKey<T>(key, before.Revision);
var cached = await provider.GetAsync<T>(storageKey, cancellationToken);
if (cached.Found) return cached.Value;
var value = await factory();
var after = await consistency.ReadAsync(cancellationToken);
if (!after.HasPendingMutations && after.Revision == before.Revision)
{
if (entryOptions.AbsoluteExpirationRelativeToNow == null && entryOptions.SlidingExpiration == null)
entryOptions.AbsoluteExpirationRelativeToNow = options.Value.CacheDuration;
await provider.SetAsync(storageKey, value, entryOptions, cancellationToken);
}
return value;
}
public void Remove(string key) => TriggerTokenAsync(key).AsTask().GetAwaiter().GetResult();
private static string CreateKey<T>(string key, string revision)
=> $"v1:{revision}:" + Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes($"{typeof(T).FullName}:{key}")));
private sealed class CompatibilityCacheEntry(object key) : ICacheEntry
{
public CacheEntryOptions Options { get; } = new();
public object Key => key;
public object? Value { get; set; }
private DateTimeOffset? _absoluteExpiration;
public DateTimeOffset? AbsoluteExpiration
{
get => _absoluteExpiration;
set { _absoluteExpiration = value; Options.AbsoluteExpirationRelativeToNow = value - DateTimeOffset.UtcNow; }
}
public TimeSpan? AbsoluteExpirationRelativeToNow { get => Options.AbsoluteExpirationRelativeToNow; set => Options.AbsoluteExpirationRelativeToNow = value; }
public TimeSpan? SlidingExpiration { get => Options.SlidingExpiration; set => Options.SlidingExpiration = value; }
public IList<IChangeToken> ExpirationTokens { get; } = new List<IChangeToken>();
public IList<PostEvictionCallbackRegistration> PostEvictionCallbacks { get; } = new List<PostEvictionCallbackRegistration>();
public CacheItemPriority Priority { get; set; }
public long? Size { get; set; }
public void Dispose() { }
}
}
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Common</PackageId> <PackageId>Caspian.Elsa.Common</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides services and utility functions common to Elsa modules. Provides services and utility functions common to Elsa modules.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Connections.Api</PackageId> <PackageId>Caspian.Elsa.Connections.Api</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description>Provides Activity Connection API endpoints</Description> <Description>Provides Activity Connection API endpoints</Description>
<PackageTags>elsa module activity connection abstraction api</PackageTags> <PackageTags>elsa module activity connection abstraction api</PackageTags>
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Connections.Core</PackageId> <PackageId>Caspian.Elsa.Connections.Core</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description>Provides an connection framework to use connection abtraction in activities</Description> <Description>Provides an connection framework to use connection abtraction in activities</Description>
<PackageTags>elsa module connection abstratction</PackageTags> <PackageTags>elsa module connection abstratction</PackageTags>
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Connections.Models</PackageId> <PackageId>Caspian.Elsa.Connections.Models</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description>Provides models that can be shared between the API and API clients</Description> <Description>Provides models that can be shared between the API and API clients</Description>
<PackageTags>elsa module connections api</PackageTags> <PackageTags>elsa module connections api</PackageTags>
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Connections.Persistence.EntityFrameworkCore.Sqlite</PackageId> <PackageId>Caspian.Elsa.Connections.Persistence.EntityFrameworkCore.Sqlite</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description>Provides an EF Core migrations for SQLite for the Connection Persistence module.</Description> <Description>Provides an EF Core migrations for SQLite for the Connection Persistence module.</Description>
<PackageTags>elsa module persistence efcore entity framework core sqlite</PackageTags> <PackageTags>elsa module persistence efcore entity framework core sqlite</PackageTags>
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Connections.Persistence.EntityFrameworkCore</PackageId> <PackageId>Caspian.Elsa.Connections.Persistence.EntityFrameworkCore</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description>Provides an Entity Framework Core provider for the persistence layer of the Connection module</Description> <Description>Provides an Entity Framework Core provider for the persistence layer of the Connection module</Description>
<PackageTags>elsa connection persistence efcore entity framework core</PackageTags> <PackageTags>elsa connection persistence efcore entity framework core</PackageTags>
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Connections.Persistence</PackageId> <PackageId>Caspian.Elsa.Connections.Persistence</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description>Provides Connections persistence services</Description> <Description>Provides Connections persistence services</Description>
<PackageTags>elsa module connections persistence </PackageTags> <PackageTags>elsa module connections persistence </PackageTags>
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Dapper.Migrations</PackageId> <PackageId>Caspian.Elsa.Dapper.Migrations</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides Dapper migrations for Dapper perisstence providers from various modules. Provides Dapper migrations for Dapper perisstence providers from various modules.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Dapper</PackageId> <PackageId>Caspian.Elsa.Dapper</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides Dapper implementations of various persistence abstractions from various modules. Provides Dapper implementations of various persistence abstractions from various modules.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Dsl</PackageId> <PackageId>Caspian.Elsa.Dsl</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides a custom DSL to write workflows with in a syntax similar to JavaScript but with native keywords translating to workflow functionality. Provides a custom DSL to write workflows with in a syntax similar to JavaScript but with native keywords translating to workflow functionality.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Elasticsearch</PackageId> <PackageId>Caspian.Elsa.Elasticsearch</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides Elasticsearch implementation for workflow state persistence. Provides Elasticsearch implementation for workflow state persistence.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Email</PackageId> <PackageId>Caspian.Elsa.Email</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides a SendEmail activity that lets you send emails from a workflow. Provides a SendEmail activity that lets you send emails from a workflow.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.EntityFrameworkCore.Common</PackageId> <PackageId>Caspian.Elsa.EntityFrameworkCore.Common</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides common Entity Framework Core types for implementing EF Core persistence of varipus ELsa modules. Provides common Entity Framework Core types for implementing EF Core persistence of varipus ELsa modules.
@@ -5,7 +5,7 @@
<PackageId>Caspian.Elsa.EntityFrameworkCore.MySql</PackageId> <PackageId>Caspian.Elsa.EntityFrameworkCore.MySql</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides MySQL EF Core migrations for various modules. Provides MySQL EF Core migrations for various modules.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.EntityFrameworkCore.Oracle</PackageId> <PackageId>Caspian.Elsa.EntityFrameworkCore.Oracle</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides Oracle EF Core migrations for various modules. Provides Oracle EF Core migrations for various modules.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.EntityFrameworkCore.PostgreSql</PackageId> <PackageId>Caspian.Elsa.EntityFrameworkCore.PostgreSql</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides PostgreSQL EF Core migrations for various modules. Provides PostgreSQL EF Core migrations for various modules.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.EntityFrameworkCore.SqlServer</PackageId> <PackageId>Caspian.Elsa.EntityFrameworkCore.SqlServer</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides SQL Server EF Core migrations for various modules. Provides SQL Server EF Core migrations for various modules.
@@ -0,0 +1,26 @@
using Elsa.EntityFrameworkCore.Modules.Management;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Migrations;
namespace Elsa.EntityFrameworkCore.SqlServer.Migrations.Management;
[DbContext(typeof(ManagementElsaDbContext))]
[Migration("20261007000000_AddCacheMutationFences")]
public class AddCacheMutationFences : Migration
{
protected override void Up(MigrationBuilder migrationBuilder)
{
migrationBuilder.CreateTable("CacheMutations", schema: "Elsa", columns: table => new
{
Id = table.Column<long>(nullable: false).Annotation("SqlServer:Identity", "1, 1"),
MutationId = table.Column<string>(maxLength: 32, nullable: false),
Scope = table.Column<string>(maxLength: 32, nullable: false),
OwnerInstance = table.Column<string>(maxLength: 256, nullable: false),
IsCompleted = table.Column<bool>(nullable: false),
CreatedAt = table.Column<DateTimeOffset>(nullable: false)
}, constraints: table => table.PrimaryKey("PK_CacheMutations", x => x.Id));
migrationBuilder.CreateIndex("IX_CacheMutations_MutationId", "CacheMutations", "MutationId", "Elsa", unique: true);
migrationBuilder.CreateIndex("IX_CacheMutations_IsCompleted", "CacheMutations", "IsCompleted", "Elsa");
}
protected override void Down(MigrationBuilder migrationBuilder) => migrationBuilder.DropTable("CacheMutations", "Elsa");
}
@@ -1,4 +1,4 @@
// <auto-generated /> // <auto-generated />
using System; using System;
using Elsa.EntityFrameworkCore.Modules.Management; using Elsa.EntityFrameworkCore.Modules.Management;
using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore;
@@ -23,6 +23,21 @@ namespace Elsa.EntityFrameworkCore.SqlServer.Migrations.Management
SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder); SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder);
modelBuilder.Entity("Elsa.EntityFrameworkCore.Modules.Management.CacheMutation", b =>
{
b.Property<long>("Id").ValueGeneratedOnAdd().HasColumnType("bigint");
SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property<long>("Id"));
b.Property<string>("MutationId").IsRequired().HasMaxLength(32).HasColumnType("nvarchar(32)");
b.Property<string>("Scope").IsRequired().HasMaxLength(32).HasColumnType("nvarchar(32)");
b.Property<string>("OwnerInstance").IsRequired().HasMaxLength(256).HasColumnType("nvarchar(256)");
b.Property<bool>("IsCompleted").HasColumnType("bit");
b.Property<DateTimeOffset>("CreatedAt").HasColumnType("datetimeoffset");
b.HasKey("Id");
b.HasIndex("MutationId").IsUnique();
b.HasIndex("IsCompleted");
b.ToTable("CacheMutations", "Elsa");
});
modelBuilder.Entity("Elsa.Workflows.Management.Entities.WorkflowDefinition", b => modelBuilder.Entity("Elsa.Workflows.Management.Entities.WorkflowDefinition", b =>
{ {
b.Property<string>("Id") b.Property<string>("Id")
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.EntityFrameworkCore.Sqlite</PackageId> <PackageId>Caspian.Elsa.EntityFrameworkCore.Sqlite</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides SQLite EF Core migrations for various modules. Provides SQLite EF Core migrations for various modules.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.EntityFrameworkCore</PackageId> <PackageId>Caspian.Elsa.EntityFrameworkCore</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides Entity Framework Core implementations of various abstractions from various modules. Provides Entity Framework Core implementations of various abstractions from various modules.
@@ -0,0 +1,13 @@
namespace Elsa.EntityFrameworkCore.Modules.Management;
/// <summary>Durable write fence. Pending records must never expire automatically.</summary>
public class CacheMutation
{
public static string CurrentOwnerInstance { get; } = $"{Environment.MachineName}:{Environment.ProcessId}:{Guid.NewGuid():N}";
public long Id { get; set; }
public string MutationId { get; set; } = Guid.NewGuid().ToString("N");
public string Scope { get; set; } = "workflow";
public string OwnerInstance { get; set; } = CurrentOwnerInstance;
public bool IsCompleted { get; set; }
public DateTimeOffset CreatedAt { get; set; } = DateTimeOffset.UtcNow;
}
@@ -24,6 +24,7 @@ public class ManagementElsaDbContext : ElsaDbContextBase
/// The workflow instances. /// The workflow instances.
/// </summary> /// </summary>
public DbSet<WorkflowInstance> WorkflowInstances { get; set; } = default!; public DbSet<WorkflowInstance> WorkflowInstances { get; set; } = default!;
public DbSet<CacheMutation> CacheMutations { get; set; } = default!;
/// <inheritdoc /> /// <inheritdoc />
protected override void OnModelCreating(ModelBuilder modelBuilder) protected override void OnModelCreating(ModelBuilder modelBuilder)
@@ -34,6 +35,17 @@ public class ManagementElsaDbContext : ElsaDbContextBase
var config = new Configurations(); var config = new Configurations();
modelBuilder.ApplyConfiguration<WorkflowDefinition>(config); modelBuilder.ApplyConfiguration<WorkflowDefinition>(config);
modelBuilder.ApplyConfiguration<WorkflowInstance>(config); modelBuilder.ApplyConfiguration<WorkflowInstance>(config);
modelBuilder.Entity<CacheMutation>(b =>
{
b.ToTable("CacheMutations");
b.HasKey(x => x.Id);
b.Property(x => x.Id).ValueGeneratedOnAdd();
b.Property(x => x.MutationId).HasMaxLength(32).IsRequired();
b.HasIndex(x => x.MutationId).IsUnique();
b.Property(x => x.Scope).HasMaxLength(32).IsRequired();
b.Property(x => x.OwnerInstance).HasMaxLength(256).IsRequired();
b.HasIndex(x => x.IsCompleted);
});
base.OnModelCreating(modelBuilder); base.OnModelCreating(modelBuilder);
} }
@@ -0,0 +1,57 @@
using Elsa.Caching;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using System.Transactions;
namespace Elsa.EntityFrameworkCore.Modules.Management;
/// <summary>Reads SQL before cache access, so missed messages and Redis failover cannot revive old revisions.</summary>
public class EFCoreCacheConsistencyStore(IServiceScopeFactory scopes) : ICacheConsistencyStore
{
public bool IsDurable => true;
public async Task<CacheConsistencyState> ReadAsync(CancellationToken cancellationToken = default)
{
using var suppressed = new TransactionScope(TransactionScopeOption.Suppress, TransactionScopeAsyncFlowOption.Enabled);
using var serviceScope = scopes.CreateScope();
var factory = serviceScope.ServiceProvider.GetRequiredService<IDbContextFactory<ManagementElsaDbContext>>();
await using var db = await factory.CreateDbContextAsync(cancellationToken);
// One SQL statement gives a consistent view of the revision and pending fences.
var state = await db.CacheMutations.AsNoTracking().GroupBy(_ => 1)
.Select(g => new { Revision = g.Max(x => x.Scope == "workflow" ? (long?)x.Id : null) ?? 0,
Pending = g.Any(x => !x.IsCompleted && x.Scope == "workflow"),
RouteRevision = g.Max(x => x.Id), PendingRoutes = g.Any(x => !x.IsCompleted) })
.SingleOrDefaultAsync(cancellationToken);
return new CacheConsistencyState(state?.Revision.ToString() ?? "0", state?.Pending ?? false, state?.RouteRevision.ToString() ?? "0", state?.PendingRoutes ?? false);
}
public async Task<string> BeginMutationAsync(CancellationToken cancellationToken = default, string scope = "workflow")
{
using var suppressed = new TransactionScope(TransactionScopeOption.Suppress, TransactionScopeAsyncFlowOption.Enabled);
if (scope is not ("workflow" or "routes")) throw new ArgumentException("Unknown workflow cache mutation scope.", nameof(scope));
using var serviceScope = scopes.CreateScope();
var factory = serviceScope.ServiceProvider.GetRequiredService<IDbContextFactory<ManagementElsaDbContext>>();
await using var db = await factory.CreateDbContextAsync(cancellationToken);
var mutation = new CacheMutation { Scope = scope };
db.CacheMutations.Add(mutation);
await db.SaveChangesAsync(cancellationToken);
return mutation.MutationId;
}
public async Task CompleteMutationAsync(string mutationId, CancellationToken cancellationToken = default)
{
using var suppressed = new TransactionScope(TransactionScopeOption.Suppress, TransactionScopeAsyncFlowOption.Enabled);
using var serviceScope = scopes.CreateScope();
var factory = serviceScope.ServiceProvider.GetRequiredService<IDbContextFactory<ManagementElsaDbContext>>();
await using var db = await factory.CreateDbContextAsync(cancellationToken);
var mutation = await db.CacheMutations.SingleOrDefaultAsync(x => x.MutationId == mutationId, cancellationToken);
if (mutation == null || mutation.IsCompleted) return;
mutation.IsCompleted = true;
// A begin marker can commit out of identity order. Always allocate a new revision AFTER the data write.
// Closing the fence and publishing that revision must commit in the same SQL transaction.
db.CacheMutations.Add(new CacheMutation { Scope = mutation.Scope, IsCompleted = true });
await db.SaveChangesAsync(cancellationToken);
// Never delete a row newer than the maximum observed for its scope: another node may complete meanwhile.
var latest = await db.CacheMutations.GroupBy(x => x.Scope)
.Select(g => new { Scope = g.Key, Maximum = g.Max(x => x.Id) }).ToListAsync(cancellationToken);
foreach (var item in latest)
await db.CacheMutations.Where(x => x.IsCompleted && x.Scope == item.Scope && x.Id < item.Maximum).ExecuteDeleteAsync(cancellationToken);
}
}
@@ -3,6 +3,7 @@ using Elsa.Features.Services;
using Elsa.Workflows.Management.Entities; using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Management.Features; using Elsa.Workflows.Management.Features;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Elsa.Caching.Features;
namespace Elsa.EntityFrameworkCore.Modules.Management; namespace Elsa.EntityFrameworkCore.Modules.Management;
@@ -16,6 +17,10 @@ public class EFCoreWorkflowDefinitionPersistenceFeature(IModule module) : Persis
/// <inheritdoc /> /// <inheritdoc />
public override void Configure() public override void Configure()
{ {
Module.Configure<CacheFeature>(feature => feature.ConsistencyStore = sp =>
sp.GetRequiredService<Elsa.Caching.ICacheProvider>().IsShared
? ActivatorUtilities.CreateInstance<EFCoreCacheConsistencyStore>(sp)
: sp.GetRequiredService<Elsa.Caching.Services.MemoryCacheConsistencyStore>());
Module.Configure<WorkflowDefinitionsFeature>(feature => Module.Configure<WorkflowDefinitionsFeature>(feature =>
{ {
feature.WorkflowDefinitionStore = sp => sp.GetRequiredService<EFCoreWorkflowDefinitionStore>(); feature.WorkflowDefinitionStore = sp => sp.GetRequiredService<EFCoreWorkflowDefinitionStore>();
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Environments</PackageId> <PackageId>Caspian.Elsa.Environments</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides API endpoints for client applications to enumerate available environments. Provides API endpoints for client applications to enumerate available environments.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Expressions</PackageId> <PackageId>Caspian.Elsa.Expressions</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides infrastructure for registering and executing workflow expressions. Provides infrastructure for registering and executing workflow expressions.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.FileStorage</PackageId> <PackageId>Caspian.Elsa.FileStorage</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides activities to save and load files to and from a confogurable storage provider. Provides activities to save and load files to and from a confogurable storage provider.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Hangfire</PackageId> <PackageId>Caspian.Elsa.Hangfire</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides Hangfire integration and an implementation of Elsa's IJobQueue implementation. Provides Hangfire integration and an implementation of Elsa's IJobQueue implementation.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Hosting.Management</PackageId> <PackageId>Caspian.Elsa.Hosting.Management</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides hosting management functionality. Provides hosting management functionality.
@@ -5,6 +5,12 @@ namespace Elsa.Http;
/// </summary> /// </summary>
public interface IRouteTable : IEnumerable<HttpRouteData> public interface IRouteTable : IEnumerable<HttpRouteData>
{ {
/// <summary>Replaces the local snapshot. Implementations should publish it atomically.</summary>
void Replace(IEnumerable<HttpRouteData> routes)
{
RemoveRange(this.Select(x => x.Route).ToArray());
foreach (var route in routes) Add(route);
}
/// <summary> /// <summary>
/// Adds a route to the table. /// Adds a route to the table.
/// </summary> /// </summary>
+2 -2
View File
@@ -1,10 +1,10 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<PackageId>Caspian.Elsa.Http</PackageId> <PackageId>Caspian.Elsa.Http</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides HTTP activities to listen for and send out HTTP requests. Provides HTTP activities to listen for and send out HTTP requests.
@@ -26,4 +26,13 @@ public static class ModuleExtensions
feature.Module.Configure(configure); feature.Module.Configure(configure);
return feature; return feature;
} }
public static HttpFeature UseCache(this HttpFeature feature,
Microsoft.Extensions.Configuration.IConfigurationSection section, Action<HttpCacheFeature>? configure = null)
{
return feature.UseCache(cache =>
{
cache.UseConfiguration(section);
configure?.Invoke(cache);
});
}
} }
@@ -1,4 +1,5 @@
using Elsa.Features.Abstractions; using Elsa.Features.Abstractions;
using Elsa.Caching.Features;
using Elsa.Features.Attributes; using Elsa.Features.Attributes;
using Elsa.Features.Services; using Elsa.Features.Services;
using Elsa.Http.Handlers; using Elsa.Http.Handlers;
@@ -13,7 +14,7 @@ namespace Elsa.Http.Features;
/// </summary> /// </summary>
[DependsOn(typeof(HttpFeature))] [DependsOn(typeof(HttpFeature))]
[DependsOn(typeof(CachingWorkflowDefinitionsFeature))] [DependsOn(typeof(CachingWorkflowDefinitionsFeature))]
public class HttpCacheFeature : FeatureBase public class HttpCacheFeature : CacheConsumerFeature
{ {
/// <inheritdoc /> /// <inheritdoc />
public HttpCacheFeature(IModule module) : base(module) public HttpCacheFeature(IModule module) : base(module)
@@ -167,6 +167,8 @@ public class HttpFeature(IModule module) : FeatureBase(module)
.AddScoped<IRouteTable, RouteTable>() .AddScoped<IRouteTable, RouteTable>()
.AddScoped<IAbsoluteUrlProvider, DefaultAbsoluteUrlProvider>() .AddScoped<IAbsoluteUrlProvider, DefaultAbsoluteUrlProvider>()
.AddScoped<IRouteTableUpdater, DefaultRouteTableUpdater>() .AddScoped<IRouteTableUpdater, DefaultRouteTableUpdater>()
.AddScoped<RouteTableRevisionRefresher>()
.AddSingleton<RouteTableRevisionState>()
.AddScoped<IHttpWorkflowLookupService, HttpWorkflowLookupService>() .AddScoped<IHttpWorkflowLookupService, HttpWorkflowLookupService>()
.AddScoped(ContentTypeProvider) .AddScoped(ContentTypeProvider)
.AddHttpContextAccessor() .AddHttpContextAccessor()
@@ -36,6 +36,8 @@ public class HttpWorkflowsMiddleware(RequestDelegate next, IOptions<HttpActivity
public async Task InvokeAsync(HttpContext httpContext, IServiceProvider serviceProvider) public async Task InvokeAsync(HttpContext httpContext, IServiceProvider serviceProvider)
{ {
var path = httpContext.Request.Path.Value!.NormalizeRoute(); var path = httpContext.Request.Path.Value!.NormalizeRoute();
await serviceProvider.GetRequiredService<Elsa.Http.Services.RouteTableRevisionRefresher>()
.EnsureFreshAsync(httpContext.RequestAborted);
var matchingPath = GetMatchingRoute(serviceProvider, path).Route; var matchingPath = GetMatchingRoute(serviceProvider, path).Route;
var basePath = options.Value.BasePath?.ToString().NormalizeRoute(); var basePath = options.Value.BasePath?.ToString().NormalizeRoute();
@@ -11,7 +11,8 @@ namespace Elsa.Http.Services;
public class CachingHttpWorkflowLookupService( public class CachingHttpWorkflowLookupService(
IHttpWorkflowLookupService decoratedService, IHttpWorkflowLookupService decoratedService,
IHttpWorkflowsCacheManager cacheManager, IHttpWorkflowsCacheManager cacheManager,
ITenantAccessor tenantAccessor) : IHttpWorkflowLookupService ITenantAccessor tenantAccessor,
Elsa.Workflows.Management.IWorkflowDefinitionService definitionService) : IHttpWorkflowLookupService
{ {
/// <inheritdoc /> /// <inheritdoc />
public async Task<HttpWorkflowLookupResult?> FindWorkflowAsync(string bookmarkHash, CancellationToken cancellationToken = default) public async Task<HttpWorkflowLookupResult?> FindWorkflowAsync(string bookmarkHash, CancellationToken cancellationToken = default)
@@ -21,6 +22,21 @@ public class CachingHttpWorkflowLookupService(
var tenantIdPrefix = !string.IsNullOrEmpty(tenantId) ? $"{tenantId}:" : string.Empty; var tenantIdPrefix = !string.IsNullOrEmpty(tenantId) ? $"{tenantId}:" : string.Empty;
var key = $"{tenantIdPrefix}http-workflow:{bookmarkHash}"; var key = $"{tenantIdPrefix}http-workflow:{bookmarkHash}";
var cache = cacheManager.Cache; var cache = cacheManager.Cache;
if (!cache.IsEnabled) return await decoratedService.FindWorkflowAsync(bookmarkHash, cancellationToken);
if (cache.IsShared)
{
// Redis stores the definition version reference and trigger data, never an executable graph.
var entryOptions = new Elsa.Caching.Models.CacheEntryOptions { SlidingExpiration = cache.CachingOptions.Value.CacheDuration };
var cached = await cache.GetOrCreateAsync(key, async () =>
{
var found = await decoratedService.FindWorkflowAsync(bookmarkHash, cancellationToken);
return found == null ? null : new CachedHttpLookup(found.WorkflowGraph?.Workflow.Identity.Id, found.Triggers);
}, entryOptions, cancellationToken);
if (cached == null) return null;
var graph = cached.DefinitionVersionId == null ? null
: await definitionService.FindWorkflowGraphAsync(cached.DefinitionVersionId, cancellationToken);
return new HttpWorkflowLookupResult(graph, cached.Triggers);
}
return await cache.GetOrCreateAsync(key, async entry => return await cache.GetOrCreateAsync(key, async entry =>
{ {
var cachingOptions = cache.CachingOptions.Value; var cachingOptions = cache.CachingOptions.Value;
@@ -43,4 +59,5 @@ public class CachingHttpWorkflowLookupService(
return result; return result;
}); });
} }
public record CachedHttpLookup(string? DefinitionVersionId, ICollection<Elsa.Workflows.Runtime.Entities.StoredTrigger> Triggers);
} }
@@ -23,17 +23,29 @@ public class DefaultRouteTableUpdater(
var stimulusName = HttpStimulusNames.HttpEndpoint; var stimulusName = HttpStimulusNames.HttpEndpoint;
var triggerFilter = new TriggerFilter var triggerFilter = new TriggerFilter
{ {
Name = stimulusName Name = stimulusName,
TenantAgnostic = true
}; };
var bookmarkFilter = new BookmarkFilter var bookmarkFilter = new BookmarkFilter
{ {
Name = stimulusName Name = stimulusName,
TenantAgnostic = true
}; };
var triggers = (await triggerStore.FindManyAsync(triggerFilter, cancellationToken)).ToList(); var triggers = (await triggerStore.FindManyAsync(triggerFilter, cancellationToken)).ToList();
var bookmarks = (await bookmarkStore.FindManyAsync(bookmarkFilter, cancellationToken)).ToList(); var bookmarks = (await bookmarkStore.FindManyAsync(bookmarkFilter, cancellationToken)).ToList();
await AddRoutesAsync(triggers, cancellationToken); var routes = new List<HttpRouteData>();
await AddRoutesAsync(bookmarks, cancellationToken); foreach (var trigger in Filter(triggers))
{
var context = new HttpEndpointRouteProviderContext(trigger.GetPayload<HttpEndpointBookmarkPayload>(), trigger.TenantId, cancellationToken);
routes.AddRange(await httpEndpointRoutesProvider.GetRoutesAsync(context));
}
foreach (var bookmark in Filter(bookmarks))
{
var context = new HttpEndpointRouteProviderContext(bookmark.GetPayload<HttpEndpointBookmarkPayload>(), bookmark.TenantId, cancellationToken);
routes.AddRange(await httpEndpointRoutesProvider.GetRoutesAsync(context));
}
routeTable.Replace(routes);
} }
public async Task AddRoutesAsync(IEnumerable<StoredTrigger> triggers, CancellationToken cancellationToken = default) public async Task AddRoutesAsync(IEnumerable<StoredTrigger> triggers, CancellationToken cancellationToken = default)
@@ -12,6 +12,13 @@ public class RouteTable(IMemoryCache cache, ILogger<RouteTable> logger) : IRoute
private static readonly object Key = new(); private static readonly object Key = new();
private ConcurrentDictionary<string, HttpRouteData> Routes => cache.GetOrCreate(Key, _ => new ConcurrentDictionary<string, HttpRouteData>())!; private ConcurrentDictionary<string, HttpRouteData> Routes => cache.GetOrCreate(Key, _ => new ConcurrentDictionary<string, HttpRouteData>())!;
public void Replace(IEnumerable<HttpRouteData> routes)
{
var snapshot = new ConcurrentDictionary<string, HttpRouteData>();
foreach (var route in routes.Where(x => !x.Route.Contains("//")))
snapshot.TryAdd(route.Route.NormalizeRoute(), route);
cache.Set(Key, snapshot);
}
/// <inheritdoc /> /// <inheritdoc />
public void Add(string route) public void Add(string route)
@@ -0,0 +1,34 @@
using Elsa.Caching;
namespace Elsa.Http.Services;
/// <summary>A local route snapshot with a revision checked against SQL on each inbound workflow request.</summary>
public class RouteTableRevisionState : IDisposable
{
public SemaphoreSlim Gate { get; } = new(1, 1);
public string? Revision { get; set; }
public void Dispose() => Gate.Dispose();
}
public class RouteTableRevisionRefresher(ICacheManager cache, IRouteTableUpdater updater, RouteTableRevisionState state)
{
public async Task EnsureFreshAsync(CancellationToken cancellationToken)
{
if (!cache.IsShared) return;
// A process-wide route snapshot must not be built from a request's ambient transaction snapshot.
using var suppressed = new System.Transactions.TransactionScope(System.Transactions.TransactionScopeOption.Suppress,
System.Transactions.TransactionScopeAsyncFlowOption.Enabled);
var before = await cache.GetConsistencyStateAsync(cancellationToken);
if (!before.HasPendingRouteMutations && (before.RouteRevision ?? before.Revision) == state.Revision) return;
await state.Gate.WaitAsync(cancellationToken);
try
{
before = await cache.GetConsistencyStateAsync(cancellationToken);
if (!before.HasPendingRouteMutations && (before.RouteRevision ?? before.Revision) == state.Revision) return;
await updater.UpdateAsync(cancellationToken);
var after = await cache.GetConsistencyStateAsync(cancellationToken);
state.Revision = !after.HasPendingRouteMutations && (before.RouteRevision ?? before.Revision) == (after.RouteRevision ?? after.Revision) ? after.RouteRevision ?? after.Revision : null;
}
finally { state.Gate.Release(); }
}
}
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.IO.Compression</PackageId> <PackageId>Caspian.Elsa.IO.Compression</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides compression and archiving activities for Elsa Workflows. Provides compression and archiving activities for Elsa Workflows.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.IO.Http</PackageId> <PackageId>Caspian.Elsa.IO.Http</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides http capabilities to IO modules in Elsa Workflows. Provides http capabilities to IO modules in Elsa Workflows.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.IO</PackageId> <PackageId>Caspian.Elsa.IO</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides IO services for resolving various content types to streams. Provides IO services for resolving various content types to streams.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Identity</PackageId> <PackageId>Caspian.Elsa.Identity</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides a default identity solution. Provides a default identity solution.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.JavaScript.Libraries</PackageId> <PackageId>Caspian.Elsa.JavaScript.Libraries</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides the following JavaScript libraries for Elsa: Provides the following JavaScript libraries for Elsa:
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.JavaScript</PackageId> <PackageId>Caspian.Elsa.JavaScript</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides a JavaScript expression provider. Provides a JavaScript expression provider.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Kafka</PackageId> <PackageId>Caspian.Elsa.Kafka</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides Apache Kafka integration and activities. Provides Apache Kafka integration and activities.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.KeyValues</PackageId> <PackageId>Caspian.Elsa.KeyValues</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides a store for holding key value records. Provides a store for holding key value records.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Labels</PackageId> <PackageId>Caspian.Elsa.Labels</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides funtionality to tag workflows with labels and filter by them. Provides funtionality to tag workflows with labels and filter by them.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.Liquid</PackageId> <PackageId>Caspian.Elsa.Liquid</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides a Liquid expression provider. Provides a Liquid expression provider.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.MassTransit.AzureServiceBus</PackageId> <PackageId>Caspian.Elsa.MassTransit.AzureServiceBus</PackageId>
<Company>Caspian.</Company> <Company>Caspian.</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides integration of Azure Service Bus with MassTransit's integration of Elsa. Provides integration of Azure Service Bus with MassTransit's integration of Elsa.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.MassTransit.RabbitMq</PackageId> <PackageId>Caspian.Elsa.MassTransit.RabbitMq</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides integration of RabbitMQ with MassTransit's integration of Elsa. Provides integration of RabbitMQ with MassTransit's integration of Elsa.
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.MassTransit</PackageId> <PackageId>Caspian.Elsa.MassTransit</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides integration with MassTransit and comes with a MassTransit implementation of Elsa's IServiceBus abstraction. Provides integration with MassTransit and comes with a MassTransit implementation of Elsa's IServiceBus abstraction.
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.MongoDb</PackageId> <PackageId>Caspian.Elsa.MongoDb</PackageId>
<Company>Caspian</Company> <Company>Caspian</Company>
<Authors>Behzad Alizadeh</Authors> <Authors>Behzad Alizadeh</Authors>
<Version>1.1.0-preview.8</Version> <Version>1.1.0-preview.11</Version>
<GeneratePackageOnBuild>True</GeneratePackageOnBuild> <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
<Description> <Description>
Provides MongoDB implementations of various abstractions from various modules. Provides MongoDB implementations of various abstractions from various modules.

Some files were not shown because too many files have changed in this diff Show More