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

Information
Class: AsyncResponse.LostSubscriberCallbackDispatcher
Assembly: AsyncResponse.Core
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/AsyncResponse.Core/LostSubscriberCallbackDispatcher.cs
Line coverage
98%
Covered lines: 160
Uncovered lines: 2
Coverable lines: 162
Total lines: 376
Line coverage: 98.7%
Branch coverage
90%
Covered branches: 74
Total branches: 82
Branch coverage: 90.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
DispatchLostResponses()86.36%222296.77%
DispatchLostExceptions()78.57%141495.45%
DispatchLostResponse()95.83%2424100%
DispatchLostException()94.44%1818100%
DispatchToFailureCallback()100%44100%
<DispatchToFailureCallback()100%11100%
InvokeAsync()100%11100%

File(s)

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

#LineLine coverage
 1using Microsoft.Extensions.DependencyInjection;
 2using Microsoft.Extensions.Logging;
 3using System.Diagnostics;
 4using System.Runtime.ExceptionServices;
 5using System.Text.Json;
 6
 7namespace AsyncResponse;
 8
 9/// <summary>
 10/// Result of a lost-subscriber dispatch attempt.
 11/// </summary>
 12/// <param name="ShouldResume">
 13/// The recovery route the payload reported (<see cref="IAsyncResponsePayload.ShouldResumeOnRecovery"/>):
 14/// <c>true</c> resume, <c>false</c> fail, or <c>null</c> when it could not be classified (no recovery
 15/// state, missing payload type, null payload, conversion failure) — treated as "do not resume".
 16/// </param>
 17/// <param name="CallbackInvoked">
 18/// <c>true</c> when a callback was invoked successfully — the recovery state is consumed and the
 19/// caller should delete it.
 20/// </param>
 21internal readonly record struct LostSubscriberDispatchResult(bool? ShouldResume, bool CallbackInvoked)
 22{
 23    /// <summary>
 24    /// <c>true</c> when a live subscriber re-appeared between the caller's empty snapshot and the
 25    /// recovery-state read. No callback was invoked and no state was consumed — the caller should
 26    /// re-snapshot and dispatch live.
 27    /// </summary>
 28    public bool RetryLive { get; init; }
 29}
 30
 31/// <summary>
 32/// The single decision point of the lost-subscriber fallback: when an async response is published
 33/// and no subscriber is listening (the original waiter died, e.g. with a redeploy/restart), this
 34/// dispatcher chooses and invokes the callback persisted in the <see cref="RecoveryState"/>.
 35/// <para>
 36/// For payload envelopes (<c>SetResponse</c>) the payload's
 37/// <see cref="IAsyncResponsePayload.ShouldResumeOnRecovery"/> decides the route: <c>true</c> goes to
 38/// the resume callback; <c>false</c> (and any unclassifiable payload, conservatively) goes to the
 39/// failure callback wrapped in an <see cref="AsyncResponseDomainFailureException"/>. For exception
 40/// envelopes (<c>SetException</c>) the failure callback is always used.
 41/// </para>
 42/// <para>
 43/// The publisher stays a plain transport: it only reports "published, but nobody was listening" and
 44/// hands over to this dispatcher. This decision is independent of the live waiter's <c>Until</c>
 45/// predicate, which no longer exists once the waiter is lost.
 46/// </para>
 47/// </summary>
 348internal sealed class LostSubscriberCallbackDispatcher(
 349    IServiceScopeFactory _scopeFactory,
 350    AsyncResponseContextPropagation _propagation,
 351    ILogger _logger)
 52{
 53    /// <summary>
 54    /// Loads every recovery registration for <paramref name="correlationId"/> and dispatches a lost
 55    /// response to each registration's resume/failure callback.
 56    /// </summary>
 57    public async Task<LostSubscriberDispatchResult> DispatchLostResponses<T>(
 58        IRecoveryStateStore recoveryStateStore,
 59        string correlationId,
 60        T response,
 61        string channel,
 62        CancellationToken cancellationToken,
 63        Func<ValueTask<bool>>? hasLiveSubscriber = null)
 64    {
 365        var recoveryStates = await recoveryStateStore.GetAllAsync(correlationId, cancellationToken).ConfigureAwait(false
 66
 67        // A waiter registers its subscription before saving its recovery state, so the snapshot
 68        // race has two shapes — and the re-check must run before the empty-state early return:
 69        // a recovery state visible here implies its subscription is visible too, and, inversely, a
 70        // freshly registered subscription may not have saved its state yet, in which case
 71        // recoveryStates is empty precisely because the waiter is about to go live. Either way a
 72        // live subscriber means the "nobody listening" premise was stale — hand the response back
 73        // for live delivery instead of consuming registrations or dropping it unrecoverably.
 374        if (hasLiveSubscriber is not null && await hasLiveSubscriber().ConfigureAwait(false))
 375            return new LostSubscriberDispatchResult(null, false) { RetryLive = true };
 76
 377        if (recoveryStates.Count == 0)
 378            return await DispatchLostResponse(null, response, channel).ConfigureAwait(false);
 79
 380        var callbackInvoked = false;
 381        bool? shouldResume = null;
 382        var routeSet = false;
 383        var routeMixed = false;
 384        ExceptionDispatchInfo? firstException = null;
 85
 386        foreach (var recoveryState in recoveryStates)
 87        {
 88            try
 89            {
 390                var result = await DispatchLostResponse(recoveryState, response, channel).ConfigureAwait(false);
 391                if (!routeSet)
 92                {
 393                    shouldResume = result.ShouldResume;
 394                    routeSet = true;
 95                }
 396                else if (shouldResume != result.ShouldResume)
 97                {
 298                    routeMixed = true;
 99                }
 100
 3101                if (!result.CallbackInvoked)
 3102                    continue;
 103
 3104                callbackInvoked = true;
 3105                await recoveryStateStore.TryDeleteAsync(correlationId, recoveryState.RegistrationId, cancellationToken).
 3106            }
 3107            catch (Exception ex)
 108            {
 3109                if (ex is OperationCanceledException && cancellationToken.IsCancellationRequested)
 0110                    throw;
 111
 112                // Capture rather than re-throw a bare variable so the original throw site's stack
 113                // trace survives the dispatch to the remaining registrations.
 2114                firstException ??= ExceptionDispatchInfo.Capture(ex);
 2115            }
 3116        }
 117
 3118        firstException?.Throw();
 119
 3120        return new LostSubscriberDispatchResult(routeMixed ? null : shouldResume, callbackInvoked);
 3121    }
 122
 123    /// <summary>
 124    /// Loads every recovery registration for <paramref name="correlationId"/> and dispatches a lost
 125    /// exception to each registration's failure callback.
 126    /// </summary>
 127    public async Task<LostSubscriberDispatchResult> DispatchLostExceptions(
 128        IRecoveryStateStore recoveryStateStore,
 129        string correlationId,
 130        Exception exception,
 131        string channel,
 132        CancellationToken cancellationToken,
 133        Func<ValueTask<bool>>? hasLiveSubscriber = null)
 134    {
 3135        var recoveryStates = await recoveryStateStore.GetAllAsync(correlationId, cancellationToken).ConfigureAwait(false
 136
 137        // Same snapshot-race re-check as DispatchLostResponses, and for the same reason it must
 138        // precede the empty-state early return: an empty snapshot may mean the waiter registered
 139        // its subscription but has not saved its recovery state yet. A live subscriber means the
 140        // exception should be delivered live instead of consumed here.
 3141        if (hasLiveSubscriber is not null && await hasLiveSubscriber().ConfigureAwait(false))
 3142            return new LostSubscriberDispatchResult(false, false) { RetryLive = true };
 143
 3144        if (recoveryStates.Count == 0)
 3145            return new LostSubscriberDispatchResult(false, await DispatchLostException(null, exception, channel).Configu
 146
 3147        var callbackInvoked = false;
 3148        ExceptionDispatchInfo? firstException = null;
 149
 3150        foreach (var recoveryState in recoveryStates)
 151        {
 152            try
 153            {
 3154                if (!await DispatchLostException(recoveryState, exception, channel).ConfigureAwait(false))
 3155                    continue;
 156
 3157                callbackInvoked = true;
 3158                await recoveryStateStore.TryDeleteAsync(correlationId, recoveryState.RegistrationId, cancellationToken).
 3159            }
 3160            catch (Exception ex)
 161            {
 3162                if (ex is OperationCanceledException && cancellationToken.IsCancellationRequested)
 0163                    throw;
 164
 165                // Capture rather than re-throw a bare variable so the original throw site's stack
 166                // trace survives the dispatch to the remaining registrations.
 2167                firstException ??= ExceptionDispatchInfo.Capture(ex);
 2168            }
 3169        }
 170
 3171        firstException?.Throw();
 172
 173        // Exception envelopes always take the failure route, so ShouldResume is fixed at false.
 3174        return new LostSubscriberDispatchResult(false, callbackInvoked);
 3175    }
 176
 177    /// <summary>Dispatches a successfully published payload that no subscriber received.</summary>
 178    public async Task<LostSubscriberDispatchResult> DispatchLostResponse<T>(RecoveryState? recoveryState, T response, st
 179    {
 3180        using var activity = AsyncResponseDiagnostics.StartActivity(
 3181            "asyncresponse.lost_subscriber.dispatch",
 3182            correlationId: recoveryState?.CorrelationId);
 3183        activity?.SetTag("asyncresponse.lost_subscriber.kind", "response");
 3184        activity?.SetTag("asyncresponse.channel_name", channel);
 3185        if (response is not null)
 3186            AsyncResponseDiagnostics.SetPayloadType(activity, response.GetType());
 187
 188        try
 189        {
 190            // The recovering process has no live Until predicate — the payload itself decides whether
 191            // this late response resumes the flow or fails it. A null (unclassifiable) verdict is
 192            // treated conservatively as "do not resume", so a payload that cannot be understood never
 193            // takes the happy path.
 3194            var shouldResume = recoveryState is null
 3195                ? (bool?)null
 3196                : PayloadRecoveryClassifier.ShouldResume(response, recoveryState.PayloadTypeFullName);
 3197            AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, shouldResume);
 198
 3199            if (shouldResume != true)
 200            {
 3201                if (recoveryState is null)
 202                {
 3203                    _logger.LogWarning("No subscribers and no recovery state for channel {Channel}.", channel);
 3204                    activity?.SetTag("asyncresponse.recovery.callback_invoked", false);
 3205                    return new LostSubscriberDispatchResult(shouldResume, false);
 206                }
 207
 3208                var invoked = await DispatchToFailureCallback(recoveryState, response, channel, activity).ConfigureAwait
 3209                activity?.SetTag("asyncresponse.recovery.callback_invoked", invoked);
 3210                return new LostSubscriberDispatchResult(shouldResume, invoked);
 211            }
 212
 213            // shouldResume == true implies recoveryState is non-null (the verdict is null otherwise).
 3214            if (recoveryState!.ResumeCallback == null)
 215            {
 3216                _logger.LogWarning("No subscribers for channel {Channel}; no resume callback available.", channel);
 2217                activity?.SetTag("asyncresponse.recovery.callback_invoked", false);
 2218                return new LostSubscriberDispatchResult(shouldResume, false);
 219            }
 220
 3221            _logger.LogWarning("No subscribers for channel {Channel}; invoking resume callback.", channel);
 222
 3223            var invocation = ReflectionExtensions.ResolveCallback(
 3224                recoveryState.ResumeCallback,
 3225                payload: response,
 3226                exception: null,
 3227                correlationId: recoveryState.CorrelationId
 3228            );
 229
 230            // Deliberately not swallowed: a failing resume propagates to the publisher's caller,
 231            // which can escalate it through SetException to the failure callback (the ingress does
 232            // exactly that). The catch below only marks the activity before rethrowing.
 3233            await InvokeAsync(invocation, recoveryState.Context).ConfigureAwait(false);
 234
 3235            _logger.LogInformation("Resume callback invoked for channel {Channel}.", channel);
 3236            activity?.SetTag("asyncresponse.recovery.callback_invoked", true);
 237
 3238            return new LostSubscriberDispatchResult(shouldResume, true);
 239        }
 3240        catch (Exception ex)
 241        {
 3242            AsyncResponseDiagnostics.SetError(activity, ex);
 2243            throw;
 244        }
 3245    }
 246
 247    /// <summary>Dispatches an exception envelope that no subscriber received.</summary>
 248    public async Task<bool> DispatchLostException(RecoveryState? recoveryState, Exception exception, string channel)
 249    {
 3250        using var activity = AsyncResponseDiagnostics.StartActivity(
 3251            "asyncresponse.lost_subscriber.dispatch",
 3252            correlationId: recoveryState?.CorrelationId);
 3253        activity?.SetTag("asyncresponse.lost_subscriber.kind", "exception");
 3254        activity?.SetTag("asyncresponse.channel_name", channel);
 3255        activity?.SetTag("asyncresponse.exception_type", exception.GetType().FullName ?? exception.GetType().Name);
 3256        AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, false);
 257
 258        try
 259        {
 3260            if (recoveryState?.FailureCallback == null)
 261            {
 3262                _logger.LogWarning("No subscribers for channel {Channel}; no failure callback available.", channel);
 3263                activity?.SetTag("asyncresponse.recovery.callback_invoked", false);
 2264                return false;
 265            }
 266
 3267            _logger.LogWarning("No subscribers for channel {Channel}; invoking failure callback.", channel);
 268
 3269            var invocation = ReflectionExtensions.ResolveCallback(
 3270                recoveryState.FailureCallback,
 3271                payload: null,
 3272                exception: exception,
 3273                correlationId: recoveryState.CorrelationId
 3274            );
 275
 3276            await InvokeAsync(invocation, recoveryState.Context).ConfigureAwait(false);
 277
 3278            _logger.LogInformation("Failure callback invoked for channel {Channel}.", channel);
 3279            activity?.SetTag("asyncresponse.recovery.callback_invoked", true);
 280
 3281            return true;
 282        }
 3283        catch (Exception ex)
 284        {
 3285            AsyncResponseDiagnostics.SetError(activity, ex);
 2286            throw;
 287        }
 3288    }
 289
 290    /// <summary>
 291    /// Routes a payload that declined to resume (<see cref="IAsyncResponsePayload.ShouldResumeOnRecovery"/>
 292    /// returned <c>false</c>, or it could not be classified) to the failure callback, wrapped in an
 293    /// <see cref="AsyncResponseDomainFailureException"/> — so it takes the same path as a technical
 294    /// <c>SetException</c>.
 295    /// </summary>
 296    private async Task<bool> DispatchToFailureCallback<T>(RecoveryState recoveryState, T response, string channel, Activ
 297    {
 3298        string? payloadJson = null;
 299        try
 300        {
 3301            payloadJson = AsyncResponseJson.Serialize(response);
 3302        }
 3303        catch (Exception)
 304        {
 305            // Ignore serialization failure here; the payload is only attached for diagnostics.
 306            // Under trimmed/AOT deployments this also covers payload types without registered
 307            // JSON metadata — the callback still fires, just without the diagnostic JSON.
 3308        }
 309
 3310        if (recoveryState.FailureCallback == null)
 311        {
 3312            _logger.LogError("No subscribers for channel {Channel} and the response declined to resume, but no failure c
 2313            return false;
 314        }
 315
 3316        _logger.LogWarning("No subscribers for channel {Channel}; response declined to resume, invoking failure callback
 317
 3318        var domainFailure = new AsyncResponseDomainFailureException(
 3319            recoveryState.CorrelationId,
 3320            recoveryState.PayloadTypeFullName,
 3321            payloadJson);
 322
 3323        var invocation = ReflectionExtensions.ResolveCallback(
 3324            recoveryState.FailureCallback,
 3325            payload: response,
 3326            exception: domainFailure,
 3327            correlationId: recoveryState.CorrelationId
 3328        );
 329
 330        try
 331        {
 332            // Bounded in-process retry, mirroring the ingress's transient-fault policy: a failure
 333            // callback is re-invocable by contract (broker redelivery re-invokes it the same way),
 334            // and a one-shot invoke turned a transient dependency blip into a silently dropped
 335            // domain-failure signal that nothing ever revisited (the watchdog is report-only).
 3336            await AsyncResponseRetry.ExecuteAsync(
 3337                async _ =>
 3338                {
 3339                    await InvokeAsync(invocation, recoveryState.Context).ConfigureAwait(false);
 3340                    return true;
 3341                },
 2342                isTransient: static _ => true,
 3343                maxAttempts: 4,
 3344                baseDelay: TimeSpan.FromMilliseconds(250),
 3345                maxDelay: TimeSpan.FromSeconds(2),
 3346                CancellationToken.None).ConfigureAwait(false);
 347
 3348            _logger.LogInformation("Failure callback invoked for channel {Channel}.", channel);
 349
 3350            return true;
 351        }
 3352        catch (Exception ex)
 353        {
 354            // Deliberately not rethrown once the retries are exhausted — keep the swallow. An
 355            // exception would bubble up to the broker ingress, which reacts with SetException and
 356            // would invoke this same failure callback a second time; routing it to transport
 357            // redelivery instead would hot-loop a permanently-throwing callback on RabbitMQ's
 358            // unbounded default. The domain failure has already been dispatched (and retried
 359            // above), and the kept recovery row is surfaced by the watchdog's staleness report,
 360            // so the drop is operator-visible rather than silent.
 3361            AsyncResponseDiagnostics.SetError(activity, ex);
 2362            _logger.LogError(ex, "Failure callback failed for channel {Channel}.", channel);
 2363            return false;
 364        }
 3365    }
 366
 367    private async Task InvokeAsync(ReflectionInvocationDto invocation, IReadOnlyDictionary<string, string>? context)
 368    {
 369        // The recovery callback may run in a different deployment than the original waiter, so
 370        // restore any ambient context captured at registration before resolving and invoking it.
 3371        using var contextScope = _propagation.Restore(context);
 3372        await using var serviceScope = _scopeFactory.CreateAsyncScope();
 3373        await serviceScope.ServiceProvider.InvokeAsync(invocation).ConfigureAwait(false);
 3374    }
 375
 376}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.DependencyInjection.IServiceScopeFactory, AsyncResponse.AsyncResponseContextPropagation, Microsoft.Extensions.Logging.ILogger)
DispatchLostResponses()
DispatchLostExceptions()
DispatchLostResponse()
DispatchLostException()
DispatchToFailureCallback()
<DispatchToFailureCallback()
InvokeAsync()