| | | 1 | | using Microsoft.Extensions.DependencyInjection; |
| | | 2 | | using Microsoft.Extensions.Hosting; |
| | | 3 | | using Microsoft.Extensions.Logging; |
| | | 4 | | using Microsoft.Extensions.Options; |
| | | 5 | | using System.Reflection; |
| | | 6 | | |
| | | 7 | | namespace AsyncResponse; |
| | | 8 | | |
| | | 9 | | /// <summary> |
| | | 10 | | /// Internal marker registered by each response-channel registration |
| | | 11 | | /// (<c>.WithInMemoryChannel()</c> / <c>.WithRedisChannel()</c>). The |
| | | 12 | | /// <see cref="AsyncResponseStartupValidator"/> asserts exactly one channel is present. |
| | | 13 | | /// </summary> |
| | | 14 | | internal sealed class AsyncResponseChannelMarker(string name) |
| | | 15 | | { |
| | | 16 | | public string Name { get; } = name; |
| | | 17 | | |
| | | 18 | | /// <summary> |
| | | 19 | | /// The channel's RESOLVED default waiter timeout (<c>DefaultTimeout ?? RecoveryStateExpiry</c>), |
| | | 20 | | /// declared by the channel registration from its bound options — the window a timeout-less |
| | | 21 | | /// wait actually runs for. The startup validator requires the durable-flow ledger TTL to |
| | | 22 | | /// out-live it, and the flow engine extends a parked ledger by it, without either referencing |
| | | 23 | | /// channel option types. <c>null</c> when the registration declares nothing; both consumers |
| | | 24 | | /// then skip their checks. |
| | | 25 | | /// </summary> |
| | | 26 | | public TimeSpan? EffectiveDefaultWaitTimeout { get; init; } |
| | | 27 | | } |
| | | 28 | | |
| | | 29 | | /// <summary> |
| | | 30 | | /// Internal marker registered by each worker-transport registration |
| | | 31 | | /// (<c>.WithInMemoryTransport()</c> / <c>.WithGooglePubSubTransport(...)</c>). The |
| | | 32 | | /// <see cref="AsyncResponseStartupValidator"/> asserts exactly one transport is present. |
| | | 33 | | /// </summary> |
| | | 34 | | internal sealed class AsyncResponseTransportMarker(string name) |
| | | 35 | | { |
| | | 36 | | public string Name { get; } = name; |
| | | 37 | | |
| | | 38 | | /// <summary> |
| | | 39 | | /// Whether the transport's worker subscriber resolved to early ACK (<c>AckAfterEnqueue</c>). |
| | | 40 | | /// Declared by the transport's registration from its bound options so the startup validator |
| | | 41 | | /// can veto the combination with durable flows without referencing transport types. |
| | | 42 | | /// </summary> |
| | | 43 | | public bool WorkerSubscriberUsesEarlyAck { get; init; } |
| | | 44 | | |
| | | 45 | | /// <summary>Worker ack-mode option path, shown in the startup error.</summary> |
| | | 46 | | public string? WorkerAckModePath { get; init; } |
| | | 47 | | |
| | | 48 | | /// <summary>Whether the transport's response subscriber resolved to early ACK.</summary> |
| | | 49 | | public bool ResponseSubscriberUsesEarlyAck { get; init; } |
| | | 50 | | |
| | | 51 | | /// <summary>Response ack-mode option path, shown in the startup warning.</summary> |
| | | 52 | | public string? ResponseAckModePath { get; init; } |
| | | 53 | | } |
| | | 54 | | |
| | | 55 | | /// <summary>Internal marker registered by each durable-flow state-store registration.</summary> |
| | | 56 | | internal sealed class AsyncResponseDurableFlowStoreMarker( |
| | | 57 | | Type storeType, |
| | | 58 | | ServiceLifetime? forwardLifetime = null, |
| | | 59 | | IServiceCollection? services = null) |
| | | 60 | | { |
| | | 61 | | private IServiceCollection? _services = services; |
| | | 62 | | |
| | | 63 | | public Type StoreType { get; } = storeType; |
| | | 64 | | public string Name { get; } = storeType.FullName ?? storeType.Name; |
| | | 65 | | |
| | | 66 | | /// <summary> |
| | | 67 | | /// The <see cref="IFlowStateStore"/> forward mirrors the concrete registration's lifetime as |
| | | 68 | | /// seen WHEN <c>WithDurableFlows</c> ran. A concrete registration added after the fluent chain |
| | | 69 | | /// with a different lifetime leaves that snapshot stale — the worst shape being a Scoped |
| | | 70 | | /// forward to a root singleton, which every flow-execution scope captures as its own |
| | | 71 | | /// disposable: the first scope's disposal kills the store (and any connection it owns) for |
| | | 72 | | /// the whole process. MS.DI resolves the concrete last-wins, so the mismatch is re-checked |
| | | 73 | | /// here against the FINAL collection and fails startup with the ordering fix instead. Holds |
| | | 74 | | /// the service collection only until the check runs, then releases it. |
| | | 75 | | /// </summary> |
| | | 76 | | public void ValidateForwardLifetime() |
| | | 77 | | { |
| | | 78 | | var services = Interlocked.Exchange(ref _services, null); |
| | | 79 | | if (services is null || forwardLifetime is null) |
| | | 80 | | return; |
| | | 81 | | |
| | | 82 | | var finalLifetime = services.LastOrDefault(descriptor => descriptor.ServiceType == StoreType)?.Lifetime; |
| | | 83 | | if (finalLifetime is null || finalLifetime == forwardLifetime) |
| | | 84 | | return; |
| | | 85 | | |
| | | 86 | | throw new InvalidOperationException( |
| | | 87 | | $"The durable-flow store '{Name}' is registered as {finalLifetime}, but the IFlowStateStore forward mirrored |
| | | 88 | | $"{forwardLifetime} when WithDurableFlows<{StoreType.Name}>() ran — the store registration was added or chan |
| | | 89 | | "after the fluent chain. " + |
| | | 90 | | (forwardLifetime == ServiceLifetime.Scoped && finalLifetime == ServiceLifetime.Singleton |
| | | 91 | | ? "A Scoped forward to a Singleton store lets the first flow-execution scope dispose the store (and any |
| | | 92 | | : "The forward and the store must share one lifetime, or resolutions through the interface and the concr |
| | | 93 | | $"Register the store (e.g. services.Add{finalLifetime}<{StoreType.Name}>()) BEFORE the WithDurableFlows call |
| | | 94 | | } |
| | | 95 | | } |
| | | 96 | | |
| | | 97 | | /// <summary> |
| | | 98 | | /// Registered by the durable-flow engine and evaluated by <see cref="AsyncResponseStartupValidator"/> |
| | | 99 | | /// at host start: every <see cref="IDurableFlowExecutionObserver"/> must be a singleton, because the |
| | | 100 | | /// singleton flow executor resolves observers once from the root provider and holds them for its |
| | | 101 | | /// lifetime. Failing here names the offending registration and the fix; without it the first flow |
| | | 102 | | /// job dies inside the transport's retry loop with an opaque "Cannot resolve scoped service ... |
| | | 103 | | /// from root provider" (ValidateOnBuild does not descend into the executor's factory registration). |
| | | 104 | | /// Holds the service collection only until the check runs, then releases it. |
| | | 105 | | /// </summary> |
| | | 106 | | internal sealed class DurableFlowObserverLifetimeAudit(IServiceCollection services) |
| | | 107 | | { |
| | | 108 | | private IServiceCollection? _services = services; |
| | | 109 | | |
| | | 110 | | public void Validate() |
| | | 111 | | { |
| | | 112 | | var services = Interlocked.Exchange(ref _services, null); |
| | | 113 | | if (services is null) |
| | | 114 | | return; |
| | | 115 | | |
| | | 116 | | var nonSingleton = services.FirstOrDefault(descriptor => |
| | | 117 | | descriptor.ServiceType == typeof(IDurableFlowExecutionObserver) |
| | | 118 | | && descriptor.Lifetime != ServiceLifetime.Singleton); |
| | | 119 | | if (nonSingleton is not null) |
| | | 120 | | { |
| | | 121 | | throw new InvalidOperationException( |
| | | 122 | | $"{nameof(IDurableFlowExecutionObserver)} '{(nonSingleton.ImplementationType ?? nonSingleton.Implementat |
| | | 123 | | $"is registered as {nonSingleton.Lifetime}, but observers are held by the singleton flow executor for it |
| | | 124 | | "singletons (services.AddSingleton<IDurableFlowExecutionObserver, ...>()). An observer that needs scoped |
| | | 125 | | "create its own scope inside the callback."); |
| | | 126 | | } |
| | | 127 | | } |
| | | 128 | | } |
| | | 129 | | |
| | | 130 | | /// <summary> |
| | | 131 | | /// The runtime half of "all AsyncResponse packages are one version". The channel, transport, |
| | | 132 | | /// durable-flow store, and Testing packages are not ordinary consumers of Core: Core (and |
| | | 133 | | /// Abstractions) grant them <c>InternalsVisibleTo</c>, and they call internal types that carry no |
| | | 134 | | /// compatibility promise between releases. NuGet sees only <c>Core >= x</c>, so bumping one |
| | | 135 | | /// package — or a transitive dependency dragging Core forward — yields an install that restores, |
| | | 136 | | /// builds, and starts, and then throws <see cref="MissingMethodException"/> or |
| | | 137 | | /// <see cref="TypeLoadException"/> at the first call into a changed internal: usually inside a |
| | | 138 | | /// background consume loop, long after startup, where it reads as a broker fault. Evaluated by |
| | | 139 | | /// <see cref="AsyncResponseStartupValidator"/> before anything else, so the mismatch fails the |
| | | 140 | | /// host start with both assemblies named instead. |
| | | 141 | | /// <para> |
| | | 142 | | /// The version compared is <see cref="AssemblyInformationalVersionAttribute"/> without its |
| | | 143 | | /// <c>+build-metadata</c> suffix — the package version. <c>AssemblyVersion</c> cannot tell |
| | | 144 | | /// <c>1.2.0-rc.1</c> from <c>1.2.0-rc.2</c>, and the metadata suffix is the source-control |
| | | 145 | | /// revision, which legitimately differs between assemblies of one incremental local build. Only |
| | | 146 | | /// plain attribute and name reads: nothing here needs trimming annotations. |
| | | 147 | | /// </para> |
| | | 148 | | /// </summary> |
| | | 149 | | internal static class AsyncResponsePackageVersions |
| | | 150 | | { |
| | | 151 | | private const string Core = "AsyncResponse.Core"; |
| | | 152 | | |
| | | 153 | | /// <summary>A loaded package assembly: its simple name and its package version.</summary> |
| | 75029 | 154 | | internal readonly record struct LoadedPackage(string Name, string Version); |
| | | 155 | | |
| | | 156 | | /// <summary> |
| | | 157 | | /// Whether <paramref name="assemblySimpleName"/> is one of the shipped packages. By family, |
| | | 158 | | /// not by the <c>AsyncResponse.</c> prefix: the application's own assemblies may share the |
| | | 159 | | /// prefix (this repository's tests and samples do) and version independently. |
| | | 160 | | /// </summary> |
| | | 161 | | internal static bool IsPackageAssembly(string assemblySimpleName) |
| | 511717 | 162 | | => assemblySimpleName is Core or "AsyncResponse.Abstractions" or "AsyncResponse.Testing" |
| | 511717 | 163 | | || assemblySimpleName.StartsWith("AsyncResponse.Channels.", StringComparison.Ordinal) |
| | 511717 | 164 | | || assemblySimpleName.StartsWith("AsyncResponse.Transports.", StringComparison.Ordinal) |
| | 511717 | 165 | | || assemblySimpleName.StartsWith("AsyncResponse.DurableFlows.", StringComparison.Ordinal); |
| | | 166 | | |
| | | 167 | | /// <summary> |
| | | 168 | | /// The package assemblies loaded right now. Provider assemblies are loaded by the time the |
| | | 169 | | /// host starts — their registration extension ran — so the ones that can fail are the ones |
| | | 170 | | /// seen. An assembly that merely borrows a family name is told apart by its strong-name |
| | | 171 | | /// token: every shipped package is signed with Core's key. |
| | | 172 | | /// </summary> |
| | | 173 | | internal static IReadOnlyList<LoadedPackage> Loaded() |
| | | 174 | | { |
| | 2462 | 175 | | var coreAssembly = typeof(AsyncResponsePackageVersions).Assembly; |
| | 2462 | 176 | | var coreToken = coreAssembly.GetName().GetPublicKeyToken() ?? []; |
| | | 177 | | |
| | 2462 | 178 | | var packages = new List<LoadedPackage>(); |
| | 1028358 | 179 | | foreach (var assembly in AppDomain.CurrentDomain.GetAssemblies()) |
| | | 180 | | { |
| | 511717 | 181 | | var name = assembly.GetName(); |
| | 511717 | 182 | | if (name.Name is not { } simpleName |
| | 511717 | 183 | | || !IsPackageAssembly(simpleName) |
| | 511717 | 184 | | || !coreToken.AsSpan().SequenceEqual(name.GetPublicKeyToken() ?? [])) |
| | | 185 | | { |
| | | 186 | | continue; |
| | | 187 | | } |
| | | 188 | | |
| | 64265 | 189 | | packages.Add(new LoadedPackage(simpleName, PackageVersion(assembly, name))); |
| | | 190 | | } |
| | | 191 | | |
| | 2462 | 192 | | return packages; |
| | | 193 | | } |
| | | 194 | | |
| | | 195 | | private static string PackageVersion(Assembly assembly, AssemblyName name) |
| | | 196 | | { |
| | 64265 | 197 | | var informational = assembly.GetCustomAttribute<AssemblyInformationalVersionAttribute>()?.InformationalVersion; |
| | 64265 | 198 | | if (string.IsNullOrEmpty(informational)) |
| | 0 | 199 | | return name.Version?.ToString() ?? string.Empty; |
| | | 200 | | |
| | 64265 | 201 | | var metadata = informational.IndexOf('+', StringComparison.Ordinal); |
| | 64265 | 202 | | return metadata < 0 ? informational : informational[..metadata]; |
| | | 203 | | } |
| | | 204 | | |
| | | 205 | | /// <summary> |
| | | 206 | | /// Throws when any of <paramref name="loaded"/> is a different version from Core. A set |
| | | 207 | | /// without Core (never the real one — this code IS Core) has nothing to compare against. |
| | | 208 | | /// </summary> |
| | | 209 | | internal static void EnsureSingleVersion(IEnumerable<LoadedPackage> loaded) |
| | | 210 | | { |
| | 2462 | 211 | | var packages = loaded as IReadOnlyCollection<LoadedPackage> ?? [.. loaded]; |
| | 2462 | 212 | | string? coreVersion = null; |
| | 19066 | 213 | | foreach (var package in packages) |
| | | 214 | | { |
| | 8302 | 215 | | if (package.Name == Core) |
| | | 216 | | { |
| | 2462 | 217 | | coreVersion = package.Version; |
| | 2462 | 218 | | break; |
| | | 219 | | } |
| | | 220 | | } |
| | | 221 | | |
| | 2462 | 222 | | if (coreVersion is null) |
| | 0 | 223 | | return; |
| | | 224 | | |
| | 133454 | 225 | | foreach (var package in packages) |
| | | 226 | | { |
| | 64265 | 227 | | if (string.Equals(package.Version, coreVersion, StringComparison.Ordinal)) |
| | | 228 | | continue; |
| | | 229 | | |
| | 0 | 230 | | throw new InvalidOperationException( |
| | 0 | 231 | | $"{package.Name} {package.Version} is loaded next to {Core} {coreVersion}. All AsyncResponse.* packages |
| | 0 | 232 | | "same version: the channel, transport, durable-flow store, and Testing packages bind to internal Core AP |
| | 0 | 233 | | "change between releases, so a mixed install fails with MissingMethodException or TypeLoadException at f |
| | 0 | 234 | | $"typically inside a background loop, long after startup. Reference every AsyncResponse.* package at {co |
| | 0 | 235 | | "(or move all of them to one newer version); a central <PackageVersion> per package, or one shared versi |
| | 0 | 236 | | "keeps them aligned."); |
| | | 237 | | } |
| | 2462 | 238 | | } |
| | | 239 | | } |
| | | 240 | | |
| | | 241 | | /// <summary> |
| | | 242 | | /// Validates at host startup that <c>AddAsyncResponse()</c> was paired with exactly one response |
| | | 243 | | /// channel, one worker transport, and one durable-flow state store. These are mandatory core |
| | | 244 | | /// choices; making each explicit keeps the fluent registration complete and prevents silently |
| | | 245 | | /// unusable services from reaching production. |
| | | 246 | | /// </summary> |
| | | 247 | | internal sealed class AsyncResponseStartupValidator( |
| | | 248 | | IEnumerable<AsyncResponseChannelMarker> _channels, |
| | | 249 | | IEnumerable<AsyncResponseTransportMarker> _transports, |
| | | 250 | | IEnumerable<AsyncResponseDurableFlowStoreMarker> _flowStores, |
| | | 251 | | IOptions<AsyncResponseOptions> _options, |
| | | 252 | | IEnumerable<DurableFlowOptions>? _flowOptions = null, |
| | | 253 | | ILogger<AsyncResponseStartupValidator>? _logger = null, |
| | | 254 | | DurableFlowObserverLifetimeAudit? _observerAudit = null, |
| | | 255 | | IServiceProvider? _serviceProvider = null) : IHostedService |
| | | 256 | | { |
| | | 257 | | /// <summary>Starts this service.</summary> |
| | | 258 | | public Task StartAsync(CancellationToken cancellationToken) |
| | | 259 | | { |
| | | 260 | | // First: every check below may already run provider code bound to Core internals. |
| | | 261 | | AsyncResponsePackageVersions.EnsureSingleVersion(AsyncResponsePackageVersions.Loaded()); |
| | | 262 | | |
| | | 263 | | ValidateWatchdogOptions(_options.Value.Watchdog); |
| | | 264 | | ValidateInboundMessageBudget(_options.Value); |
| | | 265 | | _observerAudit?.Validate(); |
| | | 266 | | |
| | | 267 | | var channelNames = _channels.Select(c => c.Name).Distinct(StringComparer.Ordinal).ToArray(); |
| | | 268 | | |
| | | 269 | | if (channelNames.Length == 0) |
| | | 270 | | throw new InvalidOperationException( |
| | | 271 | | "AsyncResponse has no response channel registered. After AddAsyncResponse(), call " + |
| | | 272 | | ".WithInMemoryChannel() (AsyncResponse.Core) or .WithRedisChannel() (AsyncResponse.Channels.Redis). " + |
| | | 273 | | "Without a channel, waiters can never receive a response."); |
| | | 274 | | |
| | | 275 | | if (channelNames.Length > 1) |
| | | 276 | | throw new InvalidOperationException( |
| | | 277 | | $"AsyncResponse has multiple response channels registered ({string.Join(", ", channelNames)}). " + |
| | | 278 | | "Register exactly one channel."); |
| | | 279 | | |
| | | 280 | | var transportNames = _transports.Select(t => t.Name).Distinct(StringComparer.Ordinal).ToArray(); |
| | | 281 | | |
| | | 282 | | if (transportNames.Length == 0) |
| | | 283 | | throw new InvalidOperationException( |
| | | 284 | | "AsyncResponse has no worker transport registered. After AddAsyncResponse(), call " + |
| | | 285 | | ".WithInMemoryTransport() (AsyncResponse.Core), .WithGooglePubSubTransport(...) " + |
| | | 286 | | "(AsyncResponse.Transports.GooglePubSub), or another full AsyncResponse transport package. " + |
| | | 287 | | "Without a transport, EnqueueWorkerAsync cannot dispatch worker jobs."); |
| | | 288 | | |
| | | 289 | | if (transportNames.Length > 1) |
| | | 290 | | throw new InvalidOperationException( |
| | | 291 | | $"AsyncResponse has multiple worker transports registered ({string.Join(", ", transportNames)}). " + |
| | | 292 | | "Register exactly one transport."); |
| | | 293 | | |
| | | 294 | | var flowStores = _flowStores.DistinctBy(store => store.StoreType).ToArray(); |
| | | 295 | | if (flowStores.Length == 0) |
| | | 296 | | { |
| | | 297 | | throw new InvalidOperationException( |
| | | 298 | | "AsyncResponse has no durable-flow state store registered. After AddAsyncResponse(), call " + |
| | | 299 | | ".WithInMemoryDurableFlows() (AsyncResponse.Core), a provider registration such as " + |
| | | 300 | | ".WithPostgreSqlDurableFlows(...), or .WithDurableFlows<TStore>() for an application-owned store."); |
| | | 301 | | } |
| | | 302 | | |
| | | 303 | | if (flowStores.Length > 1) |
| | | 304 | | { |
| | | 305 | | throw new InvalidOperationException( |
| | | 306 | | $"AsyncResponse has multiple durable-flow state stores registered ({string.Join(", ", flowStores.Select( |
| | | 307 | | "Register exactly one durable-flow store."); |
| | | 308 | | } |
| | | 309 | | |
| | | 310 | | foreach (var storeMarker in _flowStores) |
| | | 311 | | storeMarker.ValidateForwardLifetime(); |
| | | 312 | | |
| | | 313 | | ValidateSingleFlowOptions(); |
| | | 314 | | ValidateEarlyAckDeclarations(); |
| | | 315 | | ValidateAwaitedStepLedgerCoverage(); |
| | | 316 | | |
| | | 317 | | // Construct the flow store once, here: every provider store validates its options in its |
| | | 318 | | // constructor, but the store is otherwise resolved only inside per-execution scopes — so a |
| | | 319 | | // misconfigured table name (or any other store option) previously passed startup and first |
| | | 320 | | // threw inside the worker transport's retry loop, burning a real production run to the |
| | | 321 | | // delivery cap. Constructors do no I/O by convention; the scope disposes what it built. |
| | | 322 | | // (Null only in unit tests that construct the validator directly; DI always supplies it.) |
| | | 323 | | if (_serviceProvider is not null) |
| | | 324 | | { |
| | | 325 | | using var scope = _serviceProvider.CreateScope(); |
| | | 326 | | _ = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | | 327 | | } |
| | | 328 | | |
| | | 329 | | return Task.CompletedTask; |
| | | 330 | | } |
| | | 331 | | |
| | | 332 | | /// <summary> |
| | | 333 | | /// An awaited step with no explicit timeout and no <see cref="DurableFlowOptions.DefaultStepTimeout"/> |
| | | 334 | | /// waits out the CHANNEL's default timeout, so the ledger's idle TTL must out-live that window: |
| | | 335 | | /// with <see cref="DurableFlowOptions.StateExpiry"/> at or below it, the row (and the lease |
| | | 336 | | /// renewal anchored on it) is pruned mid-wait — the run becomes unrecoverable and no |
| | | 337 | | /// step-timeout fault ever fires. Channels declare their resolved default through the marker; |
| | | 338 | | /// a channel that declares nothing skips the check, and a configured |
| | | 339 | | /// <c>DefaultStepTimeout</c> makes the channel default unreachable, so the check does not apply. |
| | | 340 | | /// </summary> |
| | | 341 | | /// <summary> |
| | | 342 | | /// The engine resolves <see cref="DurableFlowOptions"/> through <c>GetRequiredService</c> — the |
| | | 343 | | /// LAST registration — while this validator enumerates them. A second <c>WithDurableFlows</c> |
| | | 344 | | /// call for the SAME store type (a provider helper followed by the generic overload to adjust |
| | | 345 | | /// a common setting) registers two forwards that the store-count check collapses into one, |
| | | 346 | | /// and the validator then judged the first while the engine ran on the second. Fail on the |
| | | 347 | | /// duplicate instead; the checks below read the last one, like the engine. |
| | | 348 | | /// </summary> |
| | | 349 | | private void ValidateSingleFlowOptions() |
| | | 350 | | { |
| | | 351 | | var registered = _flowOptions?.Distinct().ToArray() ?? []; |
| | | 352 | | if (registered.Length > 1) |
| | | 353 | | { |
| | | 354 | | throw new InvalidOperationException( |
| | | 355 | | $"{nameof(DurableFlowOptions)} is registered {registered.Length} times — WithDurableFlows was called mor |
| | | 356 | | "(e.g. a provider registration such as .WithPostgreSqlDurableFlows(...) followed by .WithDurableFlows<TS |
| | | 357 | | "The flow engine consumes only the last registration, so settings from the earlier call are silently ign |
| | | 358 | | "Configure every durable-flow setting in the single registration's callback."); |
| | | 359 | | } |
| | | 360 | | } |
| | | 361 | | |
| | | 362 | | private void ValidateAwaitedStepLedgerCoverage() |
| | | 363 | | { |
| | | 364 | | var flowOptions = _flowOptions?.LastOrDefault(); |
| | | 365 | | if (flowOptions is null || flowOptions.DefaultStepTimeout is not null) |
| | | 366 | | return; |
| | | 367 | | |
| | | 368 | | foreach (var channel in _channels) |
| | | 369 | | { |
| | | 370 | | if (channel.EffectiveDefaultWaitTimeout is not { } effectiveDefault |
| | | 371 | | || flowOptions.StateExpiry > effectiveDefault) |
| | | 372 | | continue; |
| | | 373 | | |
| | | 374 | | throw new InvalidOperationException( |
| | | 375 | | $"{nameof(DurableFlowOptions)}.{nameof(DurableFlowOptions.StateExpiry)} ({flowOptions.StateExpiry}) does |
| | | 376 | | $"{channel.Name} channel's effective default waiter timeout ({effectiveDefault} — DefaultTimeout, or Rec |
| | | 377 | | "when DefaultTimeout is null). An awaited step without an explicit timeout waits out that default, and a |
| | | 378 | | "does not out-live the wait is pruned mid-wait: the run becomes unrecoverable and no step-timeout fault |
| | | 379 | | $"Raise StateExpiry above the channel default, shorten the channel default, or set " + |
| | | 380 | | $"{nameof(DurableFlowOptions)}.{nameof(DurableFlowOptions.DefaultStepTimeout)}."); |
| | | 381 | | } |
| | | 382 | | } |
| | | 383 | | |
| | | 384 | | /// <summary> |
| | | 385 | | /// Fails fast when a transport's worker subscriber is configured for early ACK: durable-flow |
| | | 386 | | /// wake-ups ride the worker queue and rely on broker redelivery for crash recovery (the |
| | | 387 | | /// executor's lease poll deliberately delegates dead-holder liveness to redelivery of the |
| | | 388 | | /// holder's own job). With early ACK, a crash between the ACK and the handler strands the run |
| | | 389 | | /// as Running — no lease, no queued job, and no store enumeration API to even discover it. |
| | | 390 | | /// A flow store is always registered (validated above), so the veto is unconditional unless |
| | | 391 | | /// the operator accepts the risk via <see cref="DurableFlowOptions.AllowEarlyAckWorkerSubscriber"/>. |
| | | 392 | | /// Early ACK on the response queue is at-most-once response delivery — a crash after the ACK |
| | | 393 | | /// destroys the broker's only copy, the waiter burns its full timeout and fails, and a durable |
| | | 394 | | /// flow then restarts the timed-out step fresh (re-sending its request; triggers must be |
| | | 395 | | /// idempotent, which the recovery contract already requires). Nothing strands, so it warns |
| | | 396 | | /// instead of throwing. |
| | | 397 | | /// </summary> |
| | | 398 | | private void ValidateEarlyAckDeclarations() |
| | | 399 | | { |
| | | 400 | | var flowOptions = _flowOptions?.LastOrDefault(); |
| | | 401 | | foreach (var transport in _transports) |
| | | 402 | | { |
| | | 403 | | if (transport.ResponseSubscriberUsesEarlyAck) |
| | | 404 | | { |
| | | 405 | | _logger?.LogWarning( |
| | | 406 | | "The {Transport} response subscriber uses early ACK ({AckModePath} = AckAfterEnqueue): a crash after |
| | | 407 | | transport.Name, |
| | | 408 | | transport.ResponseAckModePath); |
| | | 409 | | } |
| | | 410 | | |
| | | 411 | | if (!transport.WorkerSubscriberUsesEarlyAck) |
| | | 412 | | continue; |
| | | 413 | | |
| | | 414 | | if (flowOptions?.AllowEarlyAckWorkerSubscriber == true) |
| | | 415 | | { |
| | | 416 | | _logger?.LogWarning( |
| | | 417 | | "The {Transport} worker subscriber uses early ACK with {OptOut} enabled: a crash after an ACK but be |
| | | 418 | | transport.Name, |
| | | 419 | | $"{nameof(DurableFlowOptions)}.{nameof(DurableFlowOptions.AllowEarlyAckWorkerSubscriber)}"); |
| | | 420 | | continue; |
| | | 421 | | } |
| | | 422 | | |
| | | 423 | | throw new InvalidOperationException( |
| | | 424 | | $"The {transport.Name} worker subscriber is configured for early ACK ({transport.WorkerAckModePath} = Ac |
| | | 425 | | "Flow execution relies on broker redelivery for crash recovery: a process crash after an early ACK but b |
| | | 426 | | $"Keep the worker subscriber on AckAfterHandlerCompletes (the default), or set {nameof(DurableFlowOption |
| | | 427 | | "(see docs/durable-flows.md and docs/transport-semantics.md)."); |
| | | 428 | | } |
| | | 429 | | } |
| | | 430 | | |
| | | 431 | | /// <summary> |
| | | 432 | | /// Fails fast on watchdog misconfiguration: a non-positive interval spins the scan loop, a |
| | | 433 | | /// non-positive stale threshold flags every entry, and an out-of-range delay (negative, or |
| | | 434 | | /// beyond the timer ceiling) throws deep inside <see cref="Task.Delay(TimeSpan)"/> long after |
| | | 435 | | /// registration. <c>StaleAfter</c> is only ever compared against entry age, so it needs no |
| | | 436 | | /// ceiling; <c>Interval</c> and <c>StartupDelay</c> arm timers, and zero is a valid startup |
| | | 437 | | /// delay ("scan immediately"). |
| | | 438 | | /// </summary> |
| | | 439 | | /// <summary> |
| | | 440 | | /// Rejects an inbound size budget that would silently swallow traffic. The guard acknowledges |
| | | 441 | | /// oversized messages without dispatch, so a zero or negative limit does not fail loudly — it |
| | | 442 | | /// quietly drops EVERY non-empty message on both the response and worker routes and reports |
| | | 443 | | /// success. That is total data loss wearing the shape of a healthy service, and a plausible |
| | | 444 | | /// configuration typo (`0` read as "no limit"), so it has to be caught at startup where a |
| | | 445 | | /// misconfiguration is still visible. |
| | | 446 | | /// </summary> |
| | | 447 | | private static void ValidateInboundMessageBudget(AsyncResponseOptions options) |
| | | 448 | | { |
| | | 449 | | if (options.MaxInboundMessageChars is { } limit && limit <= 0) |
| | | 450 | | { |
| | | 451 | | throw new InvalidOperationException( |
| | | 452 | | $"{nameof(AsyncResponseOptions)}.{nameof(AsyncResponseOptions.MaxInboundMessageChars)} must be positive |
| | | 453 | | $"(got {limit}); use null to remove the limit. A non-positive budget acknowledges every inbound message |
| | | 454 | | "without dispatching it."); |
| | | 455 | | } |
| | | 456 | | } |
| | | 457 | | |
| | | 458 | | private static void ValidateWatchdogOptions(AsyncResponseWatchdogOptions watchdog) |
| | | 459 | | { |
| | | 460 | | const string optionsPath = $"{nameof(AsyncResponseOptions)}.{nameof(AsyncResponseOptions.Watchdog)}"; |
| | | 461 | | AsyncResponseChannelOptions.EnsureTimerBacked(watchdog.Interval, optionsPath, nameof(AsyncResponseWatchdogOption |
| | | 462 | | if (watchdog.StaleAfter <= TimeSpan.Zero) |
| | | 463 | | throw new InvalidOperationException( |
| | | 464 | | $"{optionsPath}.{nameof(AsyncResponseWatchdogOptions.StaleAfter)} must be positive."); |
| | | 465 | | AsyncResponseChannelOptions.EnsureTimerBackedAllowZero(watchdog.StartupDelay, optionsPath, nameof(AsyncResponseW |
| | | 466 | | if (watchdog.MaxScanEntries <= 0) |
| | | 467 | | throw new InvalidOperationException( |
| | | 468 | | $"{nameof(AsyncResponseOptions)}.{nameof(AsyncResponseOptions.Watchdog)}.{nameof(AsyncResponseWatchdogOp |
| | | 469 | | } |
| | | 470 | | |
| | | 471 | | /// <summary>Stops this service.</summary> |
| | | 472 | | public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; |
| | | 473 | | } |