| | | 1 | | using Microsoft.Extensions.DependencyInjection; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | using System.Diagnostics.CodeAnalysis; |
| | | 4 | | |
| | | 5 | | namespace AsyncResponse; |
| | | 6 | | |
| | | 7 | | /// <inheritdoc cref="IDurableFlows" /> |
| | | 8 | | internal sealed class DurableFlowService : IDurableFlows |
| | | 9 | | { |
| | | 10 | | private readonly IServiceScopeFactory _scopeFactory; |
| | | 11 | | private readonly IAsyncResponseBuilder _builder; |
| | | 12 | | private readonly AsyncResponseContextPropagation _propagation; |
| | | 13 | | private readonly DurableFlowOptions _options; |
| | | 14 | | private readonly ILogger<DurableFlowService> _logger; |
| | | 15 | | |
| | | 16 | | /// <summary>Creates the durable-flows starter.</summary> |
| | 3 | 17 | | public DurableFlowService( |
| | 3 | 18 | | IServiceScopeFactory scopeFactory, |
| | 3 | 19 | | IAsyncResponseBuilder builder, |
| | 3 | 20 | | AsyncResponseContextPropagation propagation, |
| | 3 | 21 | | DurableFlowOptions options, |
| | 3 | 22 | | ILogger<DurableFlowService> logger) |
| | | 23 | | { |
| | 3 | 24 | | _scopeFactory = scopeFactory; |
| | 3 | 25 | | _builder = builder; |
| | 3 | 26 | | _propagation = propagation; |
| | 3 | 27 | | _options = options; |
| | 3 | 28 | | FlowStateConcurrency.ValidateOptions(_options); |
| | 3 | 29 | | _logger = logger; |
| | 3 | 30 | | } |
| | | 31 | | |
| | | 32 | | /// <inheritdoc /> |
| | | 33 | | public async Task<string> StartAsync<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors | |
| | | 34 | | TInput input, |
| | | 35 | | string? flowId = null, |
| | | 36 | | CancellationToken cancellationToken = default) |
| | | 37 | | where TFlow : class, IDurableFlow<TInput> |
| | | 38 | | { |
| | 3 | 39 | | ArgumentNullException.ThrowIfNull(input); |
| | 3 | 40 | | cancellationToken.ThrowIfCancellationRequested(); |
| | 3 | 41 | | if (flowId is null) |
| | 3 | 42 | | flowId = $"flow-{AsyncResponseContext.GenerateCorrelationId()}"; |
| | | 43 | | else |
| | 3 | 44 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | | 45 | | |
| | 3 | 46 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 3 | 47 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | | 48 | | |
| | 3 | 49 | | var now = DateTime.UtcNow; |
| | 3 | 50 | | var inputJson = AsyncResponseJson.Serialize(input); |
| | 3 | 51 | | var state = new FlowState |
| | 3 | 52 | | { |
| | 3 | 53 | | FlowId = flowId, |
| | 3 | 54 | | FlowTypeName = typeof(TFlow).FullName, |
| | 3 | 55 | | InputTypeName = typeof(TInput).FullName, |
| | 3 | 56 | | InputJson = inputJson, |
| | 3 | 57 | | Status = FlowRunStatus.Running, |
| | 3 | 58 | | LastMessage = "Flow started.", |
| | 3 | 59 | | CreatedAtUtc = now, |
| | 3 | 60 | | UpdatedAtUtc = now, |
| | 3 | 61 | | Context = _propagation.Capture() |
| | 3 | 62 | | }; |
| | | 63 | | |
| | 3 | 64 | | if (await FlowStateConcurrency.TryCreateAsync( |
| | 3 | 65 | | store, |
| | 3 | 66 | | flowId, |
| | 3 | 67 | | state, |
| | 3 | 68 | | _options.StateExpiry, |
| | 3 | 69 | | cancellationToken).ConfigureAwait(false)) |
| | | 70 | | { |
| | 3 | 71 | | _logger.LogInformation("Started durable flow {FlowId} ({FlowType}).", flowId, typeof(TFlow).Name); |
| | | 72 | | } |
| | | 73 | | else |
| | | 74 | | { |
| | 3 | 75 | | var existing = await store.LoadAsync(flowId, cancellationToken).ConfigureAwait(false) |
| | 3 | 76 | | ?? throw new InvalidOperationException( |
| | 3 | 77 | | $"Durable flow '{flowId}' already exists but its ledger is expired or unreadable."); |
| | 3 | 78 | | EnsureIdempotentStart<TFlow, TInput>(existing, inputJson, flowId); |
| | | 79 | | |
| | | 80 | | // A semantically identical retry re-enqueues the existing run; completed steps skip. |
| | 3 | 81 | | _logger.LogInformation("Durable flow {FlowId} already exists; re-enqueueing instead of creating a duplicate. |
| | | 82 | | } |
| | | 83 | | |
| | 3 | 84 | | var id = flowId; |
| | 3 | 85 | | await _builder.EnqueueWorkerAsync<IDurableFlowExecutor>( |
| | 3 | 86 | | executor => executor.ExecuteAsync(id), |
| | 3 | 87 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 88 | | return flowId; |
| | 3 | 89 | | } |
| | | 90 | | |
| | | 91 | | /// <inheritdoc /> |
| | | 92 | | public async Task ResumeAsync(string flowId, CancellationToken cancellationToken = default) |
| | | 93 | | { |
| | 3 | 94 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | | 95 | | |
| | 3 | 96 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 3 | 97 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | | 98 | | |
| | 3 | 99 | | var state = await store.LoadAsync(flowId, cancellationToken).ConfigureAwait(false) |
| | 3 | 100 | | ?? throw new InvalidOperationException($"No flow state found for '{flowId}' (unknown, expired, or unreadable |
| | | 101 | | |
| | 3 | 102 | | if (state.Status != FlowRunStatus.Running) |
| | | 103 | | { |
| | 3 | 104 | | _logger.LogDebug("Durable flow {FlowId} is already {Status}; ignoring resume.", flowId, state.Status); |
| | 3 | 105 | | return; |
| | | 106 | | } |
| | | 107 | | |
| | 2 | 108 | | var id = flowId; |
| | 2 | 109 | | await _builder.EnqueueWorkerAsync<IDurableFlowExecutor>( |
| | 2 | 110 | | executor => executor.ExecuteAsync(id), |
| | 2 | 111 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 112 | | } |
| | | 113 | | |
| | | 114 | | /// <inheritdoc /> |
| | | 115 | | public async Task<FlowState?> GetStateAsync(string flowId, CancellationToken cancellationToken = default) |
| | | 116 | | { |
| | 3 | 117 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | | 118 | | |
| | 3 | 119 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 3 | 120 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | 3 | 121 | | return await store.LoadAsync(flowId, cancellationToken).ConfigureAwait(false); |
| | 3 | 122 | | } |
| | | 123 | | |
| | | 124 | | private static void EnsureIdempotentStart<TFlow, TInput>( |
| | | 125 | | FlowState existing, |
| | | 126 | | string requestedInputJson, |
| | | 127 | | string flowId) |
| | | 128 | | { |
| | 3 | 129 | | var sameFlowType = string.Equals(existing.FlowTypeName, typeof(TFlow).FullName, StringComparison.Ordinal); |
| | 3 | 130 | | var sameInputType = string.Equals(existing.InputTypeName, typeof(TInput).FullName, StringComparison.Ordinal); |
| | 3 | 131 | | if (sameFlowType && sameInputType && FlowStateJson.JsonEquivalent(existing.InputJson, requestedInputJson)) |
| | 3 | 132 | | return; |
| | | 133 | | |
| | 3 | 134 | | throw new InvalidOperationException( |
| | 3 | 135 | | $"Durable flow id '{flowId}' is already bound to a different flow type or input. " + |
| | 3 | 136 | | "Idempotent retries must use the same TFlow, TInput, and semantically identical input value."); |
| | | 137 | | } |
| | | 138 | | |
| | | 139 | | } |