From efe4700bd1a2a539d25f064ed2161057f09188bc Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Sat, 16 May 2026 13:50:34 +0200 Subject: [PATCH] Harden EF trigger persistence --- .../AssemblyInfo.cs | 3 + .../DbExceptionClassifier.cs | 143 ++++++++++++++ .../Elsa.Persistence.EFCore.Common/Store.cs | 175 ++++++++++++------ .../Modules/Runtime/TriggerStore.cs | 81 +++++++- .../Services/DrainTriggerExecutor.cs | 8 + .../Quiescence/DrainOrchestratorWaitTests.cs | 31 ++++ 6 files changed, 381 insertions(+), 60 deletions(-) create mode 100644 src/modules/Elsa.Persistence.EFCore.Common/AssemblyInfo.cs create mode 100644 src/modules/Elsa.Persistence.EFCore.Common/DbExceptionClassifier.cs diff --git a/src/modules/Elsa.Persistence.EFCore.Common/AssemblyInfo.cs b/src/modules/Elsa.Persistence.EFCore.Common/AssemblyInfo.cs new file mode 100644 index 000000000..77206ee4c --- /dev/null +++ b/src/modules/Elsa.Persistence.EFCore.Common/AssemblyInfo.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Elsa.Persistence.EFCore")] diff --git a/src/modules/Elsa.Persistence.EFCore.Common/DbExceptionClassifier.cs b/src/modules/Elsa.Persistence.EFCore.Common/DbExceptionClassifier.cs new file mode 100644 index 000000000..8f8e1e677 --- /dev/null +++ b/src/modules/Elsa.Persistence.EFCore.Common/DbExceptionClassifier.cs @@ -0,0 +1,143 @@ +namespace Elsa.Persistence.EFCore; + +internal static class DbExceptionClassifier +{ + private static readonly HashSet SqlServerTransientErrorNumbers = + [ + -2, + 64, + 233, + 1205, + 4060, + 10928, + 10929, + 40197, + 40501, + 40613, + 49918, + 49919, + 49920, + ]; + + public static bool IsSqlServerTransient(string providerName, Exception exception) + { + if (!providerName.Contains("SqlServer", StringComparison.OrdinalIgnoreCase)) + return false; + + return EnumerateExceptions(exception).Any(IsSqlServerTransientException); + } + + public static bool IsDuplicateKey(Exception exception) + { + return EnumerateExceptions(exception).Any(IsDuplicateKeyException); + } + + private static bool IsSqlServerTransientException(Exception exception) + { + if (!IsSqlClientException(exception)) + return false; + + return GetErrorNumbers(exception).Any(SqlServerTransientErrorNumbers.Contains) + || exception.Message.Contains("deadlock", StringComparison.OrdinalIgnoreCase); + } + + private static bool IsDuplicateKeyException(Exception exception) + { + var type = exception.GetType(); + var typeName = type.Name; + var typeNamespace = type.Namespace ?? string.Empty; + var errorNumbers = GetErrorNumbers(exception).ToList(); + + if (IsSqlClientException(exception) && errorNumbers.Any(number => number is 2601 or 2627)) + return true; + + if (typeName.Contains("MySql", StringComparison.OrdinalIgnoreCase) && errorNumbers.Contains(1062)) + return true; + + if (typeName.Contains("Sqlite", StringComparison.OrdinalIgnoreCase) && errorNumbers.Any(number => number is 19 or 1555 or 2067)) + return true; + + if (typeName.Contains("Oracle", StringComparison.OrdinalIgnoreCase) && errorNumbers.Contains(1)) + return true; + + if (GetStringProperty(exception, "SqlState") == "23505") + return true; + + return typeNamespace.Contains("Data", StringComparison.OrdinalIgnoreCase) + && (exception.Message.Contains("duplicate key", StringComparison.OrdinalIgnoreCase) + || exception.Message.Contains("unique constraint", StringComparison.OrdinalIgnoreCase) + || exception.Message.Contains("UNIQUE constraint failed", StringComparison.OrdinalIgnoreCase) + || exception.Message.Contains("ORA-00001", StringComparison.OrdinalIgnoreCase)); + } + + private static bool IsSqlClientException(Exception exception) + { + var type = exception.GetType(); + return type.Name.Equals("SqlException", StringComparison.OrdinalIgnoreCase) + && type.Namespace?.Contains("SqlClient", StringComparison.OrdinalIgnoreCase) == true; + } + + private static IEnumerable GetErrorNumbers(object source) + { + if (GetIntProperty(source, "Number") is { } number) + yield return number; + + if (GetIntProperty(source, "SqliteErrorCode") is { } sqliteErrorCode) + yield return sqliteErrorCode; + + if (GetIntProperty(source, "SqliteExtendedErrorCode") is { } sqliteExtendedErrorCode) + yield return sqliteExtendedErrorCode; + + var errors = source.GetType().GetProperty("Errors")?.GetValue(source); + if (errors is not System.Collections.IEnumerable errorCollection) + yield break; + + foreach (var error in errorCollection) + { + if (error is null) + continue; + + if (GetIntProperty(error, "Number") is { } errorNumber) + yield return errorNumber; + } + } + + private static int? GetIntProperty(object source, string name) + { + var value = source.GetType().GetProperty(name)?.GetValue(source); + return value switch + { + int number => number, + short number => number, + long number when number is >= int.MinValue and <= int.MaxValue => (int)number, + _ => null + }; + } + + private static string? GetStringProperty(object source, string name) + { + return source.GetType().GetProperty(name)?.GetValue(source) as string; + } + + private static IEnumerable EnumerateExceptions(Exception exception) + { + var stack = new Stack(); + stack.Push(exception); + + while (stack.Count > 0) + { + var current = stack.Pop(); + yield return current; + + if (current is AggregateException aggregateException) + { + foreach (var inner in aggregateException.InnerExceptions) + stack.Push(inner); + } + else if (current.InnerException is not null) + { + stack.Push(current.InnerException); + } + } + } +} diff --git a/src/modules/Elsa.Persistence.EFCore.Common/Store.cs b/src/modules/Elsa.Persistence.EFCore.Common/Store.cs index c8ddbdd43..900bb0c95 100644 --- a/src/modules/Elsa.Persistence.EFCore.Common/Store.cs +++ b/src/modules/Elsa.Persistence.EFCore.Common/Store.cs @@ -19,6 +19,9 @@ namespace Elsa.Persistence.EFCore; [PublicAPI] public class Store(IDbContextFactory dbContextFactory, IServiceProvider serviceProvider) where TDbContext : DbContext where TEntity : class, new() { + private const int SqlServerBulkWriteMaxRetryCount = 3; + private static readonly TimeSpan SqlServerBulkWriteBaseDelay = TimeSpan.FromMilliseconds(50); + // ReSharper disable once StaticMemberInGenericType // Justification: This is a static member that is used to ensure that only one thread can access the database for TEntity at a time. private static readonly SemaphoreSlim Semaphore = new(1, 1); @@ -81,20 +84,35 @@ public class Store(IDbContextFactory dbContextF Func? onSaving = null, CancellationToken cancellationToken = default) { - var entityList = entities.ToList(); + await Semaphore.WaitAsync(cancellationToken); - if (entityList.Count == 0) - return; - - await using var dbContext = await CreateDbContextAsync(cancellationToken); - - if (onSaving != null) + try { - var savingTasks = entityList.Select(entity => onSaving(dbContext, entity, cancellationToken).AsTask()).ToList(); - await Task.WhenAll(savingTasks); - } + var entityList = entities.ToList(); - await dbContext.BulkInsertAsync(entityList, cancellationToken); + if (entityList.Count == 0) + return; + + await ExecuteBulkWriteWithSqlServerRetryAsync(async (dbContext, ct) => + { + if (onSaving != null) + { + var savingTasks = entityList.Select(entity => onSaving(dbContext, entity, ct).AsTask()).ToList(); + await Task.WhenAll(savingTasks); + } + + await dbContext.BulkInsertAsync(entityList, ct); + }, cancellationToken); + } + catch (Exception ex) + { + await HandleDbExceptionAsync(ex, cancellationToken); + throw; + } + finally + { + Semaphore.Release(); + } } /// @@ -131,14 +149,7 @@ public class Store(IDbContextFactory dbContextF } catch (Exception ex) { - var handler = serviceProvider.GetService(); - - if (handler != null) - { - var context = new DbUpdateExceptionContext(ex, cancellationToken); - await handler.HandleAsync(context); - } - + await HandleDbExceptionAsync(ex, cancellationToken); throw; } finally @@ -168,53 +179,103 @@ public class Store(IDbContextFactory dbContextF Func? onSaving = null, CancellationToken cancellationToken = default) { - var entityList = entities.ToList(); - - if (entityList.Count == 0) - return; - - await using var dbContext = await CreateDbContextAsync(cancellationToken); - - if (onSaving != null) - { - var savingTasks = entityList.Select(entity => onSaving(dbContext, entity, cancellationToken).AsTask()).ToList(); - await Task.WhenAll(savingTasks); - } - - // When doing a custom SQL query (Bulk Upsert), none of the installed query filters will be applied. Hence, we are assigning the current tenant ID explicitly. - var tenantId = serviceProvider.GetRequiredService().TenantId; - foreach (var entity in entityList) - { - if (entity is Entity entityWithTenant) - { - // Don't touch tenant-agnostic entities (marked with "*") - if (entityWithTenant.TenantId == Tenant.AgnosticTenantId) - continue; - - // Apply current tenant ID to entities without one - if (entityWithTenant.TenantId == null && tenantId != null) - entityWithTenant.TenantId = tenantId; - } - } + await Semaphore.WaitAsync(cancellationToken); try { - await dbContext.BulkUpsertAsync(entityList, keySelector, cancellationToken); + var entityList = entities.ToList(); + + if (entityList.Count == 0) + return; + + var tenantId = serviceProvider.GetRequiredService().TenantId; + + await ExecuteBulkWriteWithSqlServerRetryAsync(async (dbContext, ct) => + { + if (onSaving != null) + { + var savingTasks = entityList.Select(entity => onSaving(dbContext, entity, ct).AsTask()).ToList(); + await Task.WhenAll(savingTasks); + } + + // When doing a custom SQL query (Bulk Upsert), none of the installed query filters will be applied. Hence, we are assigning the current tenant ID explicitly. + foreach (var entity in entityList) + { + if (entity is Entity entityWithTenant) + { + // Don't touch tenant-agnostic entities (marked with "*") + if (entityWithTenant.TenantId == Tenant.AgnosticTenantId) + continue; + + // Apply current tenant ID to entities without one + if (entityWithTenant.TenantId == null && tenantId != null) + entityWithTenant.TenantId = tenantId; + } + } + + await dbContext.BulkUpsertAsync(entityList, keySelector, ct); + }, cancellationToken); } catch (Exception ex) { - var handler = serviceProvider.GetService(); - - if (handler != null) - { - var context = new DbUpdateExceptionContext(ex, cancellationToken); - await handler.HandleAsync(context); - } - + await HandleDbExceptionAsync(ex, cancellationToken); throw; } + finally + { + Semaphore.Release(); + } } + private async Task HandleDbExceptionAsync(Exception exception, CancellationToken cancellationToken) + { + var handler = serviceProvider.GetService(); + + if (handler == null) + return; + + var context = new DbUpdateExceptionContext(exception, cancellationToken); + await handler.HandleAsync(context); + } + + private async Task ExecuteBulkWriteWithSqlServerRetryAsync( + Func operation, + CancellationToken cancellationToken) + { + for (var attempt = 0;; attempt++) + { + var providerName = string.Empty; + + try + { + await using var dbContext = await CreateDbContextAsync(cancellationToken); + providerName = dbContext.Database.ProviderName ?? string.Empty; + await operation(dbContext, cancellationToken); + return; + } + catch (Exception ex) + { + if (ShouldRetrySqlServerBulkWrite(providerName, ex, attempt, cancellationToken)) + { + await Task.Delay(GetSqlServerBulkWriteRetryDelay(attempt), cancellationToken); + continue; + } + + throw; + } + } + } + + private static bool ShouldRetrySqlServerBulkWrite(string providerName, Exception exception, int attempt, CancellationToken cancellationToken) + { + return attempt < SqlServerBulkWriteMaxRetryCount + && !cancellationToken.IsCancellationRequested + && exception is not OperationCanceledException + && DbExceptionClassifier.IsSqlServerTransient(providerName, exception); + } + + private static TimeSpan GetSqlServerBulkWriteRetryDelay(int attempt) => TimeSpan.FromMilliseconds(SqlServerBulkWriteBaseDelay.TotalMilliseconds * (attempt + 1)); + /// /// Updates the entity. /// @@ -633,4 +694,4 @@ public class Store(IDbContextFactory dbContextF .Distinct() .CountAsync(cancellationToken); } -} \ No newline at end of file +} diff --git a/src/modules/Elsa.Persistence.EFCore/Modules/Runtime/TriggerStore.cs b/src/modules/Elsa.Persistence.EFCore/Modules/Runtime/TriggerStore.cs index e62c5ba29..b160b94e9 100644 --- a/src/modules/Elsa.Persistence.EFCore/Modules/Runtime/TriggerStore.cs +++ b/src/modules/Elsa.Persistence.EFCore/Modules/Runtime/TriggerStore.cs @@ -1,5 +1,6 @@ using Elsa.Common.Entities; using Elsa.Common.Models; +using Elsa.Common.Multitenancy; using Elsa.Extensions; using Elsa.Workflows; using Elsa.Workflows.Runtime; @@ -15,6 +16,7 @@ namespace Elsa.Persistence.EFCore.Modules.Runtime; [UsedImplicitly] public class EFCoreTriggerStore( EntityStore store, + ITenantAccessor tenantAccessor, IPayloadSerializer serializer) : ITriggerStore { /// @@ -57,14 +59,40 @@ public class EFCoreTriggerStore( public async ValueTask ReplaceAsync(IEnumerable removed, IEnumerable added, CancellationToken cancellationToken = default) { var removedList = removed.ToList(); + var addedList = added.ToList(); - if(removedList.Count > 0) + foreach (var trigger in addedList) + ApplyCurrentTenant(trigger); + + addedList = DistinctByLogicalKey(addedList).ToList(); + + if (removedList.Count > 0) { var filter = new TriggerFilter { Ids = removedList.Select(r => r.Id).ToList() }; await DeleteManyAsync(filter, cancellationToken); } - await store.SaveManyAsync(added, OnSaveAsync, cancellationToken); + if (addedList.Count == 0) + return; + + var newTriggers = await GetMissingLogicalTriggersAsync(addedList, cancellationToken); + + if (newTriggers.Count == 0) + return; + + try + { + await store.SaveManyAsync(newTriggers, OnSaveAsync, cancellationToken); + } + catch (Exception ex) when (DbExceptionClassifier.IsDuplicateKey(ex)) + { + var remainingTriggers = await GetMissingLogicalTriggersAsync(newTriggers, cancellationToken); + + if (remainingTriggers.Count == 0) + return; + + await store.SaveManyAsync(remainingTriggers, OnSaveAsync, cancellationToken); + } } /// @@ -89,4 +117,51 @@ public class EFCoreTriggerStore( return ValueTask.CompletedTask; } -} \ No newline at end of file + + private async Task> GetExistingLogicalKeysAsync(ICollection triggers, CancellationToken cancellationToken) + { + var workflowDefinitionIds = triggers.Select(x => x.WorkflowDefinitionId).Distinct().ToList(); + var existingTriggers = await store.QueryAsync( + queryable => queryable.Where(trigger => workflowDefinitionIds.Contains(trigger.WorkflowDefinitionId)), + cancellationToken); + + return existingTriggers + .Select(GetLogicalKey) + .ToHashSet(StringComparer.Ordinal); + } + + private async Task> GetMissingLogicalTriggersAsync(ICollection triggers, CancellationToken cancellationToken) + { + var existingKeys = await GetExistingLogicalKeysAsync(triggers, cancellationToken); + return triggers + .Where(trigger => !existingKeys.Contains(GetLogicalKey(trigger))) + .ToList(); + } + + private void ApplyCurrentTenant(StoredTrigger trigger) + { + if (trigger.TenantId == Tenant.AgnosticTenantId) + return; + + trigger.TenantId ??= tenantAccessor.TenantId; + } + + private static IEnumerable DistinctByLogicalKey(IEnumerable triggers) + { + var seen = new HashSet(StringComparer.Ordinal); + + foreach (var trigger in triggers) + { + if (seen.Add(GetLogicalKey(trigger))) + yield return trigger; + } + } + + private static string GetLogicalKey(StoredTrigger trigger) => + string.Join( + '\u001f', + trigger.WorkflowDefinitionId, + trigger.Hash, + trigger.ActivityId, + trigger.TenantId); +} diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DrainTriggerExecutor.cs b/src/modules/Elsa.Workflows.Runtime/Services/DrainTriggerExecutor.cs index 176cb5ec9..0421cdb7c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DrainTriggerExecutor.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DrainTriggerExecutor.cs @@ -53,6 +53,14 @@ internal static class DrainTriggerExecutor contextLabel, outcome.OverallResult, outcome.PausePhaseDuration, outcome.WaitPhaseDuration); } } + catch (ObjectDisposedException ex) when (cancellationToken.IsCancellationRequested) + { + logger.LogInformation(ex, "{Context} stopped because shutdown disposed a drain dependency after cancellation.", contextLabel); + } + catch (ObjectDisposedException) + { + throw; + } catch (InvalidOperationException ex) { // Parallel non-force drain rejected by the orchestrator — another trigger already drained diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs index 3266426e0..54b63636e 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/DrainOrchestratorWaitTests.cs @@ -1,5 +1,7 @@ using Elsa.Workflows.Management.Entities; using Elsa.Workflows.Management.Filters; +using Elsa.Workflows.Runtime.HostedServices; +using Microsoft.Extensions.Logging; using NSubstitute; namespace Elsa.Workflows.Runtime.UnitTests.Quiescence; @@ -97,4 +99,33 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase // Same payload modulo the WasCached flag. Assert.Equal(first with { WasCached = true }, second); } + + [Fact(DisplayName = "Host stop drain swallows ObjectDisposedException after shutdown cancellation")] + public async Task HostStopDrainSwallowsObjectDisposedExceptionAfterCancellation() + { + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + var orchestrator = Substitute.For(); + orchestrator.DrainAsync(DrainTrigger.HostStopSignal, cts.Token).Returns(_ => ThrowObjectDisposedAsync()); + var hostedService = new DrainOrchestratorHostedService(orchestrator, Substitute.For>()); + + await hostedService.StopAsync(cts.Token); + } + + [Fact(DisplayName = "Host stop drain propagates ObjectDisposedException before shutdown cancellation")] + public async Task HostStopDrainPropagatesObjectDisposedExceptionBeforeCancellation() + { + var orchestrator = Substitute.For(); + orchestrator.DrainAsync(DrainTrigger.HostStopSignal, CancellationToken.None).Returns(_ => ThrowObjectDisposedAsync()); + var hostedService = new DrainOrchestratorHostedService(orchestrator, Substitute.For>()); + + await Assert.ThrowsAsync(() => hostedService.StopAsync(CancellationToken.None)); + } + + private static async ValueTask ThrowObjectDisposedAsync() + { + await Task.Yield(); + throw new ObjectDisposedException("drain dependency"); + } }