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

Fix tenant scope for interrupted workflow restarts (#7389)

Browse source 
* Fix tenant scope for interrupted workflow restarts

* Update src/modules/Elsa.Workflows.Runtime/Tasks/RestartInterruptedWorkflowsTask.cs

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* Handle per-instance restart failures

* fix: make restart interrupted workflows tenant services optional

---------

Co-authored-by: Ralf <Ralf@Careconnections.nl>
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
This commit is contained in:
RalfvandenBurg 2026-04-15 14:11:26 +02:00 committed by GitHub
parent 3a37668619
commit 793ae15bf7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 234 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
src/modules/Elsa.Workflows.Management/Models/WorkflowInstanceSummary.cs
View file

@ -16,6 +16,7 @@ public class WorkflowInstanceSummary
return new()
{
Id = workflowInstance.Id,
TenantId = workflowInstance.TenantId,
DefinitionId = workflowInstance.DefinitionId,
DefinitionVersionId = workflowInstance.DefinitionVersionId,
Version = workflowInstance.Version,
@ -36,6 +37,7 @@ public class WorkflowInstanceSummary
public static Expression<Func<WorkflowInstance, WorkflowInstanceSummary>> FromInstanceExpression() => workflowInstance => new()
{
Id = workflowInstance.Id,
TenantId = workflowInstance.TenantId,
DefinitionId = workflowInstance.DefinitionId,
DefinitionVersionId = workflowInstance.DefinitionVersionId,
Version = workflowInstance.Version,
@ -53,6 +55,9 @@ public class WorkflowInstanceSummary
/// <summary>The ID of the workflow instance.</summary>
public string Id { get; set; } = null!;
/// <summary>The ID of the tenant that owns the workflow instance.</summary>
public string? TenantId { get; set; }
/// <summary>The ID of the workflow definition.</summary>
public string DefinitionId { get; set; } = null!;

28
src/modules/Elsa.Workflows.Runtime/Tasks/RestartInterruptedWorkflowsTask.cs
View file

@ -1,4 +1,5 @@
using Elsa.Common;
using Elsa.Common.Multitenancy;
using Elsa.Common.RecurringTasks;
using Elsa.Workflows.Management;
using Elsa.Workflows.Management.Filters;
@ -12,11 +13,13 @@ namespace Elsa.Workflows.Runtime.Tasks;
[SingleNodeTask]
[UsedImplicitly]
public class RestartInterruptedWorkflowsTask(
IWorkflowInstanceStore workflowInstanceStore,
IWorkflowRestarter workflowRestarter,
IWorkflowInstanceStore workflowInstanceStore,
ILogger<RestartInterruptedWorkflowsTask> logger,
IOptions<RuntimeOptions> options,
ISystemClock systemClock,
ILogger<RestartInterruptedWorkflowsTask> logger) : RecurringTask
ITenantService? tenantService = null,
ITenantAccessor? tenantAccessor = null) : RecurringTask
{
/// <inheritdoc />
public override async Task ExecuteAsync(CancellationToken cancellationToken)
@ -28,7 +31,26 @@ public class RestartInterruptedWorkflowsTask(
logger.LogInformation("Restarting interrupted workflows.");
await foreach (var workflowInstance in workflowInstances)
{
await workflowRestarter.RestartWorkflowAsync(workflowInstance.Id, cancellationToken: cancellationToken);
try
{
var tenantId = workflowInstance.TenantId ?? string.Empty;
if (tenantService != null && tenantAccessor != null && !string.IsNullOrWhiteSpace(tenantId) && tenantId != Tenant.AgnosticTenantId)
{
var tenant = await tenantService.FindAsync(tenantId, cancellationToken) ?? new Tenant { Id = tenantId, Name = tenantId };
using (tenantAccessor.PushContext(tenant))
await workflowRestarter.RestartWorkflowAsync(workflowInstance.Id, cancellationToken: cancellationToken);
continue;
}
await workflowRestarter.RestartWorkflowAsync(workflowInstance.Id, cancellationToken: cancellationToken);
}
catch (Exception ex)
{
logger.LogError(ex, "Failed to restart interrupted workflow {WorkflowInstanceId}", workflowInstance.Id);
}
}
logger.LogInformation("Finished restarting interrupted workflows.");
}

204
test/unit/Elsa.Workflows.Runtime.UnitTests/Services/RestartInterruptedWorkflowsTaskTests.cs Normal file
View file

@ -0,0 +1,204 @@
using Elsa.Common;
using Elsa.Common.Models;
using Elsa.Common.Multitenancy;
using Elsa.Workflows.Management;
using Elsa.Workflows.Management.Filters;
using Elsa.Workflows.Management.Models;
using Elsa.Workflows.Runtime.Options;
using Elsa.Workflows.Runtime.Tasks;
using Microsoft.Extensions.Logging;
using NSubstitute;
namespace Elsa.Workflows.Runtime.UnitTests.Services;
public class RestartInterruptedWorkflowsTaskTests
{
[Fact(DisplayName = "ExecuteAsync restarts each workflow within its tenant context")]
public async Task ExecuteAsync_RestartsWithinWorkflowTenantContext()
{
var workflowInstanceStore = Substitute.For<IWorkflowInstanceStore>();
var tenantAccessor = new DefaultTenantAccessor();
var tenantService = Substitute.For<ITenantService>();
var workflowRestarter = Substitute.For<IWorkflowRestarter>();
var clock = Substitute.For<ISystemClock>();
var logger = Substitute.For<ILogger<RestartInterruptedWorkflowsTask>>();
var now = DateTimeOffset.Parse("2026-04-14T12:00:00Z");
var workflowInstances = new List<WorkflowInstanceSummary>
{
CreateWorkflowInstance("workflow-1", "tenant-a", now),
CreateWorkflowInstance("workflow-2", "tenant-b", now)
};
var observedTenantIds = new List<string?>();
clock.UtcNow.Returns(now);
tenantService.FindAsync("tenant-a", Arg.Any<CancellationToken>()).Returns(new Tenant { Id = "tenant-a", Name = "Tenant A" });
tenantService.FindAsync("tenant-b", Arg.Any<CancellationToken>()).Returns(new Tenant { Id = "tenant-b", Name = "Tenant B" });
workflowInstanceStore
.SummarizeManyAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<PageArgs>(), Arg.Any<CancellationToken>())
.Returns(
callInfo =>
{
var pageArgs = callInfo.ArgAt<PageArgs>(1);
var items = pageArgs.Offset == 0 ? workflowInstances : new List<WorkflowInstanceSummary>();
return new ValueTask<Page<WorkflowInstanceSummary>>(Page.Of(items, workflowInstances.Count));
});
workflowRestarter
.When(x => x.RestartWorkflowAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()))
.Do(_ => observedTenantIds.Add(tenantAccessor.Tenant?.Id));
var options = Microsoft.Extensions.Options.Options.Create(new RuntimeOptions
{
RestartInterruptedWorkflowsBatchSize = 10,
InactivityThreshold = TimeSpan.FromMinutes(5)
});
var task = new RestartInterruptedWorkflowsTask(workflowRestarter, workflowInstanceStore, logger, options, clock, tenantService, tenantAccessor);
await task.ExecuteAsync(CancellationToken.None);
Assert.Equal(new[] { "tenant-a", "tenant-b" }, observedTenantIds);
Assert.Null(tenantAccessor.Tenant);
await workflowRestarter.Received(1).RestartWorkflowAsync("workflow-1", Arg.Any<CancellationToken>());
await workflowRestarter.Received(1).RestartWorkflowAsync("workflow-2", Arg.Any<CancellationToken>());
}
[Theory(DisplayName = "ExecuteAsync does not push tenant context for default or agnostic tenant instances")]
[InlineData(null)]
[InlineData("")]
[InlineData("*")]
public async Task ExecuteAsync_DefaultOrAgnosticTenant_DoesNotPushContext(string? tenantId)
{
var workflowInstanceStore = Substitute.For<IWorkflowInstanceStore>();
var tenantAccessor = new DefaultTenantAccessor();
var tenantService = Substitute.For<ITenantService>();
var workflowRestarter = Substitute.For<IWorkflowRestarter>();
var clock = Substitute.For<ISystemClock>();
var logger = Substitute.For<ILogger<RestartInterruptedWorkflowsTask>>();
var now = DateTimeOffset.Parse("2026-04-14T12:00:00Z");
var workflowInstances = new List<WorkflowInstanceSummary>
{
CreateWorkflowInstance("workflow-1", tenantId, now)
};
var observedTenantIds = new List<string?>();
clock.UtcNow.Returns(now);
workflowInstanceStore
.SummarizeManyAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<PageArgs>(), Arg.Any<CancellationToken>())
.Returns(
callInfo =>
{
var pageArgs = callInfo.ArgAt<PageArgs>(1);
var items = pageArgs.Offset == 0 ? workflowInstances : new List<WorkflowInstanceSummary>();
return new ValueTask<Page<WorkflowInstanceSummary>>(Page.Of(items, workflowInstances.Count));
});
workflowRestarter
.When(x => x.RestartWorkflowAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()))
.Do(_ => observedTenantIds.Add(tenantAccessor.Tenant?.Id));
var options = Microsoft.Extensions.Options.Options.Create(new RuntimeOptions
{
RestartInterruptedWorkflowsBatchSize = 10,
InactivityThreshold = TimeSpan.FromMinutes(5)
});
var task = new RestartInterruptedWorkflowsTask(workflowRestarter, workflowInstanceStore, logger, options, clock, tenantService, tenantAccessor);
await task.ExecuteAsync(CancellationToken.None);
Assert.Equal(new string?[] { null }, observedTenantIds);
Assert.Null(tenantAccessor.Tenant);
await tenantService.DidNotReceive().FindAsync(Arg.Any<string>(), Arg.Any<CancellationToken>());
await workflowRestarter.Received(1).RestartWorkflowAsync("workflow-1", Arg.Any<CancellationToken>());
}
[Fact(DisplayName = "ExecuteAsync continues restarting later workflows after a failure")]
public async Task ExecuteAsync_ContinuesAfterFailure()
{
var workflowInstanceStore = Substitute.For<IWorkflowInstanceStore>();
var tenantAccessor = new DefaultTenantAccessor();
var tenantService = Substitute.For<ITenantService>();
var workflowRestarter = Substitute.For<IWorkflowRestarter>();
var clock = Substitute.For<ISystemClock>();
var logger = Substitute.For<ILogger<RestartInterruptedWorkflowsTask>>();
var now = DateTimeOffset.Parse("2026-04-14T12:00:00Z");
var workflowInstances = new List<WorkflowInstanceSummary>
{
CreateWorkflowInstance("workflow-1", "tenant-a", now),
CreateWorkflowInstance("workflow-2", "tenant-b", now)
};
clock.UtcNow.Returns(now);
tenantService.FindAsync("tenant-a", Arg.Any<CancellationToken>()).Returns(_ => Task.FromException<Tenant?>(new InvalidOperationException("Transient tenant lookup failure")));
tenantService.FindAsync("tenant-b", Arg.Any<CancellationToken>()).Returns(new Tenant { Id = "tenant-b", Name = "Tenant B" });
workflowInstanceStore
.SummarizeManyAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<PageArgs>(), Arg.Any<CancellationToken>())
.Returns(
callInfo =>
{
var pageArgs = callInfo.ArgAt<PageArgs>(1);
var items = pageArgs.Offset == 0 ? workflowInstances : new List<WorkflowInstanceSummary>();
return new ValueTask<Page<WorkflowInstanceSummary>>(Page.Of(items, workflowInstances.Count));
});
var options = Microsoft.Extensions.Options.Options.Create(new RuntimeOptions
{
RestartInterruptedWorkflowsBatchSize = 10,
InactivityThreshold = TimeSpan.FromMinutes(5)
});
var task = new RestartInterruptedWorkflowsTask(workflowRestarter, workflowInstanceStore, logger, options, clock, tenantService, tenantAccessor);
await task.ExecuteAsync(CancellationToken.None);
await workflowRestarter.DidNotReceive().RestartWorkflowAsync("workflow-1", Arg.Any<CancellationToken>());
await workflowRestarter.Received(1).RestartWorkflowAsync("workflow-2", Arg.Any<CancellationToken>());
}
[Fact(DisplayName = "ExecuteAsync restarts tenant-specific workflows when tenant services are unavailable")]
public async Task ExecuteAsync_WithoutTenantServices_RestartsWorkflow()
{
var workflowInstanceStore = Substitute.For<IWorkflowInstanceStore>();
var workflowRestarter = Substitute.For<IWorkflowRestarter>();
var clock = Substitute.For<ISystemClock>();
var logger = Substitute.For<ILogger<RestartInterruptedWorkflowsTask>>();
var now = DateTimeOffset.Parse("2026-04-14T12:00:00Z");
var workflowInstances = new List<WorkflowInstanceSummary>
{
CreateWorkflowInstance("workflow-1", "tenant-a", now)
};
clock.UtcNow.Returns(now);
workflowInstanceStore
.SummarizeManyAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<PageArgs>(), Arg.Any<CancellationToken>())
.Returns(
callInfo =>
{
var pageArgs = callInfo.ArgAt<PageArgs>(1);
var items = pageArgs.Offset == 0 ? workflowInstances : new List<WorkflowInstanceSummary>();
return new ValueTask<Page<WorkflowInstanceSummary>>(Page.Of(items, workflowInstances.Count));
});
var options = Microsoft.Extensions.Options.Options.Create(new RuntimeOptions
{
RestartInterruptedWorkflowsBatchSize = 10,
InactivityThreshold = TimeSpan.FromMinutes(5)
});
var task = new RestartInterruptedWorkflowsTask(workflowRestarter, workflowInstanceStore, logger, options, clock);
await task.ExecuteAsync(CancellationToken.None);
await workflowRestarter.Received(1).RestartWorkflowAsync("workflow-1", Arg.Any<CancellationToken>());
}
private static WorkflowInstanceSummary CreateWorkflowInstance(string id, string? tenantId, DateTimeOffset updatedAt)
{
return new WorkflowInstanceSummary
{
Id = id,
TenantId = tenantId,
DefinitionId = "definition",
DefinitionVersionId = "definition:1",
Status = WorkflowStatus.Running,
SubStatus = WorkflowSubStatus.Pending,
CreatedAt = updatedAt.AddMinutes(-10),
UpdatedAt = updatedAt.AddMinutes(-10)
};
}
}