| | | 1 | | using Microsoft.Extensions.DependencyInjection; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | using System.Diagnostics; |
| | | 4 | | using System.Runtime.ExceptionServices; |
| | | 5 | | using System.Text.Json; |
| | | 6 | | |
| | | 7 | | namespace 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> |
| | 1308 | 23 | | internal 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> |
| | 410 | 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> |
| | 700 | 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> |
| | | 67 | | internal sealed class LostSubscriberCallbackDispatcher( |
| | | 68 | | IServiceScopeFactory _scopeFactory, |
| | | 69 | | AsyncResponseContextPropagation _propagation, |
| | | 70 | | ILogger _logger, |
| | | 71 | | 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 | | { |
| | | 85 | | 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. |
| | | 94 | | if (hasLiveSubscriber is not null && await hasLiveSubscriber().ConfigureAwait(false)) |
| | | 95 | | 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. |
| | | 99 | | var wirePayload = WirePayload(response); |
| | | 100 | | |
| | | 101 | | if (recoveryStates.Count == 0) |
| | | 102 | | return await DispatchLostResponse(null, wirePayload, channel).ConfigureAwait(false); |
| | | 103 | | |
| | | 104 | | var callbackInvoked = false; |
| | | 105 | | RecoveryAction? action = null; |
| | | 106 | | var routeSet = false; |
| | | 107 | | var routeMixed = false; |
| | | 108 | | List<ExceptionDispatchInfo>? failures = null; |
| | | 109 | | |
| | | 110 | | foreach (var recoveryState in recoveryStates) |
| | | 111 | | { |
| | | 112 | | try |
| | | 113 | | { |
| | | 114 | | var result = await DispatchLostResponse(recoveryState, wirePayload, channel).ConfigureAwait(false); |
| | | 115 | | if (!routeSet) |
| | | 116 | | { |
| | | 117 | | action = result.Action; |
| | | 118 | | routeSet = true; |
| | | 119 | | } |
| | | 120 | | else if (action != result.Action) |
| | | 121 | | { |
| | | 122 | | routeMixed = true; |
| | | 123 | | } |
| | | 124 | | |
| | | 125 | | if (!result.CallbackInvoked) |
| | | 126 | | continue; |
| | | 127 | | |
| | | 128 | | callbackInvoked = true; |
| | | 129 | | await DeleteConsumedRegistrationAsync(recoveryStateStore, correlationId, recoveryState.RegistrationId, c |
| | | 130 | | } |
| | | 131 | | catch (Exception ex) |
| | | 132 | | { |
| | | 133 | | if (ex is OperationCanceledException && cancellationToken.IsCancellationRequested) |
| | | 134 | | 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. |
| | | 139 | | (failures ??= []).Add(ExceptionDispatchInfo.Capture(ex)); |
| | | 140 | | } |
| | | 141 | | } |
| | | 142 | | |
| | | 143 | | if (failures is not null) |
| | | 144 | | { |
| | | 145 | | if (!callbackInvoked) |
| | | 146 | | ThrowUnsettled(failures, correlationId); |
| | | 147 | | |
| | | 148 | | SettleResidualFailures(failures, correlationId, channel, "response"); |
| | | 149 | | } |
| | | 150 | | |
| | | 151 | | return new LostSubscriberDispatchResult(routeMixed ? null : action, callbackInvoked) { RouteMixed = routeMixed } |
| | | 152 | | } |
| | | 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 | | { |
| | | 166 | | 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. |
| | | 172 | | if (hasLiveSubscriber is not null && await hasLiveSubscriber().ConfigureAwait(false)) |
| | | 173 | | return new LostSubscriberDispatchResult(RecoveryAction.Fail, false) { RetryLive = true }; |
| | | 174 | | |
| | | 175 | | if (recoveryStates.Count == 0) |
| | | 176 | | return new LostSubscriberDispatchResult(RecoveryAction.Fail, await DispatchLostException(null, exception, ch |
| | | 177 | | |
| | | 178 | | var callbackInvoked = false; |
| | | 179 | | List<ExceptionDispatchInfo>? failures = null; |
| | | 180 | | |
| | | 181 | | foreach (var recoveryState in recoveryStates) |
| | | 182 | | { |
| | | 183 | | try |
| | | 184 | | { |
| | | 185 | | if (!await DispatchLostException(recoveryState, exception, channel).ConfigureAwait(false)) |
| | | 186 | | continue; |
| | | 187 | | |
| | | 188 | | callbackInvoked = true; |
| | | 189 | | await DeleteConsumedRegistrationAsync(recoveryStateStore, correlationId, recoveryState.RegistrationId, c |
| | | 190 | | } |
| | | 191 | | catch (Exception ex) |
| | | 192 | | { |
| | | 193 | | if (ex is OperationCanceledException && cancellationToken.IsCancellationRequested) |
| | | 194 | | 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. |
| | | 199 | | (failures ??= []).Add(ExceptionDispatchInfo.Capture(ex)); |
| | | 200 | | } |
| | | 201 | | } |
| | | 202 | | |
| | | 203 | | if (failures is not null) |
| | | 204 | | { |
| | | 205 | | if (!callbackInvoked) |
| | | 206 | | ThrowUnsettled(failures, correlationId); |
| | | 207 | | |
| | | 208 | | SettleResidualFailures(failures, correlationId, channel, "exception"); |
| | | 209 | | } |
| | | 210 | | |
| | | 211 | | // Exception envelopes always take the failure route, so the action is fixed at Fail. |
| | | 212 | | return new LostSubscriberDispatchResult(RecoveryAction.Fail, callbackInvoked); |
| | | 213 | | } |
| | | 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 | | { |
| | | 227 | | foreach (var failure in failures) |
| | | 228 | | { |
| | | 229 | | if (failure.SourceException is RecoveryCallbackFailedException) |
| | | 230 | | 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. |
| | | 237 | | foreach (var failure in failures) |
| | | 238 | | { |
| | | 239 | | if (!IsPermanentCallbackFailure(failure.SourceException)) |
| | | 240 | | throw new RecoveryCallbackFailedException(correlationId, attempts: 1, failure.SourceException); |
| | | 241 | | } |
| | | 242 | | |
| | | 243 | | failures[0].Throw(); |
| | | 244 | | } |
| | | 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 | | { |
| | | 270 | | ExceptionDispatchInfo? exhausted = null; |
| | | 271 | | Exception? transient = null; |
| | | 272 | | foreach (var failure in failures) |
| | | 273 | | { |
| | | 274 | | var residual = failure.SourceException; |
| | | 275 | | |
| | | 276 | | // Already the propagating shape (a sibling's failure-callback ladder was exhausted); |
| | | 277 | | // its own ladder logged it. |
| | | 278 | | if (residual is RecoveryCallbackFailedException) |
| | | 279 | | { |
| | | 280 | | exhausted ??= failure; |
| | | 281 | | continue; |
| | | 282 | | } |
| | | 283 | | |
| | | 284 | | if (IsPermanentCallbackFailure(residual)) |
| | | 285 | | { |
| | | 286 | | _logger.LogError( |
| | | 287 | | residual, |
| | | 288 | | "Lost-{Kind} dispatch for correlationId {CorrelationId} on {Channel} failed with a deterministic fau |
| | | 289 | | kind, |
| | | 290 | | correlationId, |
| | | 291 | | channel); |
| | | 292 | | continue; |
| | | 293 | | } |
| | | 294 | | |
| | | 295 | | transient ??= residual; |
| | | 296 | | } |
| | | 297 | | |
| | | 298 | | if (exhausted is not null) |
| | | 299 | | exhausted.Throw(); |
| | | 300 | | |
| | | 301 | | if (transient is null) |
| | | 302 | | { |
| | | 303 | | _logger.LogError( |
| | | 304 | | "Lost-{Kind} dispatch for correlationId {CorrelationId} on {Channel} partially failed with deterministic |
| | | 305 | | kind, |
| | | 306 | | correlationId, |
| | | 307 | | channel); |
| | | 308 | | return; |
| | | 309 | | } |
| | | 310 | | |
| | | 311 | | _logger.LogError( |
| | | 312 | | transient, |
| | | 313 | | "Lost-{Kind} dispatch for correlationId {CorrelationId} on {Channel} partially failed transiently after anot |
| | | 314 | | kind, |
| | | 315 | | correlationId, |
| | | 316 | | channel); |
| | | 317 | | 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 | | { |
| | | 323 | | using var activity = AsyncResponseDiagnostics.StartActivity( |
| | | 324 | | "asyncresponse.lost_subscriber.dispatch", |
| | | 325 | | correlationId: recoveryState?.CorrelationId); |
| | | 326 | | activity?.SetTag("asyncresponse.lost_subscriber.kind", "response"); |
| | | 327 | | activity?.SetTag("asyncresponse.channel_name", channel); |
| | | 328 | | if (response is not null) |
| | | 329 | | 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. |
| | | 346 | | if (recoveryState?.PayloadTypeFullName is { } payloadTypeName |
| | | 347 | | && !AsyncResponseTypeResolution.IsWithinResolutionLimits(payloadTypeName)) |
| | | 348 | | { |
| | | 349 | | _logger.LogError( |
| | | 350 | | "The recovery registration for channel {Channel} names a payload type that is not resolved: {Payload |
| | | 351 | | channel, |
| | | 352 | | AsyncResponseTypeResolution.DescribeForDiagnostics(payloadTypeName)); |
| | | 353 | | } |
| | | 354 | | |
| | | 355 | | var classification = recoveryState is null |
| | | 356 | | ? default |
| | | 357 | | : PayloadRecoveryClassifier.Classify(response, recoveryState.PayloadTypeFullName); |
| | | 358 | | var action = classification.Action; |
| | | 359 | | var callbackPayload = classification.MaterializedPayload ?? (object?)response; |
| | | 360 | | AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, action); |
| | | 361 | | |
| | | 362 | | 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. |
| | | 371 | | _logger.LogInformation( |
| | | 372 | | "No subscribers for channel {Channel}; payload is a non-terminal checkpoint (KeepWaiting) — recovery |
| | | 373 | | channel); |
| | | 374 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", false); |
| | | 375 | | return new LostSubscriberDispatchResult(action, false); |
| | | 376 | | } |
| | | 377 | | |
| | | 378 | | if (action != RecoveryAction.Resume) |
| | | 379 | | { |
| | | 380 | | if (recoveryState is null) |
| | | 381 | | { |
| | | 382 | | _logger.LogWarning("No subscribers and no recovery state for channel {Channel}.", channel); |
| | | 383 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", false); |
| | | 384 | | return new LostSubscriberDispatchResult(action, false); |
| | | 385 | | } |
| | | 386 | | |
| | | 387 | | var invoked = await DispatchToFailureCallback(recoveryState, callbackPayload, channel, activity).Configu |
| | | 388 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", invoked); |
| | | 389 | | return new LostSubscriberDispatchResult(action, invoked); |
| | | 390 | | } |
| | | 391 | | |
| | | 392 | | // action == Resume implies recoveryState is non-null (the verdict is null otherwise). |
| | | 393 | | 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. |
| | | 400 | | if (recoveryState.FailureCallback != null) |
| | | 401 | | { |
| | | 402 | | _logger.LogWarning("No subscribers for channel {Channel}; payload is resumable but no resume callbac |
| | | 403 | | var fallbackInvoked = await DispatchToFailureCallback(recoveryState, callbackPayload, channel, activ |
| | | 404 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", fallbackInvoked); |
| | | 405 | | return new LostSubscriberDispatchResult(action, fallbackInvoked); |
| | | 406 | | } |
| | | 407 | | |
| | | 408 | | _logger.LogWarning("No subscribers for channel {Channel}; no resume callback available.", channel); |
| | | 409 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", false); |
| | | 410 | | return new LostSubscriberDispatchResult(action, false); |
| | | 411 | | } |
| | | 412 | | |
| | | 413 | | _logger.LogWarning("No subscribers for channel {Channel}; invoking resume callback.", channel); |
| | | 414 | | |
| | | 415 | | var invocation = ReflectionExtensions.ResolveCallback( |
| | | 416 | | recoveryState.ResumeCallback, |
| | | 417 | | payload: callbackPayload, |
| | | 418 | | exception: null, |
| | | 419 | | correlationId: recoveryState.CorrelationId |
| | | 420 | | ); |
| | | 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. |
| | | 425 | | await InvokeAsync(invocation, recoveryState.Context).ConfigureAwait(false); |
| | | 426 | | |
| | | 427 | | _logger.LogInformation("Resume callback invoked for channel {Channel}.", channel); |
| | | 428 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", true); |
| | | 429 | | |
| | | 430 | | return new LostSubscriberDispatchResult(action, true); |
| | | 431 | | } |
| | | 432 | | catch (Exception ex) |
| | | 433 | | { |
| | | 434 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | | 435 | | throw; |
| | | 436 | | } |
| | | 437 | | } |
| | | 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 | | { |
| | | 442 | | using var activity = AsyncResponseDiagnostics.StartActivity( |
| | | 443 | | "asyncresponse.lost_subscriber.dispatch", |
| | | 444 | | correlationId: recoveryState?.CorrelationId); |
| | | 445 | | activity?.SetTag("asyncresponse.lost_subscriber.kind", "exception"); |
| | | 446 | | activity?.SetTag("asyncresponse.channel_name", channel); |
| | | 447 | | activity?.SetTag("asyncresponse.exception_type", exception.GetType().FullName ?? exception.GetType().Name); |
| | | 448 | | AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, RecoveryAction.Fail); |
| | | 449 | | |
| | | 450 | | try |
| | | 451 | | { |
| | | 452 | | if (recoveryState?.FailureCallback == null) |
| | | 453 | | { |
| | | 454 | | _logger.LogWarning("No subscribers for channel {Channel}; no failure callback available.", channel); |
| | | 455 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", false); |
| | | 456 | | return false; |
| | | 457 | | } |
| | | 458 | | |
| | | 459 | | _logger.LogWarning("No subscribers for channel {Channel}; invoking failure callback.", channel); |
| | | 460 | | |
| | | 461 | | var invocation = ReflectionExtensions.ResolveCallback( |
| | | 462 | | recoveryState.FailureCallback, |
| | | 463 | | payload: null, |
| | | 464 | | exception: exception, |
| | | 465 | | correlationId: recoveryState.CorrelationId |
| | | 466 | | ); |
| | | 467 | | |
| | | 468 | | await InvokeAsync(invocation, recoveryState.Context).ConfigureAwait(false); |
| | | 469 | | |
| | | 470 | | _logger.LogInformation("Failure callback invoked for channel {Channel}.", channel); |
| | | 471 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", true); |
| | | 472 | | |
| | | 473 | | return true; |
| | | 474 | | } |
| | | 475 | | catch (Exception ex) |
| | | 476 | | { |
| | | 477 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | | 478 | | throw; |
| | | 479 | | } |
| | | 480 | | } |
| | | 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 | | { |
| | | 491 | | 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. |
| | | 496 | | payloadJson = response switch |
| | | 497 | | { |
| | | 498 | | string s => s, |
| | | 499 | | JsonElement je => je.GetRawText(), |
| | | 500 | | null => AsyncResponseJson.Serialize(response), |
| | | 501 | | _ => AsyncResponseJson.Serialize(response, response.GetType()) |
| | | 502 | | }; |
| | | 503 | | } |
| | | 504 | | 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. |
| | | 509 | | } |
| | | 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. |
| | | 516 | | if (recoveryState.FailureCallback == null) |
| | | 517 | | { |
| | | 518 | | _logger.LogError("No subscribers for channel {Channel} and the response declined to resume, but no failure c |
| | | 519 | | return false; |
| | | 520 | | } |
| | | 521 | | |
| | | 522 | | _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. |
| | | 527 | | var domainFailure = new AsyncResponseDomainFailureException( |
| | | 528 | | recoveryState.CorrelationId, |
| | | 529 | | recoveryState.PayloadTypeFullName is { } registeredTypeName |
| | | 530 | | ? AsyncResponseTypeResolution.DescribeForDiagnostics(registeredTypeName) |
| | | 531 | | : null, |
| | | 532 | | payloadJson); |
| | | 533 | | |
| | | 534 | | var invocation = ReflectionExtensions.ResolveCallback( |
| | | 535 | | recoveryState.FailureCallback, |
| | | 536 | | payload: response, |
| | | 537 | | exception: domainFailure, |
| | | 538 | | correlationId: recoveryState.CorrelationId |
| | | 539 | | ); |
| | | 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. |
| | | 555 | | await AsyncResponseRetry.ExecuteAsync( |
| | | 556 | | async _ => |
| | | 557 | | { |
| | | 558 | | await InvokeAsync(invocation, recoveryState.Context).ConfigureAwait(false); |
| | | 559 | | return true; |
| | | 560 | | }, |
| | | 561 | | isTransient: static ex => !IsPermanentCallbackFailure(ex), |
| | | 562 | | maxAttempts: FailureCallbackAttempts, |
| | | 563 | | baseDelay: TimeSpan.FromMilliseconds(250), |
| | | 564 | | maxDelay: TimeSpan.FromSeconds(2), |
| | | 565 | | CancellationToken.None, |
| | | 566 | | _timeProvider).ConfigureAwait(false); |
| | | 567 | | |
| | | 568 | | _logger.LogInformation("Failure callback invoked for channel {Channel}.", channel); |
| | | 569 | | |
| | | 570 | | return true; |
| | | 571 | | } |
| | | 572 | | catch (Exception ex) |
| | | 573 | | { |
| | | 574 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | | 575 | | |
| | | 576 | | 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. |
| | | 583 | | _logger.LogError(ex, "Failure callback for channel {Channel} cannot be invoked (deterministic fault); th |
| | | 584 | | 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. |
| | | 595 | | _logger.LogError( |
| | | 596 | | ex, |
| | | 597 | | "Failure callback for channel {Channel} failed on all {Attempts} attempts; the message is left unacknowl |
| | | 598 | | channel, |
| | | 599 | | FailureCallbackAttempts); |
| | | 600 | | throw new RecoveryCallbackFailedException(recoveryState.CorrelationId ?? string.Empty, FailureCallbackAttemp |
| | | 601 | | } |
| | | 602 | | } |
| | | 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) |
| | | 622 | | => exception is CallbackTargetUnresolvableException |
| | | 623 | | or MissingMethodException |
| | | 624 | | or MissingMemberException |
| | | 625 | | 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 | | { |
| | | 644 | | await recoveryStateStore.TryDeleteAsync(correlationId, registrationId, cancellationToken).ConfigureAwait(fal |
| | | 645 | | } |
| | | 646 | | catch (Exception ex) |
| | | 647 | | { |
| | | 648 | | _logger.LogWarning( |
| | | 649 | | ex, |
| | | 650 | | "Recovery callback for correlationId {CorrelationId} succeeded but deleting its registration {Registrati |
| | | 651 | | correlationId, |
| | | 652 | | registrationId); |
| | | 653 | | } |
| | | 654 | | } |
| | | 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<T></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 | | { |
| | | 667 | | if (response is null or JsonElement or string) |
| | | 668 | | return response; |
| | | 669 | | |
| | | 670 | | try |
| | | 671 | | { |
| | | 672 | | return AsyncResponseJson.Serialize(response); |
| | | 673 | | } |
| | | 674 | | 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. |
| | | 680 | | return response; |
| | | 681 | | } |
| | | 682 | | } |
| | | 683 | | |
| | | 684 | | private async Task InvokeAsync(ReflectionInvocationDto invocation, IReadOnlyDictionary<string, string>? context) |
| | | 685 | | { |
| | | 686 | | 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. |
| | | 693 | | ReflectionExtensions.ThrowIfNotAuthorized( |
| | | 694 | | serviceScope.ServiceProvider.GetService<IAsyncResponseCallbackAuthorizer>(), |
| | | 695 | | invocation.ServiceInterfaceFullName, |
| | | 696 | | 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. |
| | | 700 | | using var contextScope = _propagation.Restore(context); |
| | | 701 | | await serviceScope.ServiceProvider.InvokeAsync(invocation).ConfigureAwait(false); |
| | | 702 | | } |
| | | 703 | | |
| | | 704 | | } |