| | | 1 | | using Elsa.Common; |
| | | 2 | | using Elsa.Common.Multitenancy; |
| | | 3 | | using Elsa.Workflows.Management; |
| | | 4 | | using Elsa.Workflows.Management.Filters; |
| | | 5 | | using Elsa.Workflows.Runtime.Options; |
| | | 6 | | using Microsoft.Extensions.Logging; |
| | | 7 | | using Microsoft.Extensions.Options; |
| | | 8 | | |
| | | 9 | | namespace Elsa.Workflows.Runtime.Services; |
| | | 10 | | |
| | | 11 | | /// <summary> |
| | | 12 | | /// Default implementation of <see cref="IInterruptedRecoveryScanner"/>. |
| | | 13 | | /// </summary> |
| | | 14 | | public sealed class InterruptedRecoveryScanner : IInterruptedRecoveryScanner |
| | | 15 | | { |
| | | 16 | | private readonly IWorkflowRestarter _restarter; |
| | | 17 | | private readonly IWorkflowInstanceStore _instanceStore; |
| | | 18 | | private readonly IOptions<RuntimeOptions> _runtimeOptions; |
| | | 19 | | private readonly ILogger<InterruptedRecoveryScanner> _logger; |
| | | 20 | | private readonly ITenantService? _tenantService; |
| | | 21 | | private readonly ITenantAccessor? _tenantAccessor; |
| | | 22 | | |
| | 108 | 23 | | public InterruptedRecoveryScanner( |
| | 108 | 24 | | IWorkflowRestarter restarter, |
| | 108 | 25 | | IWorkflowInstanceStore instanceStore, |
| | 108 | 26 | | IOptions<RuntimeOptions> runtimeOptions, |
| | 108 | 27 | | ILogger<InterruptedRecoveryScanner> logger, |
| | 108 | 28 | | ITenantService? tenantService = null, |
| | 108 | 29 | | ITenantAccessor? tenantAccessor = null) |
| | | 30 | | { |
| | 108 | 31 | | _restarter = restarter; |
| | 108 | 32 | | _instanceStore = instanceStore; |
| | 108 | 33 | | _runtimeOptions = runtimeOptions; |
| | 108 | 34 | | _logger = logger; |
| | 108 | 35 | | _tenantService = tenantService; |
| | 108 | 36 | | _tenantAccessor = tenantAccessor; |
| | 108 | 37 | | } |
| | | 38 | | |
| | | 39 | | /// <inheritdoc /> |
| | | 40 | | public async ValueTask<int> ScanAndRequeueAsync(CancellationToken cancellationToken) |
| | | 41 | | { |
| | 108 | 42 | | var filter = new WorkflowInstanceFilter { WorkflowSubStatus = WorkflowSubStatus.Interrupted }; |
| | 108 | 43 | | var batchSize = _runtimeOptions.Value.RestartInterruptedWorkflowsBatchSize; |
| | 108 | 44 | | var instances = _instanceStore.EnumerateSummariesAsync(filter, batchSize, cancellationToken); |
| | 108 | 45 | | var requeued = 0; |
| | | 46 | | |
| | 108 | 47 | | _logger.LogInformation("Scanning for workflows interrupted by graceful drain."); |
| | | 48 | | |
| | 470 | 49 | | await foreach (var summary in instances.WithCancellation(cancellationToken)) |
| | | 50 | | { |
| | | 51 | | try |
| | | 52 | | { |
| | 127 | 53 | | var tenantId = summary.TenantId ?? string.Empty; |
| | | 54 | | |
| | 127 | 55 | | if (_tenantService is not null && _tenantAccessor is not null && !string.IsNullOrWhiteSpace(tenantId) && |
| | | 56 | | { |
| | 1 | 57 | | var tenant = await _tenantService.FindAsync(tenantId, cancellationToken) ?? new Tenant { Id = tenant |
| | | 58 | | |
| | 1 | 59 | | using (_tenantAccessor.PushContext(tenant)) |
| | 1 | 60 | | await _restarter.RestartWorkflowAsync(summary.Id, cancellationToken); |
| | | 61 | | } |
| | | 62 | | else |
| | | 63 | | { |
| | 126 | 64 | | await _restarter.RestartWorkflowAsync(summary.Id, cancellationToken); |
| | | 65 | | } |
| | | 66 | | |
| | 126 | 67 | | requeued++; |
| | 126 | 68 | | } |
| | 1 | 69 | | catch (Exception ex) when (!ex.IsFatal()) |
| | | 70 | | { |
| | 1 | 71 | | _logger.LogError(ex, "Failed to requeue interrupted workflow {WorkflowInstanceId}; will be retried by th |
| | 1 | 72 | | } |
| | 127 | 73 | | } |
| | | 74 | | |
| | 108 | 75 | | _logger.LogInformation("Interrupted-workflow scan complete; requeued {Count} instance(s).", requeued); |
| | 108 | 76 | | return requeued; |
| | 108 | 77 | | } |
| | | 78 | | } |