< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.AsyncResponseTransportMarker
Assembly: AsyncResponse.Core
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/AsyncResponse.Core/AsyncResponseStartupValidator.cs
Line coverage
100%
Covered lines: 2
Uncovered lines: 0
Coverable lines: 2
Total lines: 187
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/AsyncResponse.Core/AsyncResponseStartupValidator.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4
 5namespace AsyncResponse;
 6
 7/// <summary>
 8/// Internal marker registered by each response-channel registration
 9/// (<c>.WithInMemoryChannel()</c> / <c>.WithRedisChannel()</c>). The
 10/// <see cref="AsyncResponseStartupValidator"/> asserts exactly one channel is present.
 11/// </summary>
 12internal sealed class AsyncResponseChannelMarker(string name)
 13{
 14    public string Name { get; } = name;
 15}
 16
 17/// <summary>
 18/// Internal marker registered by each worker-transport registration
 19/// (<c>.WithInMemoryTransport()</c> / <c>.WithGooglePubSubTransport(...)</c>). The
 20/// <see cref="AsyncResponseStartupValidator"/> asserts exactly one transport is present.
 21/// </summary>
 322internal sealed class AsyncResponseTransportMarker(string name)
 23{
 324    public string Name { get; } = name;
 25
 26    /// <summary>
 27    /// Whether the transport's worker subscriber resolved to early ACK (<c>AckAfterEnqueue</c>).
 28    /// Declared by the transport's registration from its bound options so the startup validator
 29    /// can veto the combination with durable flows without referencing transport types.
 30    /// </summary>
 31    public bool WorkerSubscriberUsesEarlyAck { get; init; }
 32
 33    /// <summary>Worker ack-mode option path, shown in the startup error.</summary>
 34    public string? WorkerAckModePath { get; init; }
 35
 36    /// <summary>Whether the transport's response subscriber resolved to early ACK.</summary>
 37    public bool ResponseSubscriberUsesEarlyAck { get; init; }
 38
 39    /// <summary>Response ack-mode option path, shown in the startup warning.</summary>
 40    public string? ResponseAckModePath { get; init; }
 41}
 42
 43/// <summary>Internal marker registered by each durable-flow state-store registration.</summary>
 44internal sealed class AsyncResponseDurableFlowStoreMarker(Type storeType)
 45{
 46    public Type StoreType { get; } = storeType;
 47    public string Name { get; } = storeType.FullName ?? storeType.Name;
 48}
 49
 50/// <summary>
 51/// Validates at host startup that <c>AddAsyncResponse()</c> was paired with exactly one response
 52/// channel, one worker transport, and one durable-flow state store. These are mandatory core
 53/// choices; making each explicit keeps the fluent registration complete and prevents silently
 54/// unusable services from reaching production.
 55/// </summary>
 56internal sealed class AsyncResponseStartupValidator(
 57    IEnumerable<AsyncResponseChannelMarker> _channels,
 58    IEnumerable<AsyncResponseTransportMarker> _transports,
 59    IEnumerable<AsyncResponseDurableFlowStoreMarker> _flowStores,
 60    IOptions<AsyncResponseOptions> _options,
 61    IEnumerable<DurableFlowOptions>? _flowOptions = null,
 62    ILogger<AsyncResponseStartupValidator>? _logger = null) : IHostedService
 63{
 64    /// <summary>Starts this service.</summary>
 65    public Task StartAsync(CancellationToken cancellationToken)
 66    {
 67        ValidateWatchdogOptions(_options.Value.Watchdog);
 68
 69        var channelNames = _channels.Select(c => c.Name).Distinct(StringComparer.Ordinal).ToArray();
 70
 71        if (channelNames.Length == 0)
 72            throw new InvalidOperationException(
 73                "AsyncResponse has no response channel registered. After AddAsyncResponse(), call " +
 74                ".WithInMemoryChannel() (AsyncResponse.Core) or .WithRedisChannel() (AsyncResponse.Channels.Redis). " +
 75                "Without a channel, waiters can never receive a response.");
 76
 77        if (channelNames.Length > 1)
 78            throw new InvalidOperationException(
 79                $"AsyncResponse has multiple response channels registered ({string.Join(", ", channelNames)}). " +
 80                "Register exactly one channel.");
 81
 82        var transportNames = _transports.Select(t => t.Name).Distinct(StringComparer.Ordinal).ToArray();
 83
 84        if (transportNames.Length == 0)
 85            throw new InvalidOperationException(
 86                "AsyncResponse has no worker transport registered. After AddAsyncResponse(), call " +
 87                ".WithInMemoryTransport() (AsyncResponse.Core), .WithGooglePubSubTransport(...) " +
 88                "(AsyncResponse.Transports.GooglePubSub), or another full AsyncResponse transport package. " +
 89                "Without a transport, EnqueueWorkerAsync cannot dispatch worker jobs.");
 90
 91        if (transportNames.Length > 1)
 92            throw new InvalidOperationException(
 93                $"AsyncResponse has multiple worker transports registered ({string.Join(", ", transportNames)}). " +
 94                "Register exactly one transport.");
 95
 96        var flowStores = _flowStores.DistinctBy(store => store.StoreType).ToArray();
 97        if (flowStores.Length == 0)
 98        {
 99            throw new InvalidOperationException(
 100                "AsyncResponse has no durable-flow state store registered. After AddAsyncResponse(), call " +
 101                ".WithInMemoryDurableFlows() (AsyncResponse.Core), a provider registration such as " +
 102                ".WithPostgreSqlDurableFlows(...), or .WithDurableFlows<TStore>() for an application-owned store.");
 103        }
 104
 105        if (flowStores.Length > 1)
 106        {
 107            throw new InvalidOperationException(
 108                $"AsyncResponse has multiple durable-flow state stores registered ({string.Join(", ", flowStores.Select(
 109                "Register exactly one durable-flow store.");
 110        }
 111
 112        ValidateEarlyAckDeclarations();
 113
 114        return Task.CompletedTask;
 115    }
 116
 117    /// <summary>
 118    /// Fails fast when a transport's worker subscriber is configured for early ACK: durable-flow
 119    /// wake-ups ride the worker queue and rely on broker redelivery for crash recovery (the
 120    /// executor's lease poll deliberately delegates dead-holder liveness to redelivery of the
 121    /// holder's own job). With early ACK, a crash between the ACK and the handler strands the run
 122    /// as Running — no lease, no queued job, and no store enumeration API to even discover it.
 123    /// A flow store is always registered (validated above), so the veto is unconditional unless
 124    /// the operator accepts the risk via <see cref="DurableFlowOptions.AllowEarlyAckWorkerSubscriber"/>.
 125    /// Early ACK on the response queue is at-most-once response delivery — a crash after the ACK
 126    /// destroys the broker's only copy, the waiter burns its full timeout and fails, and a durable
 127    /// flow then restarts the timed-out step fresh (re-sending its request; triggers must be
 128    /// idempotent, which the recovery contract already requires). Nothing strands, so it warns
 129    /// instead of throwing.
 130    /// </summary>
 131    private void ValidateEarlyAckDeclarations()
 132    {
 133        var flowOptions = _flowOptions?.FirstOrDefault();
 134        foreach (var transport in _transports)
 135        {
 136            if (transport.ResponseSubscriberUsesEarlyAck)
 137            {
 138                _logger?.LogWarning(
 139                    "The {Transport} response subscriber uses early ACK ({AckModePath} = AckAfterEnqueue): a crash after
 140                    transport.Name,
 141                    transport.ResponseAckModePath);
 142            }
 143
 144            if (!transport.WorkerSubscriberUsesEarlyAck)
 145                continue;
 146
 147            if (flowOptions?.AllowEarlyAckWorkerSubscriber == true)
 148            {
 149                _logger?.LogWarning(
 150                    "The {Transport} worker subscriber uses early ACK with {OptOut} enabled: a crash after an ACK but be
 151                    transport.Name,
 152                    $"{nameof(DurableFlowOptions)}.{nameof(DurableFlowOptions.AllowEarlyAckWorkerSubscriber)}");
 153                continue;
 154            }
 155
 156            throw new InvalidOperationException(
 157                $"The {transport.Name} worker subscriber is configured for early ACK ({transport.WorkerAckModePath} = Ac
 158                "Flow execution relies on broker redelivery for crash recovery: a process crash after an early ACK but b
 159                $"Keep the worker subscriber on AckAfterHandlerCompletes (the default), or set {nameof(DurableFlowOption
 160                "(see docs/durable-flows.md and docs/transport-semantics.md).");
 161        }
 162    }
 163
 164    /// <summary>
 165    /// Fails fast on watchdog misconfiguration: a non-positive interval spins the scan loop, a
 166    /// non-positive stale threshold flags every entry, and a negative startup delay throws deep
 167    /// inside <see cref="Task.Delay(TimeSpan)"/> long after registration.
 168    /// </summary>
 169    private static void ValidateWatchdogOptions(AsyncResponseWatchdogOptions watchdog)
 170    {
 171        if (watchdog.Interval <= TimeSpan.Zero)
 172            throw new InvalidOperationException(
 173                $"{nameof(AsyncResponseOptions)}.{nameof(AsyncResponseOptions.Watchdog)}.{nameof(AsyncResponseWatchdogOp
 174        if (watchdog.StaleAfter <= TimeSpan.Zero)
 175            throw new InvalidOperationException(
 176                $"{nameof(AsyncResponseOptions)}.{nameof(AsyncResponseOptions.Watchdog)}.{nameof(AsyncResponseWatchdogOp
 177        if (watchdog.StartupDelay < TimeSpan.Zero)
 178            throw new InvalidOperationException(
 179                $"{nameof(AsyncResponseOptions)}.{nameof(AsyncResponseOptions.Watchdog)}.{nameof(AsyncResponseWatchdogOp
 180        if (watchdog.MaxScanEntries <= 0)
 181            throw new InvalidOperationException(
 182                $"{nameof(AsyncResponseOptions)}.{nameof(AsyncResponseOptions.Watchdog)}.{nameof(AsyncResponseWatchdogOp
 183    }
 184
 185    /// <summary>Stops this service.</summary>
 186    public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
 187}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(string)