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 <sipkeschoorstra@outlook.com> * 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 <sipkeschoorstra@outlook.com> --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com>
This commit is contained in:
Sipke Schoorstra
GitHub
parent
0ac7184226
commit
d2d3109024
|
|
@ -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<StoredTrigger> _store;
|
||||
private readonly ITenantAccessor? _tenantAccessor;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="MemoryTriggerStore"/> class.
|
||||
/// </summary>
|
||||
public MemoryTriggerStore(MemoryStore<StoredTrigger> store)
|
||||
public MemoryTriggerStore(MemoryStore<StoredTrigger> store, ITenantAccessor? tenantAccessor = null)
|
||||
{
|
||||
_store = store;
|
||||
_tenantAccessor = tenantAccessor;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
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();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask SaveManyAsync(IEnumerable<StoredTrigger> 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
|
|||
/// <inheritdoc />
|
||||
public ValueTask ReplaceAsync(IEnumerable<StoredTrigger> removed, IEnumerable<StoredTrigger> 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();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask<long> DeleteManyAsync(TriggerFilter filter, CancellationToken cancellationToken = default)
|
||||
public ValueTask<long> 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));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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<StoredTrigger> GetMissingLogicalTriggers(ICollection<StoredTrigger> triggers)
|
||||
{
|
||||
var existingKeys = GetExistingLogicalKeys(triggers);
|
||||
return triggers
|
||||
.Where(trigger => !existingKeys.Contains(GetLogicalKey(trigger)))
|
||||
.ToList();
|
||||
}
|
||||
|
||||
private HashSet<TriggerLogicalKey> GetExistingLogicalKeys(ICollection<StoredTrigger> 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<StoredTrigger> DistinctByLogicalKey(IEnumerable<StoredTrigger> triggers)
|
||||
{
|
||||
var seen = new HashSet<TriggerLogicalKey>();
|
||||
|
||||
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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Memory must enforce the same <see cref="StoredTrigger"/> logical uniqueness that EF Core
|
||||
/// already enforces via the unique index on (WorkflowDefinitionId, Hash, ActivityId, TenantId)
|
||||
/// and that <c>EFCoreTriggerStore.ReplaceAsync</c> mirrors.
|
||||
/// </summary>
|
||||
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<InvalidOperationException>(() =>
|
||||
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<InvalidOperationException>(() =>
|
||||
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<StoredTrigger>(), 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"
|
||||
};
|
||||
}
|
||||
Loading…
Reference in a new issue