SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Fix Alterations ownership parity and MySQL retry (#8133)

Browse source 
Closes #8125
This commit is contained in:
Sipke Schoorstra 2026-09-14 02:51:34 +02:00 committed by GitHub
parent 54209e24c2
commit e4478af2be
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 425 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationJobStore.cs
View file

@ -52,8 +52,9 @@ public class MemoryAlterationJobStore : IAlterationJobStore
foreach (var job in list)
ApplyCurrentTenant(job);
var tenantIdsById = new Dictionary<string, string?>(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<string, string?>? 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);
}
/// <summary>
/// <c>*</c> is visible to every tenant, but only an agnostic writer may replace it.
/// Named tenants may upsert their own visible rows.
/// </summary>
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)
{

11
src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationPlanStore.cs
View file

@ -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;
}
/// <summary>
/// <c>*</c> is visible to every tenant, but only an agnostic writer may replace it.
/// Named tenants may upsert their own visible rows.
/// </summary>
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)
{

30
src/modules/Elsa.Persistence.EFCore.Common/DbExceptionClassifier.cs
View file

@ -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<int> 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();

27
src/modules/Elsa.Persistence.EFCore.Common/Store.cs
View file

@ -19,8 +19,8 @@ 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 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<TDbContext, TEntity>(IDbContextFactory<TDbContext> 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<TDbContext, TEntity>(IDbContextFactory<TDbContext> dbContextF
var tenantId = serviceProvider.GetRequiredService<ITenantAccessor>().TenantId;
await ExecuteSqlServerWriteWithRetryAsync(async (dbContext, ct) =>
await ExecuteWriteWithRetryAsync(async (dbContext, ct) =>
{
if (onSaving != null)
{
@ -263,7 +263,12 @@ public class Store<TDbContext, TEntity>(IDbContextFactory<TDbContext> dbContextF
}
}
internal async Task ExecuteSqlServerWriteWithRetryAsync(
/// <summary>
/// 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.
/// </summary>
internal async Task ExecuteWriteWithRetryAsync(
Func<TDbContext, CancellationToken, Task> operation,
CancellationToken cancellationToken)
{
@ -280,9 +285,9 @@ public class Store<TDbContext, TEntity>(IDbContextFactory<TDbContext> 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<TDbContext, TEntity>(IDbContextFactory<TDbContext> 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));
/// <summary>
/// Updates the entity.

13
src/modules/Elsa.Persistence.EFCore/Modules/Alterations/AlterationJobStore.cs
View file

@ -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);
}
/// <inheritdoc />
@ -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);

90
test/integration/Elsa.Alterations.Persistence.ConformanceTests/EFCoreAlterationStoreTenantOwnershipTests.cs
View file

@ -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<InvalidOperationException>(() => 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
}
/// <summary>
/// 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.
/// </summary>
public sealed class GateFirstAlterationUpdates(string tableName) : DbCommandInterceptor
public sealed class GateFirstAlterationUpdates(
string tableName,
bool gateBeforeExecution = false) : DbCommandInterceptor
{
private readonly TaskCompletionSource<bool> _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<InterceptionResult<int>> NonQueryExecutingAsync(
DbCommand command,
CommandEventData eventData,
InterceptionResult<int> result,
CancellationToken cancellationToken = default)
{
if (!gateBeforeExecution || !IsArmedAlterationUpdate(command))
return result;
await CoordinateAsync(cancellationToken);
return result;
}
public override async ValueTask<int> 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
/// <summary>
/// Uses a deferred SQLite transaction for the gated batch race. The normal SQLite
/// transaction starts with <c>BEGIN IMMEDIATE</c>, which reserves the writer lock before
/// a command interceptor can coordinate both writers.
/// </summary>
public sealed class DeferredSqliteTransactionInterceptor : DbTransactionInterceptor
{
private readonly TaskCompletionSource<bool> _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<InterceptionResult<DbTransaction>> TransactionStartingAsync(
public override ValueTask<InterceptionResult<DbTransaction>> TransactionStartingAsync(
DbConnection connection,
TransactionStartingEventData eventData,
InterceptionResult<DbTransaction> 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<DbTransaction>.SuppressWithResult(transaction));
}
return result;
return ValueTask.FromResult(result);
}
}

102
test/unit/Elsa.Alterations.Core.UnitTests/Stores/MemoryAlterationJobStoreTenantIsolationTests.cs
View file

@ -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<AlterationJob>();
var agnostic = new MemoryAlterationJobStore(backing, new TestTenantAccessor(Tenant.AgnosticTenantId));
await agnostic.SaveAsync(Job("shared", Tenant.AgnosticTenantId));
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => 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<AlterationJob>();
var agnostic = new MemoryAlterationJobStore(backing, new TestTenantAccessor(Tenant.AgnosticTenantId));
await agnostic.SaveAsync(Job("shared", Tenant.AgnosticTenantId));
await Assert.ThrowsAsync<InvalidOperationException>(() => 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<InvalidOperationException>(() => 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<InvalidOperationException>(() => 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()
{

31
test/unit/Elsa.Alterations.Core.UnitTests/Stores/MemoryAlterationPlanStoreTenantIsolationTests.cs
View file

@ -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<AlterationPlan>();
var agnostic = new MemoryAlterationPlanStore(backing, new TestTenantAccessor(Tenant.AgnosticTenantId));
await agnostic.SaveAsync(Plan("shared", Tenant.AgnosticTenantId));
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => 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()
{

34
test/unit/Elsa.Persistence.EFCore.UnitTests/DbExceptionClassifierTests.cs
View file

@ -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;
}
}

6
test/unit/Elsa.Persistence.EFCore.UnitTests/FakeMySqlException.cs Normal file
View file

@ -0,0 +1,6 @@
namespace MySqlConnector;
internal sealed class MySqlException(int number) : Exception
{
public int Number { get; } = number;
}

103
test/unit/Elsa.Persistence.EFCore.UnitTests/StoreWriteRetryTests.cs Normal file
View file

@ -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<RetryDbContext, RetryEntity>(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<RetryDbContext, RetryEntity>(factory, new ServiceCollection().BuildServiceProvider());
var attempts = 0;
await Assert.ThrowsAsync<MySqlConnector.MySqlException>(() => store.ExecuteWriteWithRetryAsync(
(_, _) =>
{
Interlocked.Increment(ref attempts);
throw new MySqlConnector.MySqlException(1062);
},
CancellationToken.None));
Assert.Equal(1, attempts);
Assert.Single(factory.Contexts);
}
private static DbContextOptions<RetryDbContext> CreateOptions(SqliteConnection connection) =>
new DbContextOptionsBuilder<RetryDbContext>()
.UseSqlite(connection)
.ReplaceService<IDatabaseProvider, FakeMySqlDatabaseProvider>()
.Options;
private sealed class RetryDbContext(DbContextOptions<RetryDbContext> options) : DbContext(options);
private sealed class RetryEntity
{
public string Id { get; set; } = string.Empty;
}
private sealed class TrackingDbContextFactory(DbContextOptions<RetryDbContext> options) : IDbContextFactory<RetryDbContext>
{
public List<RetryDbContext> Contexts { get; } = [];
public RetryDbContext CreateDbContext()
{
var context = new RetryDbContext(options);
Contexts.Add(context);
return context;
}
public Task<RetryDbContext> 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;
}
}