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

Information
Class: AsyncResponse.LostSubscriberCallbackDispatcher
Assembly: AsyncResponse.Core
File(s): /_/src/AsyncResponse.Core/LostSubscriberCallbackDispatcher.cs
Line coverage
95%
Covered lines: 257
Uncovered lines: 12
Coverable lines: 269
Total lines: 704
Line coverage: 95.5%
Branch coverage
89%
Covered branches: 146
Total branches: 164
Branch coverage: 89%
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()92.85%282897.14%
DispatchLostExceptions()90.9%222296%
ThrowUnsettled(...)87.5%9875%
SettleResidualFailures(...)78.57%141490.62%
DispatchLostResponse()95%414092.98%
DispatchLostException()94.44%1818100%
DispatchToFailureCallback()83.33%181897.95%
<DispatchToFailureCallback()100%11100%
IsPermanentCallbackFailure(...)62.5%88100%
DeleteConsumedRegistrationAsync()100%11100%
WirePayload(...)87.5%88100%
InvokeAsync()100%11100%

File(s)

/_/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="Action">
 13/// The recovery route the payload reported (<see cref="IAsyncResponsePayload.OnRecovery"/>):
 14/// <see cref="RecoveryAction.Resume"/>, <see cref="RecoveryAction.Fail"/>,
 15/// <see cref="RecoveryAction.KeepWaiting"/> (non-terminal checkpoint — nothing invoked, the
 16/// registration stays armed), or <c>null</c> when it could not be classified (no recovery state,
 17/// missing payload type, null payload, conversion failure) — treated as "do not resume".
 18/// </param>
 19/// <param name="CallbackInvoked">
 20/// <c>true</c> when a callback was invoked successfully — the recovery state is consumed and the
 21/// caller should delete it.
 22/// </param>
 23internal readonly record struct LostSubscriberDispatchResult(RecoveryAction? Action, bool CallbackInvoked)
 24{
 25    /// <summary>
 26    /// <c>true</c> when a live subscriber re-appeared between the caller's empty snapshot and the
 27    /// recovery-state read. No callback was invoked and no state was consumed — the caller should
 28    /// re-snapshot and dispatch live.
 29    /// </summary>
 30    public bool RetryLive { get; init; }
 31
 32    /// <summary>
 33    /// <c>true</c> when shared-correlation registrations legitimately took DIFFERENT routes in
 34    /// this one dispatch (each registration classifies as the payload type IT registered).
 35    /// <see cref="Action"/> stays <c>null</c> — no single action describes the aggregate — but
 36    /// diagnostics report the route as <c>mixed</c> rather than <c>unclassified</c>; each
 37    /// registration's own dispatch activity carries its true route.
 38    /// </summary>
 39    public bool RouteMixed { get; init; }
 40}
 41
 42/// <summary>
 43/// The single decision point of the lost-subscriber fallback: when an async response is published
 44/// and no subscriber is listening (the original waiter died, e.g. with a redeploy/restart), this
 45/// dispatcher chooses and invokes the callback persisted in the <see cref="RecoveryState"/>.
 46/// <para>
 47/// For payload envelopes (<c>SetResponse</c>) the payload's
 48/// <see cref="IAsyncResponsePayload.OnRecovery"/> decides the route:
 49/// <see cref="RecoveryAction.Resume"/> goes to the resume callback;
 50/// <see cref="RecoveryAction.Fail"/> (and any unclassifiable payload, conservatively) goes to the
 51/// failure callback wrapped in an <see cref="AsyncResponseDomainFailureException"/>; and
 52/// <see cref="RecoveryAction.KeepWaiting"/> — a non-terminal checkpoint — invokes nothing and
 53/// leaves the registration armed for a later response. For exception envelopes
 54/// (<c>SetException</c>) the failure callback is always used.
 55/// </para>
 56/// <para>
 57/// The chosen callback receives the <em>materialized</em> payload (the registered payload type),
 58/// not the raw broker JSON — an <c>object</c>-/interface-/base-typed callback parameter must get
 59/// the concrete instance, or every type guard in the consuming flow silently fails.
 60/// </para>
 61/// <para>
 62/// The publisher stays a plain transport: it only reports "published, but nobody was listening" and
 63/// hands over to this dispatcher. This decision is independent of the live waiter's <c>Until</c>
 64/// predicate, which no longer exists once the waiter is lost.
 65/// </para>
 66/// </summary>
 366767internal sealed class LostSubscriberCallbackDispatcher(
 366768    IServiceScopeFactory _scopeFactory,
 366769    AsyncResponseContextPropagation _propagation,
 366770    ILogger _logger,
 366771    TimeProvider? _timeProvider = null)
 72{
 73    /// <summary>
 74    /// Loads every recovery registration for <paramref name="correlationId"/> and dispatches a lost
 75    /// response to each registration's resume/failure callback.
 76    /// </summary>
 77    public async Task<LostSubscriberDispatchResult> DispatchLostResponses<T>(
 78        IRecoveryStateStore recoveryStateStore,
 79        string correlationId,
 80        T response,
 81        string channel,
 82        CancellationToken cancellationToken,
 83        Func<ValueTask<bool>>? hasLiveSubscriber = null)
 84    {
 33585        var recoveryStates = await recoveryStateStore.GetAllAsync(correlationId, cancellationToken).ConfigureAwait(false
 86
 87        // A waiter registers its subscription before saving its recovery state, so the snapshot
 88        // race has two shapes — and the re-check must run before the empty-state early return:
 89        // a recovery state visible here implies its subscription is visible too, and, inversely, a
 90        // freshly registered subscription may not have saved its state yet, in which case
 91        // recoveryStates is empty precisely because the waiter is about to go live. Either way a
 92        // live subscriber means the "nobody listening" premise was stale — hand the response back
 93        // for live delivery instead of consuming registrations or dropping it unrecoverably.
 33194        if (hasLiveSubscriber is not null && await hasLiveSubscriber().ConfigureAwait(false))
 3895            return new LostSubscriberDispatchResult(null, false) { RetryLive = true };
 96
 97        // Classification and callbacks must see the payload exactly as a broker delivery would
 98        // have carried it, whichever process publishes.
 28799        var wirePayload = WirePayload(response);
 100
 287101        if (recoveryStates.Count == 0)
 67102            return await DispatchLostResponse(null, wirePayload, channel).ConfigureAwait(false);
 103
 220104        var callbackInvoked = false;
 220105        RecoveryAction? action = null;
 220106        var routeSet = false;
 220107        var routeMixed = false;
 220108        List<ExceptionDispatchInfo>? failures = null;
 109
 972110        foreach (var recoveryState in recoveryStates)
 111        {
 112            try
 113            {
 266114                var result = await DispatchLostResponse(recoveryState, wirePayload, channel).ConfigureAwait(false);
 216115                if (!routeSet)
 116                {
 206117                    action = result.Action;
 206118                    routeSet = true;
 119                }
 10120                else if (action != result.Action)
 121                {
 8122                    routeMixed = true;
 123                }
 124
 216125                if (!result.CallbackInvoked)
 58126                    continue;
 127
 158128                callbackInvoked = true;
 158129                await DeleteConsumedRegistrationAsync(recoveryStateStore, correlationId, recoveryState.RegistrationId, c
 158130            }
 50131            catch (Exception ex)
 132            {
 50133                if (ex is OperationCanceledException && cancellationToken.IsCancellationRequested)
 0134                    throw;
 135
 136                // Capture rather than re-throw a bare variable so the original throw site's stack
 137                // trace survives the dispatch to the remaining registrations. EVERY failure is
 138                // kept: settlement below classifies the whole set, not the first one.
 50139                (failures ??= []).Add(ExceptionDispatchInfo.Capture(ex));
 50140            }
 208141        }
 142
 220143        if (failures is not null)
 144        {
 34145            if (!callbackInvoked)
 14146                ThrowUnsettled(failures, correlationId);
 147
 20148            SettleResidualFailures(failures, correlationId, channel, "response");
 149        }
 150
 190151        return new LostSubscriberDispatchResult(routeMixed ? null : action, callbackInvoked) { RouteMixed = routeMixed }
 295152    }
 153
 154    /// <summary>
 155    /// Loads every recovery registration for <paramref name="correlationId"/> and dispatches a lost
 156    /// exception to each registration's failure callback.
 157    /// </summary>
 158    public async Task<LostSubscriberDispatchResult> DispatchLostExceptions(
 159        IRecoveryStateStore recoveryStateStore,
 160        string correlationId,
 161        Exception exception,
 162        string channel,
 163        CancellationToken cancellationToken,
 164        Func<ValueTask<bool>>? hasLiveSubscriber = null)
 165    {
 88166        var recoveryStates = await recoveryStateStore.GetAllAsync(correlationId, cancellationToken).ConfigureAwait(false
 167
 168        // Same snapshot-race re-check as DispatchLostResponses, and for the same reason it must
 169        // precede the empty-state early return: an empty snapshot may mean the waiter registered
 170        // its subscription but has not saved its recovery state yet. A live subscriber means the
 171        // exception should be delivered live instead of consumed here.
 86172        if (hasLiveSubscriber is not null && await hasLiveSubscriber().ConfigureAwait(false))
 20173            return new LostSubscriberDispatchResult(RecoveryAction.Fail, false) { RetryLive = true };
 174
 64175        if (recoveryStates.Count == 0)
 15176            return new LostSubscriberDispatchResult(RecoveryAction.Fail, await DispatchLostException(null, exception, ch
 177
 49178        var callbackInvoked = false;
 49179        List<ExceptionDispatchInfo>? failures = null;
 180
 252181        foreach (var recoveryState in recoveryStates)
 182        {
 183            try
 184            {
 77185                if (!await DispatchLostException(recoveryState, exception, channel).ConfigureAwait(false))
 4186                    continue;
 187
 43188                callbackInvoked = true;
 43189                await DeleteConsumedRegistrationAsync(recoveryStateStore, correlationId, recoveryState.RegistrationId, c
 43190            }
 30191            catch (Exception ex)
 192            {
 30193                if (ex is OperationCanceledException && cancellationToken.IsCancellationRequested)
 0194                    throw;
 195
 196                // Capture rather than re-throw a bare variable so the original throw site's stack
 197                // trace survives the dispatch to the remaining registrations. EVERY failure is
 198                // kept: settlement below classifies the whole set, not the first one.
 30199                (failures ??= []).Add(ExceptionDispatchInfo.Capture(ex));
 30200            }
 73201        }
 202
 49203        if (failures is not null)
 204        {
 18205            if (!callbackInvoked)
 4206                ThrowUnsettled(failures, correlationId);
 207
 14208            SettleResidualFailures(failures, correlationId, channel, "exception");
 209        }
 210
 211        // Exception envelopes always take the failure route, so the action is fixed at Fail.
 31212        return new LostSubscriberDispatchResult(RecoveryAction.Fail, callbackInvoked);
 66213    }
 214
 215    /// <summary>
 216    /// Rethrows for a fan-out in which NO registration's callback succeeded. A sibling whose
 217    /// failure-callback ladder was already exhausted (<see cref="RecoveryCallbackFailedException"/>)
 218    /// wins whatever its position: the ingress passes that shape through untouched, so the
 219    /// transport redelivers the still-unacknowledged signal to every registration — instead of
 220    /// the ingress burning its own retry ladder on an earlier sibling's failure and then escalating
 221    /// through <c>SetException</c>, which re-invokes the very failure callback that just gave up.
 222    /// Any other transient failure is wrapped for the same redelivery path. Only an entirely
 223    /// deterministic set retains the ingress's retry-then-escalate handling.
 224    /// </summary>
 225    private static void ThrowUnsettled(List<ExceptionDispatchInfo> failures, string correlationId)
 226    {
 72227        foreach (var failure in failures)
 228        {
 22229            if (failure.SourceException is RecoveryCallbackFailedException)
 8230                failure.Throw();
 231        }
 232
 233        // No successful sibling does not make a transient resume failure a business failure.
 234        // In particular, RecoverAsync may have checkpointed the response before its wake-up
 235        // publish failed. Escalating through SetException then consumes its registration without
 236        // publishing that wake-up. Preserve the original signal for transport redelivery.
 34237        foreach (var failure in failures)
 238        {
 12239            if (!IsPermanentCallbackFailure(failure.SourceException))
 10240                throw new RecoveryCallbackFailedException(correlationId, attempts: 1, failure.SourceException);
 241        }
 242
 0243        failures[0].Throw();
 0244    }
 245
 246    /// <summary>
 247    /// Settles a fan-out dispatch in which at least one registration's callback succeeded (and
 248    /// was consumed) while one or more others failed. Shared-correlation registrations are an
 249    /// expected shape — a worker that died mid-await leaves its registration beside the
 250    /// replacement's — and each carries its own delivery guarantee, so the verdict is taken over
 251    /// the WHOLE set of failures, never the first one alone.
 252    /// <para>
 253    /// A <b>deterministic</b> failure (the target is unauthorized, unresolvable, not registered,
 254    /// or no longer binds) is logged: redelivery cannot fix it and its registration stays for the
 255    /// watchdog to surface. A <b>transient</b> one — any failure that is not deterministic —
 256    /// propagates as <see cref="RecoveryCallbackFailedException"/>, which the ingress passes
 257    /// through untouched (no second retry ladder, no <c>SetException</c> escalation that would
 258    /// invoke the FAILURE callbacks of registrations whose resume merely blipped), so the
 259    /// transport redelivers the terminal signal; the consumed registrations are already deleted,
 260    /// so the redelivery reaches only the registrations that failed. The message is acknowledged
 261    /// only when EVERY failure was deterministic. Before round 39 the verdict was taken from the
 262    /// first failure alone: a deterministic failure first in the set hid a transient sibling
 263    /// behind it, the message was acknowledged, and the transient registration — a valid waiter
 264    /// whose dependency was briefly down — lost the only copy of its payload, with the outcome
 265    /// depending on the order the store returned the registrations in.
 266    /// </para>
 267    /// </summary>
 268    private void SettleResidualFailures(List<ExceptionDispatchInfo> failures, string correlationId, string channel, stri
 269    {
 34270        ExceptionDispatchInfo? exhausted = null;
 34271        Exception? transient = null;
 184272        foreach (var failure in failures)
 273        {
 58274            var residual = failure.SourceException;
 275
 276            // Already the propagating shape (a sibling's failure-callback ladder was exhausted);
 277            // its own ladder logged it.
 58278            if (residual is RecoveryCallbackFailedException)
 279            {
 0280                exhausted ??= failure;
 0281                continue;
 282            }
 283
 58284            if (IsPermanentCallbackFailure(residual))
 285            {
 28286                _logger.LogError(
 28287                    residual,
 28288                    "Lost-{Kind} dispatch for correlationId {CorrelationId} on {Channel} failed with a deterministic fau
 28289                    kind,
 28290                    correlationId,
 28291                    channel);
 28292                continue;
 293            }
 294
 30295            transient ??= residual;
 296        }
 297
 34298        if (exhausted is not null)
 0299            exhausted.Throw();
 300
 34301        if (transient is null)
 302        {
 4303            _logger.LogError(
 4304                "Lost-{Kind} dispatch for correlationId {CorrelationId} on {Channel} partially failed with deterministic
 4305                kind,
 4306                correlationId,
 4307                channel);
 4308            return;
 309        }
 310
 30311        _logger.LogError(
 30312            transient,
 30313            "Lost-{Kind} dispatch for correlationId {CorrelationId} on {Channel} partially failed transiently after anot
 30314            kind,
 30315            correlationId,
 30316            channel);
 30317        throw new RecoveryCallbackFailedException(correlationId, attempts: 1, transient);
 318    }
 319
 320    /// <summary>Dispatches a successfully published payload that no subscriber received.</summary>
 321    public async Task<LostSubscriberDispatchResult> DispatchLostResponse<T>(RecoveryState? recoveryState, T response, st
 322    {
 333323        using var activity = AsyncResponseDiagnostics.StartActivity(
 333324            "asyncresponse.lost_subscriber.dispatch",
 333325            correlationId: recoveryState?.CorrelationId);
 333326        activity?.SetTag("asyncresponse.lost_subscriber.kind", "response");
 333327        activity?.SetTag("asyncresponse.channel_name", channel);
 333328        if (response is not null)
 331329            AsyncResponseDiagnostics.SetPayloadType(activity, response.GetType());
 330
 331        try
 332        {
 333            // The recovering process has no live Until predicate — the payload itself decides
 334            // whether this late response resumes the flow, fails it, or is a non-terminal
 335            // checkpoint to wait past. Classification also MATERIALIZES the payload as the
 336            // registered type; the chosen callback must receive that instance, never the raw
 337            // broker JSON (an object-typed parameter otherwise gets a JsonElement and every type
 338            // guard in the consuming flow silently fails). A null (unclassifiable) verdict is
 339            // treated conservatively as "do not resume", so a payload that cannot be understood
 340            // never takes the happy path.
 341            //
 342            // A payload type name past the resolution limits is never parsed (it can overflow the
 343            // stack inside the CLR's type-name parser — see IsWithinResolutionLimits), so it
 344            // classifies as unresolvable and takes the same conservative route. Said out loud,
 345            // because unlike a renamed type this is a recovery row nobody's code wrote.
 333346            if (recoveryState?.PayloadTypeFullName is { } payloadTypeName
 333347                && !AsyncResponseTypeResolution.IsWithinResolutionLimits(payloadTypeName))
 348            {
 0349                _logger.LogError(
 0350                    "The recovery registration for channel {Channel} names a payload type that is not resolved: {Payload
 0351                    channel,
 0352                    AsyncResponseTypeResolution.DescribeForDiagnostics(payloadTypeName));
 353            }
 354
 333355            var classification = recoveryState is null
 333356                ? default
 333357                : PayloadRecoveryClassifier.Classify(response, recoveryState.PayloadTypeFullName);
 333358            var action = classification.Action;
 333359            var callbackPayload = classification.MaterializedPayload ?? (object?)response;
 333360            AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, action);
 361
 333362            if (action == RecoveryAction.KeepWaiting)
 363            {
 364                // A non-terminal checkpoint (progress report) with nobody listening: invoking the
 365                // resume callback here would spawn a worker per checkpoint, and the failure route
 366                // would fail a flow that is still running — both consume the registration and
 367                // leave the REAL terminal response with nothing to route against (the flow then
 368                // deadlocks re-attached to a correlation id nothing can answer). Invoke nothing;
 369                // the caller keeps the registration armed, bounded by its TTL and visible to the
 370                // watchdog.
 44371                _logger.LogInformation(
 44372                    "No subscribers for channel {Channel}; payload is a non-terminal checkpoint (KeepWaiting) — recovery
 44373                    channel);
 44374                activity?.SetTag("asyncresponse.recovery.callback_invoked", false);
 44375                return new LostSubscriberDispatchResult(action, false);
 376            }
 377
 289378            if (action != RecoveryAction.Resume)
 379            {
 120380                if (recoveryState is null)
 381                {
 67382                    _logger.LogWarning("No subscribers and no recovery state for channel {Channel}.", channel);
 67383                    activity?.SetTag("asyncresponse.recovery.callback_invoked", false);
 67384                    return new LostSubscriberDispatchResult(action, false);
 385                }
 386
 53387                var invoked = await DispatchToFailureCallback(recoveryState, callbackPayload, channel, activity).Configu
 47388                activity?.SetTag("asyncresponse.recovery.callback_invoked", invoked);
 47389                return new LostSubscriberDispatchResult(action, invoked);
 390            }
 391
 392            // action == Resume implies recoveryState is non-null (the verdict is null otherwise).
 169393            if (recoveryState!.ResumeCallback == null)
 394            {
 395                // A resumable response with no resume callback registered: the flow cannot
 396                // proceed, which is exactly what the failure route reports — engage the armed
 397                // failure callback rather than repeatedly discarding the terminal signal until
 398                // the registration's TTL. Mirrors the unclassifiable route's conservatism; a
 399                // registration with NEITHER callback keeps the old warn-and-retain behavior.
 8400                if (recoveryState.FailureCallback != null)
 401                {
 4402                    _logger.LogWarning("No subscribers for channel {Channel}; payload is resumable but no resume callbac
 4403                    var fallbackInvoked = await DispatchToFailureCallback(recoveryState, callbackPayload, channel, activ
 2404                    activity?.SetTag("asyncresponse.recovery.callback_invoked", fallbackInvoked);
 2405                    return new LostSubscriberDispatchResult(action, fallbackInvoked);
 406                }
 407
 4408                _logger.LogWarning("No subscribers for channel {Channel}; no resume callback available.", channel);
 4409                activity?.SetTag("asyncresponse.recovery.callback_invoked", false);
 4410                return new LostSubscriberDispatchResult(action, false);
 411            }
 412
 161413            _logger.LogWarning("No subscribers for channel {Channel}; invoking resume callback.", channel);
 414
 161415            var invocation = ReflectionExtensions.ResolveCallback(
 161416                recoveryState.ResumeCallback,
 161417                payload: callbackPayload,
 161418                exception: null,
 161419                correlationId: recoveryState.CorrelationId
 161420            );
 421
 422            // The outer fan-out settlement wraps transient failures for transport redelivery,
 423            // including a single failed resume. Infrastructure failure must not become a
 424            // business-failure callback. This catch only marks the activity before rethrowing.
 161425            await InvokeAsync(invocation, recoveryState.Context).ConfigureAwait(false);
 426
 119427            _logger.LogInformation("Resume callback invoked for channel {Channel}.", channel);
 119428            activity?.SetTag("asyncresponse.recovery.callback_invoked", true);
 429
 119430            return new LostSubscriberDispatchResult(action, true);
 431        }
 50432        catch (Exception ex)
 433        {
 50434            AsyncResponseDiagnostics.SetError(activity, ex);
 50435            throw;
 436        }
 283437    }
 438
 439    /// <summary>Dispatches an exception envelope that no subscriber received.</summary>
 440    public async Task<bool> DispatchLostException(RecoveryState? recoveryState, Exception exception, string channel)
 441    {
 92442        using var activity = AsyncResponseDiagnostics.StartActivity(
 92443            "asyncresponse.lost_subscriber.dispatch",
 92444            correlationId: recoveryState?.CorrelationId);
 92445        activity?.SetTag("asyncresponse.lost_subscriber.kind", "exception");
 92446        activity?.SetTag("asyncresponse.channel_name", channel);
 92447        activity?.SetTag("asyncresponse.exception_type", exception.GetType().FullName ?? exception.GetType().Name);
 92448        AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, RecoveryAction.Fail);
 449
 450        try
 451        {
 92452            if (recoveryState?.FailureCallback == null)
 453            {
 19454                _logger.LogWarning("No subscribers for channel {Channel}; no failure callback available.", channel);
 19455                activity?.SetTag("asyncresponse.recovery.callback_invoked", false);
 19456                return false;
 457            }
 458
 73459            _logger.LogWarning("No subscribers for channel {Channel}; invoking failure callback.", channel);
 460
 73461            var invocation = ReflectionExtensions.ResolveCallback(
 73462                recoveryState.FailureCallback,
 73463                payload: null,
 73464                exception: exception,
 73465                correlationId: recoveryState.CorrelationId
 73466            );
 467
 73468            await InvokeAsync(invocation, recoveryState.Context).ConfigureAwait(false);
 469
 43470            _logger.LogInformation("Failure callback invoked for channel {Channel}.", channel);
 43471            activity?.SetTag("asyncresponse.recovery.callback_invoked", true);
 472
 43473            return true;
 474        }
 30475        catch (Exception ex)
 476        {
 30477            AsyncResponseDiagnostics.SetError(activity, ex);
 30478            throw;
 479        }
 62480    }
 481
 482    /// <summary>
 483    /// Routes a payload that declined to resume (<see cref="IAsyncResponsePayload.OnRecovery"/>
 484    /// returned <see cref="RecoveryAction.Fail"/>, or it could not be classified) to the failure
 485    /// callback, wrapped in an <see cref="AsyncResponseDomainFailureException"/> — so it takes the
 486    /// same path as a technical <c>SetException</c>. <paramref name="response"/> is the
 487    /// materialized payload when classification succeeded, the raw one otherwise.
 488    /// </summary>
 489    private async Task<bool> DispatchToFailureCallback(RecoveryState recoveryState, object? response, string channel, Ac
 490    {
 57491        string? payloadJson = null;
 492        try
 493        {
 494            // The payload arrives as its wire representation (declared-type-normalized JSON or
 495            // raw ingress JSON); reuse it verbatim for diagnostics rather than re-serializing.
 57496            payloadJson = response switch
 57497            {
 0498                string s => s,
 6499                JsonElement je => je.GetRawText(),
 2500                null => AsyncResponseJson.Serialize(response),
 49501                _ => AsyncResponseJson.Serialize(response, response.GetType())
 57502            };
 55503        }
 2504        catch (Exception)
 505        {
 506            // Ignore serialization failure here; the payload is only attached for diagnostics.
 507            // Under trimmed/AOT deployments this also covers payload types without registered
 508            // JSON metadata — the callback still fires, just without the diagnostic JSON.
 2509        }
 510
 511        // Size only, never the JSON itself: these run at Error/Warning in production, and the
 512        // payload body is business data (the same rule the ingress applies — no content, not even
 513        // a hash). The full JSON still travels on AsyncResponseDomainFailureException.PayloadJson,
 514        // which deliberately keeps it out of Exception.Message and therefore out of generic
 515        // exception logging.
 57516        if (recoveryState.FailureCallback == null)
 517        {
 6518            _logger.LogError("No subscribers for channel {Channel} and the response declined to resume, but no failure c
 6519            return false;
 520        }
 521
 51522        _logger.LogWarning("No subscribers for channel {Channel}; response declined to resume, invoking failure callback
 523
 524        // The type name goes into the exception's MESSAGE, which generic exception logging
 525        // sweeps up: bounded and escaped like every other quote of a persisted name. An ordinary
 526        // name passes through unchanged.
 51527        var domainFailure = new AsyncResponseDomainFailureException(
 51528            recoveryState.CorrelationId,
 51529            recoveryState.PayloadTypeFullName is { } registeredTypeName
 51530                ? AsyncResponseTypeResolution.DescribeForDiagnostics(registeredTypeName)
 51531                : null,
 51532            payloadJson);
 533
 51534        var invocation = ReflectionExtensions.ResolveCallback(
 51535            recoveryState.FailureCallback,
 51536            payload: response,
 51537            exception: domainFailure,
 51538            correlationId: recoveryState.CorrelationId
 51539        );
 540
 541        try
 542        {
 543            // Bounded in-process retry, mirroring the ingress's transient-fault policy: a failure
 544            // callback is re-invocable by contract (broker redelivery re-invokes it the same way),
 545            // and a one-shot invoke turned a transient dependency blip into a silently dropped
 546            // domain-failure signal that nothing ever revisited (the watchdog is report-only).
 547            //
 548            // Retry only what a retry can fix. A blanket "everything is transient" spent the full
 549            // ~1.75s backoff ladder on faults that are deterministic by construction — a callback
 550            // whose target is not registered in DI, or whose persisted type/method no longer
 551            // resolves, fails identically on attempt 4 — so a single misconfiguration taxed EVERY
 552            // lost-response dispatch on the publisher's thread and compounded with the transport's
 553            // own redelivery. Those cases now fail fast and loudly on the first attempt; genuine
 554            // dependency blips keep the full ladder.
 51555            await AsyncResponseRetry.ExecuteAsync(
 51556                async _ =>
 51557                {
 77558                    await InvokeAsync(invocation, recoveryState.Context).ConfigureAwait(false);
 39559                    return true;
 39560                },
 30561                isTransient: static ex => !IsPermanentCallbackFailure(ex),
 51562                maxAttempts: FailureCallbackAttempts,
 51563                baseDelay: TimeSpan.FromMilliseconds(250),
 51564                maxDelay: TimeSpan.FromSeconds(2),
 51565                CancellationToken.None,
 51566                _timeProvider).ConfigureAwait(false);
 567
 39568            _logger.LogInformation("Failure callback invoked for channel {Channel}.", channel);
 569
 39570            return true;
 571        }
 12572        catch (Exception ex)
 573        {
 12574            AsyncResponseDiagnostics.SetError(activity, ex);
 575
 12576            if (IsPermanentCallbackFailure(ex))
 577            {
 578                // Deterministic by construction: the same call fails the same way on every
 579                // delivery, so redelivery would only burn the transport's attempts (or hot-loop
 580                // on RabbitMQ's unbounded default). Swallow: the message is acknowledged, the
 581                // kept recovery row is surfaced by the watchdog's staleness report, and the
 582                // error log names the misconfiguration to fix.
 4583                _logger.LogError(ex, "Failure callback for channel {Channel} cannot be invoked (deterministic fault); th
 4584                return false;
 585            }
 586
 587            // Transient and exhausted: the response is a TERMINAL signal that, once acknowledged,
 588            // exists nowhere (the recovery row keeps the callback, not the payload; the watchdog
 589            // only reports). Propagate as a dedicated type the ingress passes through untouched —
 590            // no retry (the ladder above already ran) and no SetException escalation (that would
 591            // only re-invoke this same callback) — so the transport keeps the message for its own
 592            // bounded redelivery and dead-letter policy. On RabbitMQ's default MaxDeliveryAttempts
 593            // = 0 that is the documented unlimited requeue any failing handler gets; configure a
 594            // cap there as for worker jobs.
 8595            _logger.LogError(
 8596                ex,
 8597                "Failure callback for channel {Channel} failed on all {Attempts} attempts; the message is left unacknowl
 8598                channel,
 8599                FailureCallbackAttempts);
 8600            throw new RecoveryCallbackFailedException(recoveryState.CorrelationId ?? string.Empty, FailureCallbackAttemp
 601        }
 49602    }
 603
 604    /// <summary>In-process invocations of a failure callback per delivery before the delivery is handed back to the tra
 605    internal const int FailureCallbackAttempts = 4;
 606
 607    /// <summary>
 608    /// Whether a failed callback invocation is deterministic — the same call will fail the same
 609    /// way on every attempt, so retrying only burns the backoff ladder on the publish path.
 610    /// <para>
 611    /// Narrow on purpose: only faults raised while WIRING UP the call qualify — the target is
 612    /// unauthorized, its persisted type no longer resolves, its service is not registered
 613    /// (<see cref="CallbackTargetUnresolvableException"/>), or its method/arguments no longer bind
 614    /// (<see cref="MissingMethodException"/>, <see cref="TypeLoadException"/>). A failure thrown by
 615    /// the callback BODY is never classified here, whatever its type: a handler that throws
 616    /// <see cref="InvalidOperationException"/> for a transient reason is ordinary application code
 617    /// and keeps the full retry ladder, which is why the marker type exists rather than a plain
 618    /// <c>is InvalidOperationException</c> test.
 619    /// </para>
 620    /// </summary>
 621    private static bool IsPermanentCallbackFailure(Exception exception)
 112622        => exception is CallbackTargetUnresolvableException
 112623            or MissingMethodException
 112624            or MissingMemberException
 112625            or TypeLoadException;
 626
 627    /// <summary>
 628    /// Deletes a registration whose callback has already been invoked successfully. Best-effort by
 629    /// contract: the callback IS the outcome, and a cleanup fault must never be reinterpreted as a
 630    /// failed response — rethrowing here made the ingress retry the whole delivery (re-invoking
 631    /// the callback) and then publish the CLEANUP exception through SetException, invoking the
 632    /// failure callback for a flow whose resume had already succeeded. A failed delete leaves the
 633    /// registration to its TTL and the watchdog; recovery is at-least-once, so a later delivery
 634    /// re-invoking the callback is within contract.
 635    /// </summary>
 636    private async Task DeleteConsumedRegistrationAsync(
 637        IRecoveryStateStore recoveryStateStore,
 638        string correlationId,
 639        Guid registrationId,
 640        CancellationToken cancellationToken)
 641    {
 642        try
 643        {
 201644            await recoveryStateStore.TryDeleteAsync(correlationId, registrationId, cancellationToken).ConfigureAwait(fal
 197645        }
 4646        catch (Exception ex)
 647        {
 4648            _logger.LogWarning(
 4649                ex,
 4650                "Recovery callback for correlationId {CorrelationId} succeeded but deleting its registration {Registrati
 4651                correlationId,
 4652                registrationId);
 4653        }
 201654    }
 655
 656    /// <summary>
 657    /// Normalizes a typed payload to its WIRE representation — serialized as the publisher's
 658    /// DECLARED <typeparamref name="T"/>, exactly as <c>AsyncResponseEnvelope&lt;T&gt;</c> writes
 659    /// it. Reusing the live instance leaked state that never crosses the wire
 660    /// (<c>[JsonIgnore]</c>) into recovery routing, and serializing the RUNTIME type dropped the
 661    /// polymorphic discriminators only the declared-type contract emits — either way the verdict
 662    /// depended on which side of a serialization boundary the publisher sat. Raw ingress payloads
 663    /// (<see cref="JsonElement"/> / JSON string) already are wire representations.
 664    /// </summary>
 665    private static object? WirePayload<T>(T response)
 666    {
 287667        if (response is null or JsonElement or string)
 158668            return response;
 669
 670        try
 671        {
 129672            return AsyncResponseJson.Serialize(response);
 673        }
 2674        catch
 675        {
 676            // No wire representation exists (unserializable payload — cycles, unregistered AOT
 677            // metadata). Hand the instance through: the wire-only classifier treats it as
 678            // unclassifiable, so it takes the conservative failure route with the instance
 679            // attached — a payload that could never have crossed the wire never resumes a flow.
 2680            return response;
 681        }
 129682    }
 683
 684    private async Task InvokeAsync(ReflectionInvocationDto invocation, IReadOnlyDictionary<string, string>? context)
 685    {
 311686        await using var serviceScope = _scopeFactory.CreateAsyncScope();
 687
 688        // Authorize first, on the raw persisted descriptor. Both the callback target and the
 689        // context carrier come out of the recovery store, so anyone who can write a row there
 690        // supplies both — and restoring the context first would let the row pick the ambient
 691        // tenant/principal an authorizer consults to decide whether that same row's target may
 692        // run. The scope exists by now only to resolve the authorizer; nothing has been invoked.
 311693        ReflectionExtensions.ThrowIfNotAuthorized(
 311694            serviceScope.ServiceProvider.GetService<IAsyncResponseCallbackAuthorizer>(),
 311695            invocation.ServiceInterfaceFullName,
 311696            invocation.MethodName);
 697
 698        // The recovery callback may run in a different deployment than the original waiter, so
 699        // restore any ambient context captured at registration before resolving and invoking it.
 311700        using var contextScope = _propagation.Restore(context);
 311701        await serviceScope.ServiceProvider.InvokeAsync(invocation).ConfigureAwait(false);
 201702    }
 703
 704}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.DependencyInjection.IServiceScopeFactory,AsyncResponse.AsyncResponseContextPropagation,Microsoft.Extensions.Logging.ILogger,System.TimeProvider)
DispatchLostResponses()
DispatchLostExceptions()
ThrowUnsettled(System.Collections.Generic.List`1<System.Runtime.ExceptionServices.ExceptionDispatchInfo>,System.String)
SettleResidualFailures(System.Collections.Generic.List`1<System.Runtime.ExceptionServices.ExceptionDispatchInfo>,System.String,System.String,System.String)
DispatchLostResponse()
DispatchLostException()
DispatchToFailureCallback()
<DispatchToFailureCallback()
IsPermanentCallbackFailure(System.Exception)
DeleteConsumedRegistrationAsync()
WirePayload(T)
InvokeAsync()