From e4478af2be9005b5fd0a831059b0b4725f035005 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Mon, 14 Sep 2026 02:51:34 +0200 Subject: [PATCH] Fix Alterations ownership parity and MySQL retry (#8133) Closes #8125 --- .../Stores/MemoryAlterationJobStore.cs | 31 +++++- .../Stores/MemoryAlterationPlanStore.cs | 11 +- .../DbExceptionClassifier.cs | 30 ++++- .../Elsa.Persistence.EFCore.Common/Store.cs | 27 +++-- .../Modules/Alterations/AlterationJobStore.cs | 13 ++- ...CoreAlterationStoreTenantOwnershipTests.cs | 90 ++++++++++----- ...yAlterationJobStoreTenantIsolationTests.cs | 102 +++++++++++++++++ ...AlterationPlanStoreTenantIsolationTests.cs | 31 ++++++ .../DbExceptionClassifierTests.cs | 34 ++++++ .../FakeMySqlException.cs | 6 + .../StoreWriteRetryTests.cs | 103 ++++++++++++++++++ 11 files changed, 425 insertions(+), 53 deletions(-) create mode 100644 test/unit/Elsa.Persistence.EFCore.UnitTests/FakeMySqlException.cs create mode 100644 test/unit/Elsa.Persistence.EFCore.UnitTests/StoreWriteRetryTests.cs diff --git a/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationJobStore.cs b/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationJobStore.cs index 7f5032616..0c5d4edd0 100644 --- a/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationJobStore.cs +++ b/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationJobStore.cs @@ -52,8 +52,9 @@ public class MemoryAlterationJobStore : IAlterationJobStore foreach (var job in list) ApplyCurrentTenant(job); + var tenantIdsById = new Dictionary(StringComparer.Ordinal); foreach (var job in list) - EnsureIdAvailable(job); + EnsureIdAvailable(job, tenantIdsById); _store.SaveMany(list, x => x.Id); } @@ -98,19 +99,41 @@ public class MemoryAlterationJobStore : IAlterationJobStore private string CurrentTenantId => _tenantAccessor?.TenantId ?? Tenant.DefaultTenantId; - private void EnsureIdAvailable(AlterationJob job) + private void EnsureIdAvailable(AlterationJob job, IDictionary? tenantIdsById = null) { + if (tenantIdsById?.TryGetValue(job.Id, out var stagedTenantId) == true) + { + if (!CanReplace(stagedTenantId, job.TenantId)) + throw AlterationStoreConflict.HiddenJobId(job.Id); + + // A repeated ID in one batch is an update of the row established by the + // earlier item, even when that row was absent before the batch started. + job.TenantId = stagedTenantId; + return; + } + var existing = _store.Find(x => x.Id == job.Id); - if (existing is not null && !CanReplace(existing)) + if (existing is null) + { + tenantIdsById?.Add(job.Id, job.TenantId); + return; + } + + if (!CanReplace(existing.TenantId, job.TenantId)) throw AlterationStoreConflict.HiddenJobId(job.Id); + + // An accepted update may change the payload, but it must not rehome the row. + job.TenantId = existing.TenantId; + tenantIdsById?.Add(job.Id, existing.TenantId); } /// /// * is visible to every tenant, but only an agnostic writer may replace it. /// Named tenants may upsert their own visible rows. /// - private bool CanReplace(Entity existing) => TenantVisibility.CanReplace(existing.TenantId, CurrentTenantId); + private bool CanReplace(string? existingTenantId, string? incomingTenantId) => + TenantVisibility.CanReplaceOwnedRow(existingTenantId, incomingTenantId, CurrentTenantId); private void ApplyCurrentTenant(Entity entity) { diff --git a/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationPlanStore.cs b/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationPlanStore.cs index 77646945f..78cee58bd 100644 --- a/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationPlanStore.cs +++ b/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationPlanStore.cs @@ -69,15 +69,22 @@ public class MemoryAlterationPlanStore : IAlterationPlanStore { var existing = _store.Find(x => x.Id == plan.Id); - if (existing is not null && !CanReplace(existing)) + if (existing is null) + return; + + if (!CanReplace(existing, plan)) throw AlterationStoreConflict.HiddenPlanId(plan.Id); + + // An accepted update may change the payload, but it must not rehome the row. + plan.TenantId = existing.TenantId; } /// /// * is visible to every tenant, but only an agnostic writer may replace it. /// Named tenants may upsert their own visible rows. /// - private bool CanReplace(Entity existing) => TenantVisibility.CanReplace(existing.TenantId, CurrentTenantId); + private bool CanReplace(Entity existing, Entity incoming) => + TenantVisibility.CanReplaceOwnedRow(existing.TenantId, incoming.TenantId, CurrentTenantId); private void ApplyCurrentTenant(Entity entity) { diff --git a/src/modules/Elsa.Persistence.EFCore.Common/DbExceptionClassifier.cs b/src/modules/Elsa.Persistence.EFCore.Common/DbExceptionClassifier.cs index 0e062f61d..ab01a9ca4 100644 --- a/src/modules/Elsa.Persistence.EFCore.Common/DbExceptionClassifier.cs +++ b/src/modules/Elsa.Persistence.EFCore.Common/DbExceptionClassifier.cs @@ -19,12 +19,21 @@ internal static class DbExceptionClassifier 49920, ]; - public static bool IsSqlServerTransient(string providerName, Exception exception) - { - if (!providerName.Contains("SqlServer", StringComparison.OrdinalIgnoreCase)) - return false; + private static readonly HashSet MySqlTransientErrorNumbers = + [ + 1205, // ER_LOCK_WAIT_TIMEOUT + 1213, // ER_LOCK_DEADLOCK + ]; - return EnumerateExceptions(exception).Any(IsSqlServerTransientException); + public static bool IsTransient(string providerName, Exception exception) + { + if (providerName.Contains("SqlServer", StringComparison.OrdinalIgnoreCase)) + return EnumerateExceptions(exception).Any(IsSqlServerTransientException); + + if (providerName.Contains("MySql", StringComparison.OrdinalIgnoreCase)) + return EnumerateExceptions(exception).Any(IsMySqlTransientException); + + return false; } public static bool IsDuplicateKey(Exception exception) @@ -41,6 +50,17 @@ internal static class DbExceptionClassifier || exception.Message.Contains("deadlock", StringComparison.OrdinalIgnoreCase); } + private static bool IsMySqlTransientException(Exception exception) + { + var type = exception.GetType(); + var typeNamespace = type.Namespace ?? string.Empty; + var isMySqlException = type.Name.Equals("MySqlException", StringComparison.OrdinalIgnoreCase) + && (typeNamespace.Equals("MySqlConnector", StringComparison.OrdinalIgnoreCase) + || typeNamespace.Equals("MySql.Data.MySqlClient", StringComparison.OrdinalIgnoreCase)); + + return isMySqlException && GetErrorNumbers(exception).Any(MySqlTransientErrorNumbers.Contains); + } + private static bool IsDuplicateKeyException(Exception exception) { var type = exception.GetType(); diff --git a/src/modules/Elsa.Persistence.EFCore.Common/Store.cs b/src/modules/Elsa.Persistence.EFCore.Common/Store.cs index 9ee59a48e..872ed31b6 100644 --- a/src/modules/Elsa.Persistence.EFCore.Common/Store.cs +++ b/src/modules/Elsa.Persistence.EFCore.Common/Store.cs @@ -19,8 +19,8 @@ namespace Elsa.Persistence.EFCore; [PublicAPI] public class Store(IDbContextFactory dbContextFactory, IServiceProvider serviceProvider) where TDbContext : DbContext where TEntity : class, new() { - private const int SqlServerWriteMaxRetryCount = 3; - private static readonly TimeSpan SqlServerWriteBaseDelay = TimeSpan.FromMilliseconds(50); + private const int WriteMaxRetryCount = 3; + private static readonly TimeSpan WriteBaseDelay = 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. @@ -93,7 +93,7 @@ public class Store(IDbContextFactory dbContextF if (entityList.Count == 0) return; - await ExecuteSqlServerWriteWithRetryAsync(async (dbContext, ct) => + await ExecuteWriteWithRetryAsync(async (dbContext, ct) => { if (onSaving != null) { @@ -190,7 +190,7 @@ public class Store(IDbContextFactory dbContextF var tenantId = serviceProvider.GetRequiredService().TenantId; - await ExecuteSqlServerWriteWithRetryAsync(async (dbContext, ct) => + await ExecuteWriteWithRetryAsync(async (dbContext, ct) => { if (onSaving != null) { @@ -263,7 +263,12 @@ public class Store(IDbContextFactory dbContextF } } - internal async Task ExecuteSqlServerWriteWithRetryAsync( + /// + /// Executes a whole database write operation with a fresh context after a provider + /// transient failure. The caller owns the transaction boundary so a retry never + /// resumes a partially completed transaction. + /// + internal async Task ExecuteWriteWithRetryAsync( Func operation, CancellationToken cancellationToken) { @@ -280,9 +285,9 @@ public class Store(IDbContextFactory dbContextF } catch (Exception ex) { - if (ShouldRetrySqlServerWrite(providerName, ex, attempt, cancellationToken)) + if (ShouldRetryWrite(providerName, ex, attempt, cancellationToken)) { - await Task.Delay(GetSqlServerWriteRetryDelay(attempt), cancellationToken); + await Task.Delay(GetWriteRetryDelay(attempt), cancellationToken); continue; } @@ -291,15 +296,15 @@ public class Store(IDbContextFactory dbContextF } } - private static bool ShouldRetrySqlServerWrite(string providerName, Exception exception, int attempt, CancellationToken cancellationToken) + private static bool ShouldRetryWrite(string providerName, Exception exception, int attempt, CancellationToken cancellationToken) { - return attempt < SqlServerWriteMaxRetryCount + return attempt < WriteMaxRetryCount && !cancellationToken.IsCancellationRequested && exception is not OperationCanceledException - && DbExceptionClassifier.IsSqlServerTransient(providerName, exception); + && DbExceptionClassifier.IsTransient(providerName, exception); } - private static TimeSpan GetSqlServerWriteRetryDelay(int attempt) => TimeSpan.FromMilliseconds(SqlServerWriteBaseDelay.TotalMilliseconds * (attempt + 1)); + private static TimeSpan GetWriteRetryDelay(int attempt) => TimeSpan.FromMilliseconds(WriteBaseDelay.TotalMilliseconds * (attempt + 1)); /// /// Updates the entity. diff --git a/src/modules/Elsa.Persistence.EFCore/Modules/Alterations/AlterationJobStore.cs b/src/modules/Elsa.Persistence.EFCore/Modules/Alterations/AlterationJobStore.cs index 4db2753ed..300c2573f 100644 --- a/src/modules/Elsa.Persistence.EFCore/Modules/Alterations/AlterationJobStore.cs +++ b/src/modules/Elsa.Persistence.EFCore/Modules/Alterations/AlterationJobStore.cs @@ -49,8 +49,15 @@ public class EFCoreAlterationJobStore : IAlterationJobStore return; } - await using var dbContext = await _store.CreateDbContextAsync(cancellationToken); - await UpsertAsync(dbContext, record, cancellationToken); + await _store.ExecuteWithDbExceptionHandlingAsync( + async () => + { + await using var dbContext = await _store.CreateDbContextAsync(cancellationToken); + await UpsertAsync(dbContext, record, cancellationToken, handleDbExceptions: false); + return true; + }, + cancellationToken, + IsDatabaseException); } /// @@ -69,7 +76,7 @@ public class EFCoreAlterationJobStore : IAlterationJobStore await _store.ExecuteWithDbExceptionHandlingAsync( async () => { - await _store.ExecuteSqlServerWriteWithRetryAsync(async (dbContext, ct) => + await _store.ExecuteWriteWithRetryAsync(async (dbContext, ct) => { await using var transaction = await dbContext.Database.BeginTransactionAsync(ct); diff --git a/test/integration/Elsa.Alterations.Persistence.ConformanceTests/EFCoreAlterationStoreTenantOwnershipTests.cs b/test/integration/Elsa.Alterations.Persistence.ConformanceTests/EFCoreAlterationStoreTenantOwnershipTests.cs index a821261ff..73df3a0ee 100644 --- a/test/integration/Elsa.Alterations.Persistence.ConformanceTests/EFCoreAlterationStoreTenantOwnershipTests.cs +++ b/test/integration/Elsa.Alterations.Persistence.ConformanceTests/EFCoreAlterationStoreTenantOwnershipTests.cs @@ -5,6 +5,7 @@ using Elsa.Alterations.Core.Filters; using Elsa.Alterations.Core.Models; using Elsa.Common.Multitenancy; using Elsa.Persistence.EFCore; +using Microsoft.Data.Sqlite; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Diagnostics; using Microsoft.Extensions.Logging; @@ -323,6 +324,25 @@ public sealed class EFCoreAlterationStoreTenantOwnershipTests Assert.Null(handler.Exception); } + [Fact] + public async Task SaveAsync_WhenJobIdIsHidden_DoesNotSendOwnershipConflictToDbExceptionHandler() + { + var handler = new RecordingDbExceptionHandler(); + await using var scenario = await AlterationStoreScenario.CreateSqliteAsync( + "tenant-a", + tenantsEnabled: true, + dbExceptionHandler: handler); + await scenario.Jobs.SaveAsync(Job("handler-conflict", "tenant-a", AlterationJobStatus.Pending, "owner")); + + using (scenario.UseTenant("tenant-b")) + { + await Assert.ThrowsAsync(() => scenario.Jobs.SaveAsync( + Job("handler-conflict", "tenant-b", AlterationJobStatus.Completed, "hidden"))); + } + + Assert.Null(handler.Exception); + } + [Fact] public async Task SaveAsync_ConcurrentSameTenantPlanId_BothWritersSucceedAndOnePayloadWins() { @@ -425,11 +445,13 @@ public sealed class EFCoreAlterationStoreTenantOwnershipTests [Fact] public async Task SaveManyAsync_ConcurrentNamedTenantsOnEmptyJobId_OneOwnerKeepsPayload() { - var gate = new GateFirstAlterationTransactions(); + var gate = new GateFirstAlterationUpdates("AlterationJobs", gateBeforeExecution: true); + var transactionInterceptor = new DeferredSqliteTransactionInterceptor(); await using var pair = await AlterationStoreScenario.CreateSqlitePairAsync( "tenant-a", "tenant-b", - transactionInterceptor: gate); + gate, + transactionInterceptor); gate.Arm(); var results = await Task.WhenAll( @@ -438,7 +460,8 @@ public sealed class EFCoreAlterationStoreTenantOwnershipTests Assert.Equal(1, results.Count(ex => ex is null)); Assert.Equal(1, results.Count(ex => ex is InvalidOperationException)); - Assert.Equal(2, gate.MatchedTransactionCount); + Assert.True(gate.BothReached); + Assert.Equal(3, gate.MatchedCommandCount); var winnerIsA = results[0] is null; using (pair.First.UseTenant(winnerIsA ? "tenant-a" : "tenant-b")) @@ -555,27 +578,44 @@ public sealed class EFCoreAlterationStoreTenantOwnershipTests } /// -/// Releases the first two relevant alteration UPDATE commands after they complete, so competing -/// Save calls reach their INSERT/retry paths together. The gate is armed explicitly after any -/// setup writes so only the concurrent operation is coordinated. +/// Coordinates the first two relevant alteration UPDATE commands so competing writers +/// reach their INSERT/retry paths together. By default the gate releases them after +/// execution; the batch race gates before execution so an explicit transaction does not +/// hold a SQLite writer lock while waiting. The gate is armed explicitly after setup writes. /// -public sealed class GateFirstAlterationUpdates(string tableName) : DbCommandInterceptor +public sealed class GateFirstAlterationUpdates( + string tableName, + bool gateBeforeExecution = false) : DbCommandInterceptor { private readonly TaskCompletionSource _bothReached = new(TaskCreationOptions.RunContinuationsAsynchronously); private int _armed; private int _matchedCommandCount; public int MatchedCommandCount => Volatile.Read(ref _matchedCommandCount); + public bool BothReached => _bothReached.Task.IsCompletedSuccessfully; public void Arm() => Volatile.Write(ref _armed, 1); + public override async ValueTask> NonQueryExecutingAsync( + DbCommand command, + CommandEventData eventData, + InterceptionResult result, + CancellationToken cancellationToken = default) + { + if (!gateBeforeExecution || !IsArmedAlterationUpdate(command)) + return result; + + await CoordinateAsync(cancellationToken); + return result; + } + public override async ValueTask NonQueryExecutedAsync( DbCommand command, CommandExecutedEventData eventData, int result, CancellationToken cancellationToken = default) { - if (!IsArmedAlterationUpdate(command)) + if (gateBeforeExecution || !IsArmedAlterationUpdate(command)) return result; await CoordinateAsync(cancellationToken); @@ -587,6 +627,9 @@ public sealed class GateFirstAlterationUpdates(string tableName) : DbCommandInte private async Task CoordinateAsync(CancellationToken cancellationToken) { + if (Volatile.Read(ref _armed) == 0) + return; + var commandNumber = Interlocked.Increment(ref _matchedCommandCount); if (commandNumber <= 2) { @@ -598,35 +641,26 @@ public sealed class GateFirstAlterationUpdates(string tableName) : DbCommandInte } } -public sealed class GateFirstAlterationTransactions : DbTransactionInterceptor +/// +/// Uses a deferred SQLite transaction for the gated batch race. The normal SQLite +/// transaction starts with BEGIN IMMEDIATE, which reserves the writer lock before +/// a command interceptor can coordinate both writers. +/// +public sealed class DeferredSqliteTransactionInterceptor : DbTransactionInterceptor { - private readonly TaskCompletionSource _bothReached = new(TaskCreationOptions.RunContinuationsAsynchronously); - private int _armed; - private int _matchedTransactionCount; - - public int MatchedTransactionCount => Volatile.Read(ref _matchedTransactionCount); - - public void Arm() => Volatile.Write(ref _armed, 1); - - public override async ValueTask> TransactionStartingAsync( + public override ValueTask> TransactionStartingAsync( DbConnection connection, TransactionStartingEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) { - if (Volatile.Read(ref _armed) != 0) + if (connection is SqliteConnection sqliteConnection) { - var transactionNumber = Interlocked.Increment(ref _matchedTransactionCount); - if (transactionNumber <= 2) - { - if (transactionNumber == 2) - _bothReached.TrySetResult(true); - - await _bothReached.Task.WaitAsync(TimeSpan.FromSeconds(15), cancellationToken); - } + var transaction = sqliteConnection.BeginTransaction(deferred: true); + return ValueTask.FromResult(InterceptionResult.SuppressWithResult(transaction)); } - return result; + return ValueTask.FromResult(result); } } diff --git a/test/unit/Elsa.Alterations.Core.UnitTests/Stores/MemoryAlterationJobStoreTenantIsolationTests.cs b/test/unit/Elsa.Alterations.Core.UnitTests/Stores/MemoryAlterationJobStoreTenantIsolationTests.cs index 6bbde4814..2d2919852 100644 --- a/test/unit/Elsa.Alterations.Core.UnitTests/Stores/MemoryAlterationJobStoreTenantIsolationTests.cs +++ b/test/unit/Elsa.Alterations.Core.UnitTests/Stores/MemoryAlterationJobStoreTenantIsolationTests.cs @@ -142,6 +142,21 @@ public class MemoryAlterationJobStoreTenantIsolationTests Assert.Equal(Tenant.AgnosticTenantId, remaining.TenantId); } + [Fact(DisplayName = "SaveAsync refuses a named source when an agnostic writer updates a * row")] + public async Task SaveAsync_WhenAgnosticAmbientReceivesNamedSource_ThrowsAndLeavesExisting() + { + var backing = new MemoryStore(); + var agnostic = new MemoryAlterationJobStore(backing, new TestTenantAccessor(Tenant.AgnosticTenantId)); + await agnostic.SaveAsync(Job("shared", Tenant.AgnosticTenantId)); + + var ex = await Assert.ThrowsAsync(() => agnostic.SaveAsync(Job("shared", "tenant-b"))); + var remaining = await agnostic.FindAsync(new AlterationJobFilter { Id = "shared" }); + + Assert.Contains("shared", ex.Message); + Assert.NotNull(remaining); + Assert.Equal(Tenant.AgnosticTenantId, remaining.TenantId); + } + [Fact(DisplayName = "SaveManyAsync refuses to overwrite a tenant-agnostic row by Id")] public async Task SaveManyAsync_WhenAgnosticRowExists_NamedTenantThrowsAndLeavesExisting() { @@ -157,6 +172,20 @@ public class MemoryAlterationJobStoreTenantIsolationTests Assert.Equal(Tenant.AgnosticTenantId, remaining.TenantId); } + [Fact(DisplayName = "SaveManyAsync refuses a named source when an agnostic writer updates a * row")] + public async Task SaveManyAsync_WhenAgnosticAmbientReceivesNamedSource_ThrowsAndLeavesExisting() + { + var backing = new MemoryStore(); + var agnostic = new MemoryAlterationJobStore(backing, new TestTenantAccessor(Tenant.AgnosticTenantId)); + await agnostic.SaveAsync(Job("shared", Tenant.AgnosticTenantId)); + + await Assert.ThrowsAsync(() => agnostic.SaveManyAsync([Job("shared", "tenant-b")])); + var remaining = await agnostic.FindAsync(new AlterationJobFilter { Id = "shared" }); + + Assert.NotNull(remaining); + Assert.Equal(Tenant.AgnosticTenantId, remaining.TenantId); + } + [Fact(DisplayName = "SaveAsync still lets an agnostic writer update a * row")] public async Task SaveAsync_WhenAmbientIsAgnostic_UpsertsAgnosticRow() { @@ -189,6 +218,79 @@ public class MemoryAlterationJobStoreTenantIsolationTests Assert.Equal("tenant-a", found.TenantId); } + [Fact(DisplayName = "SaveAsync preserves the stored TenantId on an accepted update")] + public async Task SaveAsync_WhenIncomingTenantDiffers_PreservesExistingTenantId() + { + var store = CreateStore("tenant-a"); + await store.SaveAsync(Job("job-a", "tenant-a")); + var updated = Job("job-a", "tenant-b"); + updated.Status = AlterationJobStatus.Completed; + + await store.SaveAsync(updated); + + var found = await store.FindAsync(new AlterationJobFilter { Id = "job-a" }); + Assert.NotNull(found); + Assert.Equal(AlterationJobStatus.Completed, found.Status); + Assert.Equal("tenant-a", found.TenantId); + } + + [Fact(DisplayName = "SaveManyAsync preserves the stored TenantId for repeated accepted updates")] + public async Task SaveManyAsync_WhenRepeatedIdIncomingTenantsDiffer_PreservesExistingTenantId() + { + var store = CreateStore("tenant-a"); + await store.SaveAsync(Job("job-a", "tenant-a")); + var first = Job("job-a", "tenant-b"); + var second = Job("job-a", "tenant-c"); + second.Status = AlterationJobStatus.Completed; + + await store.SaveManyAsync([first, second]); + + var found = await store.FindAsync(new AlterationJobFilter { Id = "job-a" }); + Assert.NotNull(found); + Assert.Equal(AlterationJobStatus.Completed, found.Status); + Assert.Equal("tenant-a", found.TenantId); + } + + [Fact(DisplayName = "SaveManyAsync preserves the owner for repeated updates of a new Id")] + public async Task SaveManyAsync_WhenRepeatedNewIdUsesSameOwner_SucceedsAndPreservesTenantId() + { + var store = CreateStore("tenant-a"); + var first = Job("job-new", "tenant-a"); + var second = Job("job-new", "tenant-a"); + second.Status = AlterationJobStatus.Completed; + + await store.SaveManyAsync([first, second]); + + var found = await store.FindAsync(new AlterationJobFilter { Id = "job-new" }); + Assert.NotNull(found); + Assert.Equal(AlterationJobStatus.Completed, found.Status); + Assert.Equal("tenant-a", found.TenantId); + } + + [Fact(DisplayName = "SaveManyAsync rejects a repeated new Id that changes * to a named source")] + public async Task SaveManyAsync_WhenRepeatedNewIdChangesAgnosticToNamed_ThrowsAndPersistsNothing() + { + var store = CreateStore(Tenant.AgnosticTenantId); + var first = Job("job-new", Tenant.AgnosticTenantId); + var second = Job("job-new", "tenant-b"); + + await Assert.ThrowsAsync(() => store.SaveManyAsync([first, second])); + + Assert.Null(await store.FindAsync(new AlterationJobFilter { Id = "job-new" })); + } + + [Fact(DisplayName = "SaveManyAsync rejects a repeated new Id that changes its named owner")] + public async Task SaveManyAsync_WhenRepeatedNewIdChangesNamedOwner_ThrowsAndPersistsNothing() + { + var store = CreateStore("tenant-a"); + var first = Job("job-new", "tenant-b"); + var second = Job("job-new", "tenant-c"); + + await Assert.ThrowsAsync(() => store.SaveManyAsync([first, second])); + + Assert.Null(await store.FindAsync(new AlterationJobFilter { Id = "job-new" })); + } + [Fact(DisplayName = "SaveAsync stamps the ambient tenant when TenantId is unset")] public async Task SaveAsync_WhenTenantIdUnset_StampsAmbientTenant() { diff --git a/test/unit/Elsa.Alterations.Core.UnitTests/Stores/MemoryAlterationPlanStoreTenantIsolationTests.cs b/test/unit/Elsa.Alterations.Core.UnitTests/Stores/MemoryAlterationPlanStoreTenantIsolationTests.cs index 8b275286d..a8cb35d78 100644 --- a/test/unit/Elsa.Alterations.Core.UnitTests/Stores/MemoryAlterationPlanStoreTenantIsolationTests.cs +++ b/test/unit/Elsa.Alterations.Core.UnitTests/Stores/MemoryAlterationPlanStoreTenantIsolationTests.cs @@ -101,6 +101,21 @@ public class MemoryAlterationPlanStoreTenantIsolationTests Assert.Equal(Tenant.AgnosticTenantId, remaining.TenantId); } + [Fact(DisplayName = "SaveAsync refuses a named source when an agnostic writer updates a * row")] + public async Task SaveAsync_WhenAgnosticAmbientReceivesNamedSource_ThrowsAndLeavesExisting() + { + var backing = new MemoryStore(); + var agnostic = new MemoryAlterationPlanStore(backing, new TestTenantAccessor(Tenant.AgnosticTenantId)); + await agnostic.SaveAsync(Plan("shared", Tenant.AgnosticTenantId)); + + var ex = await Assert.ThrowsAsync(() => agnostic.SaveAsync(Plan("shared", "tenant-b"))); + var remaining = await agnostic.FindAsync(new AlterationPlanFilter { Id = "shared" }); + + Assert.Contains("shared", ex.Message); + Assert.NotNull(remaining); + Assert.Equal(Tenant.AgnosticTenantId, remaining.TenantId); + } + [Fact(DisplayName = "SaveAsync still lets an agnostic writer update a * row")] public async Task SaveAsync_WhenAmbientIsAgnostic_UpsertsAgnosticRow() { @@ -133,6 +148,22 @@ public class MemoryAlterationPlanStoreTenantIsolationTests Assert.Equal("tenant-a", found.TenantId); } + [Fact(DisplayName = "SaveAsync preserves the stored TenantId on an accepted update")] + public async Task SaveAsync_WhenIncomingTenantDiffers_PreservesExistingTenantId() + { + var store = CreateStore("tenant-a"); + await store.SaveAsync(Plan("plan-a", "tenant-a")); + var updated = Plan("plan-a", "tenant-b"); + updated.Status = AlterationPlanStatus.Completed; + + await store.SaveAsync(updated); + + var found = await store.FindAsync(new AlterationPlanFilter { Id = "plan-a" }); + Assert.NotNull(found); + Assert.Equal(AlterationPlanStatus.Completed, found.Status); + Assert.Equal("tenant-a", found.TenantId); + } + [Fact(DisplayName = "SaveAsync stamps the ambient tenant when TenantId is unset")] public async Task SaveAsync_WhenTenantIdUnset_StampsAmbientTenant() { diff --git a/test/unit/Elsa.Persistence.EFCore.UnitTests/DbExceptionClassifierTests.cs b/test/unit/Elsa.Persistence.EFCore.UnitTests/DbExceptionClassifierTests.cs index 8f7d6e125..c546778b7 100644 --- a/test/unit/Elsa.Persistence.EFCore.UnitTests/DbExceptionClassifierTests.cs +++ b/test/unit/Elsa.Persistence.EFCore.UnitTests/DbExceptionClassifierTests.cs @@ -1,3 +1,5 @@ +using Microsoft.EntityFrameworkCore; + namespace Elsa.Persistence.EFCore.UnitTests; public sealed class DbExceptionClassifierTests @@ -24,6 +26,33 @@ public sealed class DbExceptionClassifierTests Assert.Equal(expected, DbExceptionClassifier.IsDuplicateKey(exception)); } + [Theory] + [InlineData(1205, true)] + [InlineData(1213, true)] + [InlineData(1062, false)] + public void IsTransient_WhenMySqlProviderUsesInnoDbLockErrorCodes(int number, bool expected) + { + var exception = new MySqlConnector.MySqlException(number); + + Assert.Equal(expected, DbExceptionClassifier.IsTransient("Pomelo.EntityFrameworkCore.MySql", exception)); + } + + [Fact] + public void IsTransient_WhenMySqlProviderExceptionOnlyHasAnUnrelatedNumberProperty_ReturnsFalse() + { + var exception = new UnrelatedNumberException(1213); + + Assert.False(DbExceptionClassifier.IsTransient("Pomelo.EntityFrameworkCore.MySql", exception)); + } + + [Fact] + public void IsTransient_WhenMySqlDeadlockIsWrappedByDbUpdateException_ReturnsTrue() + { + var exception = new DbUpdateException("Write failed", new MySqlConnector.MySqlException(1213)); + + Assert.True(DbExceptionClassifier.IsTransient("Pomelo.EntityFrameworkCore.MySql", exception)); + } + private sealed class SqliteExceptionWithExtendedCode(int sqliteErrorCode, int sqliteExtendedErrorCode) : Exception { public int SqliteErrorCode { get; } = sqliteErrorCode; @@ -34,4 +63,9 @@ public sealed class DbExceptionClassifierTests { public int SqliteErrorCode { get; } = sqliteErrorCode; } + + private sealed class UnrelatedNumberException(int number) : Exception + { + public int Number { get; } = number; + } } diff --git a/test/unit/Elsa.Persistence.EFCore.UnitTests/FakeMySqlException.cs b/test/unit/Elsa.Persistence.EFCore.UnitTests/FakeMySqlException.cs new file mode 100644 index 000000000..4025aff19 --- /dev/null +++ b/test/unit/Elsa.Persistence.EFCore.UnitTests/FakeMySqlException.cs @@ -0,0 +1,6 @@ +namespace MySqlConnector; + +internal sealed class MySqlException(int number) : Exception +{ + public int Number { get; } = number; +} diff --git a/test/unit/Elsa.Persistence.EFCore.UnitTests/StoreWriteRetryTests.cs b/test/unit/Elsa.Persistence.EFCore.UnitTests/StoreWriteRetryTests.cs new file mode 100644 index 000000000..4f39c41a8 --- /dev/null +++ b/test/unit/Elsa.Persistence.EFCore.UnitTests/StoreWriteRetryTests.cs @@ -0,0 +1,103 @@ +using Elsa.Persistence.EFCore; +using Microsoft.Data.Sqlite; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Storage; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Persistence.EFCore.UnitTests; + +public sealed class StoreWriteRetryTests +{ + [Theory] + [InlineData(1205)] + [InlineData(1213)] + public async Task ExecuteWriteWithRetryAsync_WhenMySqlLockFailureOccurs_RetriesWholeOperationWithFreshContextAndTransaction(int errorNumber) + { + await using var connection = new SqliteConnection("Data Source=:memory:"); + await connection.OpenAsync(); + var factory = new TrackingDbContextFactory(CreateOptions(connection)); + var store = new Store(factory, new ServiceCollection().BuildServiceProvider()); + var attempts = 0; + var transactionCount = 0; + + await store.ExecuteWriteWithRetryAsync( + async (dbContext, cancellationToken) => + { + var attempt = Interlocked.Increment(ref attempts); + await using var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken); + Interlocked.Increment(ref transactionCount); + + if (attempt == 1) + throw new MySqlConnector.MySqlException(errorNumber); + + await transaction.CommitAsync(cancellationToken); + }, + CancellationToken.None); + + Assert.Equal(2, attempts); + Assert.Equal(2, transactionCount); + Assert.Equal(2, factory.Contexts.Count); + Assert.NotSame(factory.Contexts[0], factory.Contexts[1]); + Assert.NotEqual(factory.Contexts[0].ContextId.InstanceId, factory.Contexts[1].ContextId.InstanceId); + } + + [Fact] + public async Task ExecuteWriteWithRetryAsync_WhenMySqlDuplicateKeyOccurs_DoesNotRetry() + { + await using var connection = new SqliteConnection("Data Source=:memory:"); + await connection.OpenAsync(); + var factory = new TrackingDbContextFactory(CreateOptions(connection)); + var store = new Store(factory, new ServiceCollection().BuildServiceProvider()); + var attempts = 0; + + await Assert.ThrowsAsync(() => store.ExecuteWriteWithRetryAsync( + (_, _) => + { + Interlocked.Increment(ref attempts); + throw new MySqlConnector.MySqlException(1062); + }, + CancellationToken.None)); + + Assert.Equal(1, attempts); + Assert.Single(factory.Contexts); + } + + private static DbContextOptions CreateOptions(SqliteConnection connection) => + new DbContextOptionsBuilder() + .UseSqlite(connection) + .ReplaceService() + .Options; + + private sealed class RetryDbContext(DbContextOptions options) : DbContext(options); + + private sealed class RetryEntity + { + public string Id { get; set; } = string.Empty; + } + + private sealed class TrackingDbContextFactory(DbContextOptions options) : IDbContextFactory + { + public List Contexts { get; } = []; + + public RetryDbContext CreateDbContext() + { + var context = new RetryDbContext(options); + Contexts.Add(context); + return context; + } + + public Task CreateDbContextAsync(CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + return Task.FromResult(CreateDbContext()); + } + } + + private sealed class FakeMySqlDatabaseProvider : IDatabaseProvider + { + public string Name => "Pomelo.EntityFrameworkCore.MySql"; + + public bool IsConfigured(IDbContextOptions options) => true; + } +}