From d2d3109024535eb264db84891c84d34de368e5e1 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Sun, 13 Sep 2026 02:20:28 -0700 Subject: [PATCH] fix(runtime): enforce MemoryTriggerStore logical uniqueness (#8098) * fix(runtime): enforce MemoryTriggerStore logical uniqueness MemoryTriggerStore upserted only by Id, so two records with different Ids but the same (WorkflowDefinitionId, Hash, ActivityId, TenantId) were accepted in memory and rejected under EF. Mirror EFCoreTriggerStore: distinct-by-logical-key, skip already-present keys on ReplaceAsync, reject Save* collisions, and stamp the current tenant when unset. Co-authored-by: Sipke Schoorstra * fix(runtime): keep FindAsync first-match and use a structural trigger key FindAsync must return the first matching trigger. SingleOrDefault threw when a valid filter (for example WorkflowDefinitionId) matched several distinct logical keys. Restore FirstOrDefault to match ITriggerStore and EF. Represent the logical key as a record so fields that contain U+001F cannot collide. Co-authored-by: Sipke Schoorstra --------- Co-authored-by: Cursor Agent --- .../Stores/MemoryTriggerStore.cs | 119 ++++++++- .../Stores/MemoryTriggerStoreTests.cs | 228 ++++++++++++++++++ 2 files changed, 338 insertions(+), 9 deletions(-) create mode 100644 test/unit/Elsa.Workflows.Runtime.UnitTests/Stores/MemoryTriggerStoreTests.cs diff --git a/src/modules/Elsa.Workflows.Runtime/Stores/MemoryTriggerStore.cs b/src/modules/Elsa.Workflows.Runtime/Stores/MemoryTriggerStore.cs index 88e454146..55aee2ad7 100644 --- a/src/modules/Elsa.Workflows.Runtime/Stores/MemoryTriggerStore.cs +++ b/src/modules/Elsa.Workflows.Runtime/Stores/MemoryTriggerStore.cs @@ -1,5 +1,6 @@ using Elsa.Common.Entities; using Elsa.Common.Models; +using Elsa.Common.Multitenancy; using Elsa.Common.Services; using Elsa.Extensions; using Elsa.Workflows.Runtime.Entities; @@ -14,26 +15,48 @@ namespace Elsa.Workflows.Runtime.Stores; public class MemoryTriggerStore : ITriggerStore { private readonly MemoryStore _store; + private readonly ITenantAccessor? _tenantAccessor; /// /// Initializes a new instance of the class. /// - public MemoryTriggerStore(MemoryStore store) + public MemoryTriggerStore(MemoryStore store, ITenantAccessor? tenantAccessor = null) { _store = store; + _tenantAccessor = tenantAccessor; } /// public ValueTask SaveAsync(StoredTrigger record, CancellationToken cancellationToken = default) { - _store.Save(record, x => x.Id); + lock (_store.Sync) + { + ApplyCurrentTenant(record); + EnsureLogicalKeyAvailable(record); + _store.Save(record, x => x.Id); + } + return new(); } /// public ValueTask SaveManyAsync(IEnumerable records, CancellationToken cancellationToken = default) { - _store.SaveMany(records, x => x.Id); + lock (_store.Sync) + { + var recordList = records.ToList(); + + foreach (var record in recordList) + ApplyCurrentTenant(record); + + var uniqueRecords = DistinctByLogicalKey(recordList).ToList(); + + foreach (var record in uniqueRecords) + EnsureLogicalKeyAvailable(record); + + _store.SaveMany(uniqueRecords, x => x.Id); + } + return new(); } @@ -66,15 +89,93 @@ public class MemoryTriggerStore : ITriggerStore /// public ValueTask ReplaceAsync(IEnumerable removed, IEnumerable added, CancellationToken cancellationToken = default) { - _store.DeleteMany(removed, x => x.Id); - _store.SaveMany(added, x => x.Id); + lock (_store.Sync) + { + var removedList = removed.ToList(); + var addedList = added.ToList(); + + foreach (var trigger in addedList) + ApplyCurrentTenant(trigger); + + addedList = DistinctByLogicalKey(addedList).ToList(); + + if (removedList.Count > 0) + _store.DeleteMany(removedList, x => x.Id); + + if (addedList.Count == 0) + return new(); + + var newTriggers = GetMissingLogicalTriggers(addedList); + + if (newTriggers.Count == 0) + return new(); + + _store.SaveMany(newTriggers, x => x.Id); + } + return new(); } /// - public async ValueTask DeleteManyAsync(TriggerFilter filter, CancellationToken cancellationToken = default) + public ValueTask DeleteManyAsync(TriggerFilter filter, CancellationToken cancellationToken = default) { - var ids = (await FindManyAsync(filter, cancellationToken)).Select(x => x.Id); - return _store.DeleteMany(ids); + lock (_store.Sync) + { + var ids = _store.Query(filter.Apply).Select(x => x.Id).ToList(); + return new(_store.DeleteMany(ids)); + } } -} \ No newline at end of file + + private void EnsureLogicalKeyAvailable(StoredTrigger record) + { + var logicalKey = GetLogicalKey(record); + var existing = _store.Find(x => x.Id != record.Id && GetLogicalKey(x) == logicalKey); + + if (existing is not null) + { + throw new InvalidOperationException( + $"A stored trigger already exists for workflow '{record.WorkflowDefinitionId}', hash '{record.Hash}', activity '{record.ActivityId}', tenant '{record.TenantId}'."); + } + } + + private List GetMissingLogicalTriggers(ICollection triggers) + { + var existingKeys = GetExistingLogicalKeys(triggers); + return triggers + .Where(trigger => !existingKeys.Contains(GetLogicalKey(trigger))) + .ToList(); + } + + private HashSet GetExistingLogicalKeys(ICollection triggers) + { + var workflowDefinitionIds = triggers.Select(x => x.WorkflowDefinitionId).Distinct().ToHashSet(); + return _store + .FindMany(trigger => workflowDefinitionIds.Contains(trigger.WorkflowDefinitionId)) + .Select(GetLogicalKey) + .ToHashSet(); + } + + private void ApplyCurrentTenant(StoredTrigger trigger) + { + if (trigger.TenantId == Tenant.AgnosticTenantId || _tenantAccessor is null) + return; + + trigger.TenantId ??= _tenantAccessor.TenantId; + } + + private static IEnumerable DistinctByLogicalKey(IEnumerable triggers) + { + var seen = new HashSet(); + + foreach (var trigger in triggers) + { + if (seen.Add(GetLogicalKey(trigger))) + yield return trigger; + } + } + + private static TriggerLogicalKey GetLogicalKey(StoredTrigger trigger) => + new(trigger.WorkflowDefinitionId, trigger.Hash, trigger.ActivityId, trigger.TenantId); + + private readonly record struct TriggerLogicalKey(string WorkflowDefinitionId, string? Hash, string ActivityId, string? TenantId); +} diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Stores/MemoryTriggerStoreTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Stores/MemoryTriggerStoreTests.cs new file mode 100644 index 000000000..624c2f4bc --- /dev/null +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Stores/MemoryTriggerStoreTests.cs @@ -0,0 +1,228 @@ +using Elsa.Common.Multitenancy; +using Elsa.Common.Services; +using Elsa.Testing.Shared.Multitenancy; +using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Stores; + +namespace Elsa.Workflows.Runtime.UnitTests.Stores; + +/// +/// Memory must enforce the same logical uniqueness that EF Core +/// already enforces via the unique index on (WorkflowDefinitionId, Hash, ActivityId, TenantId) +/// and that EFCoreTriggerStore.ReplaceAsync mirrors. +/// +public class MemoryTriggerStoreTests +{ + [Fact(DisplayName = "ReplaceAsync keeps the first added record when the incoming batch repeats a logical key")] + public async Task ReplaceAsync_WhenAddedBatchRepeatsLogicalKey_StoresOnlyTheFirst() + { + var store = CreateStore(); + var first = Trigger("id-1"); + var duplicate = Trigger("id-2"); + + await store.ReplaceAsync([], [first, duplicate]); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + var match = Assert.Single(stored); + Assert.Equal("id-1", match.Id); + } + + [Fact(DisplayName = "ReplaceAsync skips an added record whose logical key is already present")] + public async Task ReplaceAsync_WhenLogicalKeyAlreadyPresent_SkipsTheAddedRecord() + { + var store = CreateStore(); + var existing = Trigger("existing-id", hash: "hash-1"); + await store.SaveAsync(existing); + + await store.ReplaceAsync([], [Trigger("new-id", hash: "hash-1")]); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + var match = Assert.Single(stored); + Assert.Equal("existing-id", match.Id); + Assert.Equal("v1", match.WorkflowDefinitionVersionId); + } + + [Fact(DisplayName = "ReplaceAsync can insert a new Id after the existing logical-key row is removed")] + public async Task ReplaceAsync_WhenExistingLogicalKeyIsRemoved_InsertsTheReplacement() + { + var store = CreateStore(); + var existing = Trigger("existing-id", hash: "hash-1"); + await store.SaveAsync(existing); + + var replacement = Trigger("new-id", hash: "hash-1", workflowDefinitionVersionId: "v2"); + await store.ReplaceAsync([existing], [replacement]); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + var match = Assert.Single(stored); + Assert.Equal("new-id", match.Id); + Assert.Equal("v2", match.WorkflowDefinitionVersionId); + } + + [Fact(DisplayName = "ReplaceAsync stamps the current tenant when TenantId is unset")] + public async Task ReplaceAsync_WhenTenantIdIsUnset_AppliesCurrentTenant() + { + var store = CreateStore(new TestTenantAccessor("tenant-a")); + var trigger = Trigger("id-1"); + trigger.TenantId = null; + + await store.ReplaceAsync([], [trigger]); + + var stored = await store.FindAsync(new TriggerFilter { Id = "id-1" }); + Assert.Equal("tenant-a", stored!.TenantId); + } + + [Fact(DisplayName = "ReplaceAsync leaves a tenant-agnostic TenantId untouched")] + public async Task ReplaceAsync_WhenTenantIdIsAgnostic_DoesNotOverwriteTenant() + { + var store = CreateStore(new TestTenantAccessor("tenant-a")); + var trigger = Trigger("id-1"); + trigger.TenantId = Tenant.AgnosticTenantId; + + await store.ReplaceAsync([], [trigger]); + + var stored = await store.FindAsync(new TriggerFilter { Id = "id-1" }); + Assert.Equal(Tenant.AgnosticTenantId, stored!.TenantId); + } + + [Fact(DisplayName = "ReplaceAsync treats different tenants as distinct logical keys")] + public async Task ReplaceAsync_WhenTenantIdsDiffer_StoresBothRecords() + { + var store = CreateStore(); + var tenantA = Trigger("id-a"); + tenantA.TenantId = "tenant-a"; + var tenantB = Trigger("id-b"); + tenantB.TenantId = "tenant-b"; + + await store.ReplaceAsync([], [tenantA, tenantB]); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + Assert.Equal(2, stored.Count); + Assert.Contains(stored, x => x.Id == "id-a"); + Assert.Contains(stored, x => x.Id == "id-b"); + } + + [Fact(DisplayName = "SaveAsync stamps the current tenant when TenantId is unset")] + public async Task SaveAsync_WhenTenantIdIsUnset_AppliesCurrentTenant() + { + var store = CreateStore(new TestTenantAccessor("tenant-a")); + var trigger = Trigger("id-1"); + trigger.TenantId = null; + + await store.SaveAsync(trigger); + + var stored = await store.FindAsync(new TriggerFilter { Id = "id-1" }); + Assert.Equal("tenant-a", stored!.TenantId); + } + + [Fact(DisplayName = "SaveAsync updates the existing row when the Id matches")] + public async Task SaveAsync_WhenIdMatches_UpdatesTheExistingRow() + { + var store = CreateStore(); + await store.SaveAsync(Trigger("id-1", hash: "hash-1")); + + await store.SaveAsync(Trigger("id-1", hash: "hash-2", workflowDefinitionVersionId: "v2")); + + var stored = await store.FindAsync(new TriggerFilter { Id = "id-1" }); + Assert.Equal("hash-2", stored!.Hash); + Assert.Equal("v2", stored.WorkflowDefinitionVersionId); + } + + [Fact(DisplayName = "SaveAsync rejects a different Id that repeats an existing logical key")] + public async Task SaveAsync_WhenLogicalKeyExistsUnderAnotherId_Throws() + { + var store = CreateStore(); + await store.SaveAsync(Trigger("id-1", hash: "hash-1")); + + var exception = await Assert.ThrowsAsync(() => + store.SaveAsync(Trigger("id-2", hash: "hash-1")).AsTask()); + + Assert.Contains("already exists", exception.Message); + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + Assert.Equal("id-1", Assert.Single(stored).Id); + } + + [Fact(DisplayName = "SaveManyAsync keeps the first record when the batch repeats a logical key")] + public async Task SaveManyAsync_WhenBatchRepeatsLogicalKey_StoresOnlyTheFirst() + { + var store = CreateStore(); + + await store.SaveManyAsync([Trigger("id-1"), Trigger("id-2")]); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + Assert.Equal("id-1", Assert.Single(stored).Id); + } + + [Fact(DisplayName = "SaveManyAsync rejects a batch whose logical key is already present under another Id")] + public async Task SaveManyAsync_WhenLogicalKeyExistsUnderAnotherId_Throws() + { + var store = CreateStore(); + await store.SaveAsync(Trigger("id-1", hash: "hash-1")); + + var exception = await Assert.ThrowsAsync(() => + store.SaveManyAsync([Trigger("id-2", hash: "hash-1")]).AsTask()); + + Assert.Contains("already exists", exception.Message); + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + Assert.Equal("id-1", Assert.Single(stored).Id); + } + + [Fact(DisplayName = "FindAsync returns the first record when a filter matches several distinct triggers")] + public async Task FindAsync_WhenFilterMatchesMultipleTriggers_ReturnsTheFirst() + { + var store = CreateStore(); + await store.SaveAsync(Trigger("id-1", hash: "hash-1")); + await store.SaveAsync(Trigger("id-2", hash: "hash-2")); + + var found = await store.FindAsync(new TriggerFilter { WorkflowDefinitionId = "workflow-1" }); + + Assert.True(found!.Id is "id-1" or "id-2"); + } + + [Fact(DisplayName = "FindAsync returns the matching record when the filter is unique")] + public async Task FindAsync_WhenOneLogicalKeyMatches_ReturnsThatRecord() + { + var store = CreateStore(); + await store.SaveAsync(Trigger("id-1", hash: "hash-1")); + await store.SaveAsync(Trigger("id-2", hash: "hash-2")); + + var found = await store.FindAsync(new TriggerFilter { Hash = "hash-1" }); + + Assert.Equal("id-1", found!.Id); + } + + [Fact(DisplayName = "SaveAsync keeps distinct triggers whose fields contain the old delimiter character")] + public async Task SaveAsync_WhenFieldsContainUnitSeparator_DoesNotCollide() + { + var store = CreateStore(); + var first = Trigger("id-1", hash: "b\u001fc"); + var second = Trigger("id-2", workflowDefinitionId: "workflow-1\u001fb", hash: "c"); + + await store.SaveAsync(first); + await store.SaveAsync(second); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + Assert.Equal(2, stored.Count); + Assert.Contains(stored, x => x.Id == "id-1"); + Assert.Contains(stored, x => x.Id == "id-2"); + } + + private static MemoryTriggerStore CreateStore(ITenantAccessor? tenantAccessor = null) => + new(new MemoryStore(), tenantAccessor ?? TestTenantAccessor.Default); + + private static StoredTrigger Trigger( + string id, + string workflowDefinitionId = "workflow-1", + string workflowDefinitionVersionId = "v1", + string activityId = "activity-1", + string hash = "hash-1") => + new() + { + Id = id, + WorkflowDefinitionId = workflowDefinitionId, + WorkflowDefinitionVersionId = workflowDefinitionVersionId, + ActivityId = activityId, + Hash = hash, + Name = "Elsa.HttpEndpoint" + }; +}