From 1b7f9851c1de245a7c254918b3bd58c6acf19a3e Mon Sep 17 00:00:00 2001 From: Rostislav Statko Date: Wed, 19 Aug 2026 19:09:57 +0300 Subject: [PATCH 1/3] Fix tenant context propagation for alteration jobs Capture the current tenant when dispatching a background alteration job and restore it while executing the queued callback. Add regression coverage verifying that the dispatch-time tenant is used and the worker's previous tenant context is restored. Fixes #7961 --- .../BackgroundAlterationJobDispatcher.cs | 15 +++-- .../BackgroundAlterationJobDispatcherTests.cs | 67 +++++++++++++++++++ 2 files changed, 77 insertions(+), 5 deletions(-) create mode 100644 test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs diff --git a/src/modules/Elsa.Alterations/Services/BackgroundAlterationJobDispatcher.cs b/src/modules/Elsa.Alterations/Services/BackgroundAlterationJobDispatcher.cs index e9180a566a..0b426dc5a8 100644 --- a/src/modules/Elsa.Alterations/Services/BackgroundAlterationJobDispatcher.cs +++ b/src/modules/Elsa.Alterations/Services/BackgroundAlterationJobDispatcher.cs @@ -1,4 +1,5 @@ using Elsa.Alterations.Core.Contracts; +using Elsa.Common.Multitenancy; using Elsa.Mediator.Contracts; using Microsoft.Extensions.DependencyInjection; @@ -7,19 +8,23 @@ namespace Elsa.Alterations.Services; /// /// Dispatches an alteration job for execution using an in-memory channel. /// -public class BackgroundAlterationJobDispatcher(IJobQueue jobQueue, IServiceScopeFactory scopeFactory) : IAlterationJobDispatcher +public class BackgroundAlterationJobDispatcher( + IJobQueue jobQueue, + ITenantAccessor tenantAccessor, + ITenantScopeFactory tenantScopeFactory) : IAlterationJobDispatcher { /// public ValueTask DispatchAsync(string jobId, CancellationToken cancellationToken = default) { - jobQueue.Enqueue(ct => ExecuteJobAsync(jobId, ct)); + var tenant = tenantAccessor.Tenant; + jobQueue.Enqueue(ct => ExecuteJobAsync(jobId, tenant, ct)); return default; } - private async Task ExecuteJobAsync(string alterationJobId, CancellationToken cancellationToken) + private async Task ExecuteJobAsync(string alterationJobId, Tenant? tenant, CancellationToken cancellationToken) { - using var scope = scopeFactory.CreateScope(); - var alterationJobRunner = scope.ServiceProvider.GetRequiredService(); + await using var tenantScope = tenantScopeFactory.CreateScope(tenant); + var alterationJobRunner = tenantScope.ServiceProvider.GetRequiredService(); await alterationJobRunner.RunAsync(alterationJobId, cancellationToken); } } diff --git a/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs b/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs new file mode 100644 index 0000000000..ca72829670 --- /dev/null +++ b/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs @@ -0,0 +1,67 @@ +using Elsa.Alterations.Core.Contracts; +using Elsa.Alterations.Core.Entities; +using Elsa.Alterations.Services; +using Elsa.Common.Multitenancy; +using Elsa.Mediator.Contracts; +using Microsoft.Extensions.DependencyInjection; +using NSubstitute; + +namespace Elsa.Alterations.IntegrationTests; + +public class BackgroundAlterationJobDispatcherTests +{ + [Fact] + public async Task DispatchAsync_WhenQueuedWorkRunsAfterDispatchScopeEnds_PreservesTenant() + { + const string jobId = "alteration-job"; + Func? queuedCallback = null; + var jobQueue = Substitute.For(); + jobQueue + .Enqueue(Arg.Any>()) + .Returns(callInfo => + { + queuedCallback = callInfo.Arg>(); + return "queued-job"; + }); + + var tenantAccessor = new DefaultTenantAccessor(); + var runner = new RecordingAlterationJobRunner(tenantAccessor); + var services = new ServiceCollection() + .AddSingleton(jobQueue) + .AddSingleton(tenantAccessor) + .AddSingleton() + .AddScoped(_ => runner) + .AddScoped(); + await using var serviceProvider = services.BuildServiceProvider(validateScopes: true); + var dispatchingTenant = new Tenant { Id = "tenant-a", Name = "Tenant A" }; + var workerTenant = new Tenant { Id = "tenant-b", Name = "Tenant B" }; + + using (tenantAccessor.PushContext(dispatchingTenant)) + using (var dispatchScope = serviceProvider.CreateScope()) + { + var dispatcher = dispatchScope.ServiceProvider.GetRequiredService(); + await dispatcher.DispatchAsync(jobId); + } + + var callback = Assert.IsType>(queuedCallback); + using (tenantAccessor.PushContext(workerTenant)) + { + await callback(CancellationToken.None); + Assert.Same(workerTenant, tenantAccessor.Tenant); + } + + Assert.Equal(dispatchingTenant.Id, runner.ObservedTenantId); + Assert.Null(tenantAccessor.Tenant); + } + + private sealed class RecordingAlterationJobRunner(ITenantAccessor tenantAccessor) : IAlterationJobRunner + { + public string? ObservedTenantId { get; private set; } + + public Task RunAsync(string jobId, CancellationToken cancellationToken = default) + { + ObservedTenantId = tenantAccessor.TenantId; + return Task.FromResult(new AlterationJob { Id = jobId }); + } + } +} From 1f95ca93f09be5ea099ad4964697b14bb6475455 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Mon, 7 Sep 2026 04:51:38 +0200 Subject: [PATCH 2/3] test(alterations): cover tenant restore on failure and concurrent tenant isolation Add coverage for the spec lines that were previously untested: the worker's tenant context is restored after execution even when the job runner throws, and concurrent jobs dispatched under different tenants do not exchange tenant context. Extract the shared service provider / job queue / recording runner arrange logic into constructor-initialized fields so the three tests stay DRY. Co-Authored-By: Claude Fable 5.1 --- .../BackgroundAlterationJobDispatcherTests.cs | 122 ++++++++++++++---- 1 file changed, 96 insertions(+), 26 deletions(-) diff --git a/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs b/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs index ca72829670..aa7818c563 100644 --- a/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs +++ b/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using Elsa.Alterations.Core.Contracts; using Elsa.Alterations.Core.Entities; using Elsa.Alterations.Services; @@ -8,60 +9,129 @@ namespace Elsa.Alterations.IntegrationTests; -public class BackgroundAlterationJobDispatcherTests +public class BackgroundAlterationJobDispatcherTests : IAsyncLifetime { - [Fact] - public async Task DispatchAsync_WhenQueuedWorkRunsAfterDispatchScopeEnds_PreservesTenant() + private readonly List> _queuedCallbacks = []; + private readonly DefaultTenantAccessor _tenantAccessor; + private readonly RecordingAlterationJobRunner _runner; + private readonly ServiceProvider _serviceProvider; + + public BackgroundAlterationJobDispatcherTests() { - const string jobId = "alteration-job"; - Func? queuedCallback = null; var jobQueue = Substitute.For(); jobQueue .Enqueue(Arg.Any>()) .Returns(callInfo => { - queuedCallback = callInfo.Arg>(); - return "queued-job"; + var callback = callInfo.Arg>(); + _queuedCallbacks.Add(callback); + return $"queued-job-{_queuedCallbacks.Count}"; }); - var tenantAccessor = new DefaultTenantAccessor(); - var runner = new RecordingAlterationJobRunner(tenantAccessor); + _tenantAccessor = new DefaultTenantAccessor(); + _runner = new RecordingAlterationJobRunner(_tenantAccessor); var services = new ServiceCollection() .AddSingleton(jobQueue) - .AddSingleton(tenantAccessor) + .AddSingleton(_tenantAccessor) .AddSingleton() - .AddScoped(_ => runner) + .AddScoped(_ => _runner) .AddScoped(); - await using var serviceProvider = services.BuildServiceProvider(validateScopes: true); + _serviceProvider = services.BuildServiceProvider(validateScopes: true); + } + + public Task InitializeAsync() => Task.CompletedTask; + + public async Task DisposeAsync() => await _serviceProvider.DisposeAsync(); + + [Fact] + public async Task DispatchAsync_WhenQueuedWorkRunsAfterDispatchScopeEnds_PreservesTenant() + { + const string jobId = "alteration-job"; var dispatchingTenant = new Tenant { Id = "tenant-a", Name = "Tenant A" }; var workerTenant = new Tenant { Id = "tenant-b", Name = "Tenant B" }; - using (tenantAccessor.PushContext(dispatchingTenant)) - using (var dispatchScope = serviceProvider.CreateScope()) + await DispatchAsync(jobId, dispatchingTenant); + + var callback = Assert.Single(_queuedCallbacks); + using (_tenantAccessor.PushContext(workerTenant)) { - var dispatcher = dispatchScope.ServiceProvider.GetRequiredService(); - await dispatcher.DispatchAsync(jobId); + await callback(CancellationToken.None); + Assert.Same(workerTenant, _tenantAccessor.Tenant); } - var callback = Assert.IsType>(queuedCallback); - using (tenantAccessor.PushContext(workerTenant)) + Assert.Equal(dispatchingTenant.Id, _runner.GetObservedTenantId(jobId)); + Assert.Null(_tenantAccessor.Tenant); + } + + [Fact] + public async Task DispatchAsync_WhenRunnerThrows_StillRestoresWorkerTenant() + { + const string jobId = "alteration-job"; + var dispatchingTenant = new Tenant { Id = "tenant-a", Name = "Tenant A" }; + var workerTenant = new Tenant { Id = "tenant-b", Name = "Tenant B" }; + _runner.ExceptionToThrow = new InvalidOperationException("Runner failure"); + + await DispatchAsync(jobId, dispatchingTenant); + + var callback = Assert.Single(_queuedCallbacks); + using (_tenantAccessor.PushContext(workerTenant)) { - await callback(CancellationToken.None); - Assert.Same(workerTenant, tenantAccessor.Tenant); + await Assert.ThrowsAsync(() => callback(CancellationToken.None)); + Assert.Same(workerTenant, _tenantAccessor.Tenant); } - Assert.Equal(dispatchingTenant.Id, runner.ObservedTenantId); - Assert.Null(tenantAccessor.Tenant); + Assert.Equal(dispatchingTenant.Id, _runner.GetObservedTenantId(jobId)); + Assert.Null(_tenantAccessor.Tenant); + } + + [Fact] + public async Task DispatchAsync_WhenConcurrentJobsBelongToDifferentTenants_TenantsAreNotExchanged() + { + const string jobAId = "alteration-job-a"; + const string jobBId = "alteration-job-b"; + var tenantA = new Tenant { Id = "tenant-a", Name = "Tenant A" }; + var tenantB = new Tenant { Id = "tenant-b", Name = "Tenant B" }; + + await DispatchAsync(jobAId, tenantA); + await DispatchAsync(jobBId, tenantB); + + Assert.Equal(2, _queuedCallbacks.Count); + var callbackA = _queuedCallbacks[0]; + var callbackB = _queuedCallbacks[1]; + + await Task.WhenAll(callbackA(CancellationToken.None), callbackB(CancellationToken.None)); + + Assert.Equal(tenantA.Id, _runner.GetObservedTenantId(jobAId)); + Assert.Equal(tenantB.Id, _runner.GetObservedTenantId(jobBId)); + } + + private async Task DispatchAsync(string jobId, Tenant? tenant) + { + using (_tenantAccessor.PushContext(tenant)) + using (var dispatchScope = _serviceProvider.CreateScope()) + { + var dispatcher = dispatchScope.ServiceProvider.GetRequiredService(); + await dispatcher.DispatchAsync(jobId); + } } private sealed class RecordingAlterationJobRunner(ITenantAccessor tenantAccessor) : IAlterationJobRunner { - public string? ObservedTenantId { get; private set; } + private readonly ConcurrentDictionary _observedTenantIdsByJobId = new(); + + public Exception? ExceptionToThrow { get; set; } + + public string? GetObservedTenantId(string jobId) => _observedTenantIdsByJobId.TryGetValue(jobId, out var tenantId) ? tenantId : null; - public Task RunAsync(string jobId, CancellationToken cancellationToken = default) + public async Task RunAsync(string jobId, CancellationToken cancellationToken = default) { - ObservedTenantId = tenantAccessor.TenantId; - return Task.FromResult(new AlterationJob { Id = jobId }); + await Task.Yield(); + _observedTenantIdsByJobId[jobId] = tenantAccessor.TenantId; + + if (ExceptionToThrow is not null) + throw ExceptionToThrow; + + return new AlterationJob { Id = jobId }; } } } From e044dc74679ff2a05bd5797262756dcc3ed0ec7f Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Mon, 7 Sep 2026 04:55:37 +0200 Subject: [PATCH 3/3] test(alterations): cover default-tenant dispatch and share dispatcher test setup Co-Authored-By: Claude Fable 5.1 --- .../BackgroundAlterationJobDispatcherTests.cs | 51 ++++++++++++++----- 1 file changed, 39 insertions(+), 12 deletions(-) diff --git a/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs b/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs index aa7818c563..4c3fa73e0c 100644 --- a/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs +++ b/test/integration/Elsa.Alterations.IntegrationTests/BackgroundAlterationJobDispatcherTests.cs @@ -50,17 +50,7 @@ public async Task DispatchAsync_WhenQueuedWorkRunsAfterDispatchScopeEnds_Preserv var dispatchingTenant = new Tenant { Id = "tenant-a", Name = "Tenant A" }; var workerTenant = new Tenant { Id = "tenant-b", Name = "Tenant B" }; - await DispatchAsync(jobId, dispatchingTenant); - - var callback = Assert.Single(_queuedCallbacks); - using (_tenantAccessor.PushContext(workerTenant)) - { - await callback(CancellationToken.None); - Assert.Same(workerTenant, _tenantAccessor.Tenant); - } - - Assert.Equal(dispatchingTenant.Id, _runner.GetObservedTenantId(jobId)); - Assert.Null(_tenantAccessor.Tenant); + await DispatchAndRunUnderWorkerTenantAsync(jobId, dispatchingTenant, workerTenant, callback => callback(CancellationToken.None)); } [Fact] @@ -71,12 +61,49 @@ public async Task DispatchAsync_WhenRunnerThrows_StillRestoresWorkerTenant() var workerTenant = new Tenant { Id = "tenant-b", Name = "Tenant B" }; _runner.ExceptionToThrow = new InvalidOperationException("Runner failure"); + await DispatchAndRunUnderWorkerTenantAsync( + jobId, + dispatchingTenant, + workerTenant, + callback => Assert.ThrowsAsync(() => callback(CancellationToken.None))); + } + + [Fact] + public async Task DispatchAsync_WhenNoTenantIsPushedAtDispatchTime_UsesDefaultTenant() + { + const string jobId = "alteration-job"; + + await DispatchAsync(jobId, tenant: null); + + var callback = Assert.Single(_queuedCallbacks); + await callback(CancellationToken.None); + + // With no tenant pushed at dispatch time, DefaultTenantScopeFactory.CreateScope(null) pushes a null + // tenant onto the accessor. DefaultTenantAccessor.TenantId then falls back to Tenant.DefaultTenantId + // (an empty string) rather than null, so that is the value the runner observes. + Assert.Equal(Tenant.DefaultTenantId, _runner.GetObservedTenantId(jobId)); + Assert.Null(_tenantAccessor.Tenant); + } + + /// + /// Dispatches a job under , then runs the single queued callback while a + /// different is active on the accessor, invoking it via + /// so callers can assert success or failure. Asserts the tenant behavior + /// common to both outcomes: the worker tenant remains active for the duration of the callback, the runner + /// observed the dispatching tenant, and the accessor's tenant is restored to null afterward. + /// + private async Task DispatchAndRunUnderWorkerTenantAsync( + string jobId, + Tenant dispatchingTenant, + Tenant workerTenant, + Func, Task> runCallbackAsync) + { await DispatchAsync(jobId, dispatchingTenant); var callback = Assert.Single(_queuedCallbacks); using (_tenantAccessor.PushContext(workerTenant)) { - await Assert.ThrowsAsync(() => callback(CancellationToken.None)); + await runCallbackAsync(callback); Assert.Same(workerTenant, _tenantAccessor.Tenant); }