Harden EF trigger persistence

src/modules/Elsa.Persistence.EFCore.Common/AssemblyInfo.cs

@@ -0,0 +1,3 @@
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("Elsa.Persistence.EFCore")]

src/modules/Elsa.Persistence.EFCore.Common/DbExceptionClassifier.cs

@@ -0,0 +1,143 @@
+namespace Elsa.Persistence.EFCore;
+
+internal static class DbExceptionClassifier
+{
+    private static readonly HashSet<int> 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<int> 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<Exception> EnumerateExceptions(Exception exception)
+    {
+        var stack = new Stack<Exception>();
+        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);
+            }
+        }
+    }
+}

src/modules/Elsa.Persistence.EFCore.Common/Store.cs

@@ -19,6 +19,9 @@ namespace Elsa.Persistence.EFCore;
 [PublicAPI]
 public class Store<TDbContext, TEntity>(IDbContextFactory<TDbContext> 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<TDbContext, TEntity>(IDbContextFactory<TDbContext> dbContextF
         Func<TDbContext, TEntity, CancellationToken, ValueTask>? 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();
+        }
     }
 
     /// <summary>
@@ -131,14 +149,7 @@ public class Store<TDbContext, TEntity>(IDbContextFactory<TDbContext> dbContextF
         }
         catch (Exception ex)
         {
-            var handler = serviceProvider.GetService<IDbExceptionHandler>();
-
-            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<TDbContext, TEntity>(IDbContextFactory<TDbContext> dbContextF
         Func<TDbContext, TEntity, CancellationToken, ValueTask>? 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<ITenantAccessor>().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<ITenantAccessor>().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<IDbExceptionHandler>();
-
-            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<IDbExceptionHandler>();
+
+        if (handler == null)
+            return;
+
+        var context = new DbUpdateExceptionContext(exception, cancellationToken);
+        await handler.HandleAsync(context);
+    }
+
+    private async Task ExecuteBulkWriteWithSqlServerRetryAsync(
+        Func<TDbContext, CancellationToken, Task> 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));
+
     /// <summary>
     /// Updates the entity.
     /// </summary>
@@ -633,4 +694,4 @@ public class Store<TDbContext, TEntity>(IDbContextFactory<TDbContext> dbContextF
             .Distinct()
             .CountAsync(cancellationToken);
     }
-}
+}

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<RuntimeElsaDbContext, StoredTrigger> store,
+    ITenantAccessor tenantAccessor,
     IPayloadSerializer serializer) : ITriggerStore
 {
     /// <inheritdoc />
@@ -57,14 +59,40 @@ public class EFCoreTriggerStore(
     public async ValueTask ReplaceAsync(IEnumerable<StoredTrigger> removed, IEnumerable<StoredTrigger> 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);
+        }
     }
 
     /// <inheritdoc />
@@ -89,4 +117,51 @@ public class EFCoreTriggerStore(
 
         return ValueTask.CompletedTask;
     }
-}
+
+    private async Task<HashSet<string>> GetExistingLogicalKeysAsync(ICollection<StoredTrigger> 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<List<StoredTrigger>> GetMissingLogicalTriggersAsync(ICollection<StoredTrigger> 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<StoredTrigger> DistinctByLogicalKey(IEnumerable<StoredTrigger> triggers)
+    {
+        var seen = new HashSet<string>(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);
+}

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

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<IDrainOrchestrator>();
+        orchestrator.DrainAsync(DrainTrigger.HostStopSignal, cts.Token).Returns(_ => ThrowObjectDisposedAsync());
+        var hostedService = new DrainOrchestratorHostedService(orchestrator, Substitute.For<ILogger<DrainOrchestratorHostedService>>());
+
+        await hostedService.StopAsync(cts.Token);
+    }
+
+    [Fact(DisplayName = "Host stop drain propagates ObjectDisposedException before shutdown cancellation")]
+    public async Task HostStopDrainPropagatesObjectDisposedExceptionBeforeCancellation()
+    {
+        var orchestrator = Substitute.For<IDrainOrchestrator>();
+        orchestrator.DrainAsync(DrainTrigger.HostStopSignal, CancellationToken.None).Returns(_ => ThrowObjectDisposedAsync());
+        var hostedService = new DrainOrchestratorHostedService(orchestrator, Substitute.For<ILogger<DrainOrchestratorHostedService>>());
+
+        await Assert.ThrowsAsync<ObjectDisposedException>(() => hostedService.StopAsync(CancellationToken.None));
+    }
+
+    private static async ValueTask<DrainOutcome> ThrowObjectDisposedAsync()
+    {
+        await Task.Yield();
+        throw new ObjectDisposedException("drain dependency");
+    }
 }