| | | 1 | | using Microsoft.Extensions.Logging; |
| | | 2 | | using System.Diagnostics; |
| | | 3 | | |
| | | 4 | | namespace AsyncResponse; |
| | | 5 | | |
| | | 6 | | /// <summary> |
| | | 7 | | /// Transport-neutral ingress implementation. Broker/webhook adapters can feed response payloads |
| | | 8 | | /// and worker-job envelopes into this service without depending on a specific response channel. |
| | | 9 | | /// </summary> |
| | 2086 | 10 | | internal sealed class AsyncResponseIngress( |
| | 2086 | 11 | | IRawAsyncResponsePublisher _rawPublisher, |
| | 2086 | 12 | | IAsyncResponsePublisher _publisher, |
| | 2086 | 13 | | WorkerJobExecutor _workerJobExecutor, |
| | 2086 | 14 | | AsyncResponseContextPropagation _propagation, |
| | 2086 | 15 | | ILogger<AsyncResponseIngress> _logger, |
| | 2086 | 16 | | TimeProvider? _timeProvider = null, |
| | 2086 | 17 | | IAsyncResponseCallbackAuthorizer? _authorizer = null, |
| | 2086 | 18 | | Microsoft.Extensions.Options.IOptions<AsyncResponseOptions>? _options = null) : IAsyncResponseIngress |
| | | 19 | | { |
| | | 20 | | /// <inheritdoc /> |
| | | 21 | | public bool IsOverInboundBudget(string messageJson) |
| | 4311 | 22 | | => _options?.Value.MaxInboundMessageChars is { } limit |
| | 4311 | 23 | | && messageJson is not null |
| | 4311 | 24 | | && messageJson.Length > limit; |
| | | 25 | | |
| | | 26 | | /// <summary> |
| | | 27 | | /// Enforces <see cref="AsyncResponseOptions.MaxInboundMessageChars"/>. Returns <c>true</c> when |
| | | 28 | | /// the message was rejected, in which case the caller returns cleanly and the transport acks — |
| | | 29 | | /// see the option's remarks for why an oversized message is dropped rather than redelivered. |
| | | 30 | | /// Only the LENGTH is logged, never a prefix: an oversized body is still a body. |
| | | 31 | | /// </summary> |
| | | 32 | | private bool RejectIfOversized(string messageJson, string route, Activity? activity) |
| | | 33 | | { |
| | 4305 | 34 | | if (!IsOverInboundBudget(messageJson)) |
| | 4303 | 35 | | return false; |
| | | 36 | | |
| | 2 | 37 | | var limit = _options!.Value.MaxInboundMessageChars!.Value; |
| | | 38 | | |
| | 2 | 39 | | _logger.LogError( |
| | 2 | 40 | | "Ingress received an oversized {Route} message and acknowledged it without dispatch: {PayloadLength} UTF-16 |
| | 2 | 41 | | route, |
| | 2 | 42 | | messageJson.Length, |
| | 2 | 43 | | limit); |
| | 2 | 44 | | AsyncResponseDiagnostics.SetError( |
| | 2 | 45 | | activity, |
| | 2 | 46 | | "oversized_message", |
| | 2 | 47 | | $"Inbound {route} message exceeds the configured size budget of {limit} UTF-16 code units."); |
| | 2 | 48 | | AsyncResponseDiagnostics.RecordOversizedInboundMessage(route); |
| | 2 | 49 | | return true; |
| | | 50 | | } |
| | | 51 | | |
| | | 52 | | /// <summary>Handles the delivered message.</summary> |
| | | 53 | | public async Task HandleResponseMessageAsync(string messageJson, string? correlationId) |
| | | 54 | | { |
| | 129 | 55 | | using var activity = AsyncResponseDiagnostics.StartActivity( |
| | 129 | 56 | | "asyncresponse.ingress.response", |
| | 129 | 57 | | ActivityKind.Consumer, |
| | 129 | 58 | | correlationId); |
| | | 59 | | |
| | | 60 | | // An id extracted from an untrusted broker message is unroutable in two ways — missing |
| | | 61 | | // outright, or present but outside the portable contract (over-long, or space-padded, which |
| | | 62 | | // a relational store treats as the SAME key as the trimmed form while the library compares |
| | | 63 | | // ids ordinally, so storing a payload under it could surface it at another conversation's |
| | | 64 | | // waiter). Here they get one answer, which is the OPPOSITE of the answer a public publisher |
| | | 65 | | // gives: deliberately acknowledged, not thrown, because the message can never route and |
| | | 66 | | // redelivery would retry it forever (RabbitMQ's default MaxDeliveryAttempts = 0 has no cap) |
| | | 67 | | // or burn dead-letter attempts on brokers that do. Error-level log + counter make the drop |
| | | 68 | | // loud — every occurrence is a producer-side contract violation. The ACTIVITY carries the |
| | | 69 | | // routing context (trace id, the id as extracted); nothing about the body is logged, not |
| | | 70 | | // even a hash of it — see the note on payload metadata below. |
| | 129 | 71 | | if (RejectIfOversized(messageJson, "response", activity)) |
| | 0 | 72 | | return; |
| | | 73 | | |
| | 129 | 74 | | if (CorrelationIdGuard.IsUnroutable(correlationId, out var unroutable)) |
| | | 75 | | { |
| | 20 | 76 | | _logger.LogError( |
| | 20 | 77 | | "Ingress received a response message with an unusable correlation id ({UnroutableReason}); it cannot be |
| | 20 | 78 | | unroutable.Description, |
| | 20 | 79 | | messageJson.Length); |
| | 20 | 80 | | AsyncResponseDiagnostics.SetError(activity, unroutable.ErrorType, $"Inbound response message has an unusable |
| | 20 | 81 | | AsyncResponseDiagnostics.RecordUnroutableResponse(); |
| | 20 | 82 | | return; |
| | | 83 | | } |
| | | 84 | | |
| | | 85 | | try |
| | | 86 | | { |
| | | 87 | | // Correlation id and size, and deliberately nothing derived from the CONTENT. A hash |
| | | 88 | | // prefix looks like harmless metadata but is a content oracle: it is deterministic, so |
| | | 89 | | // equal payloads are visibly equal across messages and hosts, and a low-entropy payload |
| | | 90 | | // (a status enum, a small id, a boolean result) can be confirmed outright by hashing |
| | | 91 | | // the guesses. Trace and correlation ids already tie an entry to its conversation. |
| | 109 | 92 | | _logger.LogDebug( |
| | 109 | 93 | | "Ingress received an inbound response message for {CorrelationId}. Payload: {PayloadLength} UTF-16 code |
| | 109 | 94 | | correlationId, |
| | 109 | 95 | | messageJson.Length); |
| | | 96 | | |
| | | 97 | | // A transient infrastructure fault (channel store briefly unreachable, recovery-state |
| | | 98 | | // read hiccup, resume-callback dependency blip) must not finalize the waiter on the |
| | | 99 | | // first attempt — that would convert a recoverable response into a permanent business |
| | | 100 | | // failure. Retry briefly in-process before escalating. Parse failures are excluded: |
| | | 101 | | // an unparseable message never becomes parseable, so it escalates immediately. |
| | | 102 | | // Cancellation is excluded from BOTH the retry and the escalation below: it is not a |
| | | 103 | | // handler failure (a durable flow losing its execution lease mid-dispatch surfaces |
| | | 104 | | // here as an OperationCanceledException), so it propagates for the transport to |
| | | 105 | | // NAK/redeliver instead of terminally failing a waiter whose response was never lost. |
| | | 106 | | // Recovery resume callbacks may be re-invoked by these retries, which matches their |
| | | 107 | | // contract — broker redelivery re-invokes them the same way. |
| | | 108 | | // |
| | | 109 | | // RecoveryCallbackFailedException is excluded from both as well: the lost-subscriber |
| | | 110 | | // dispatcher already ran its own ladder against the failure callback, and escalating |
| | | 111 | | // through SetException would only invoke that same failing callback again. It |
| | | 112 | | // propagates so the transport redelivers the still-unacknowledged terminal signal. |
| | 109 | 113 | | await AsyncResponseRetry.ExecuteAsync( |
| | 109 | 114 | | async _ => |
| | 109 | 115 | | { |
| | 119 | 116 | | await _rawPublisher.SetRawResponseJson(messageJson, correlationId).ConfigureAwait(false); |
| | 93 | 117 | | return true; |
| | 93 | 118 | | }, |
| | 24 | 119 | | isTransient: static ex => ex is not (System.Text.Json.JsonException or InvalidDataException or Operation |
| | 109 | 120 | | maxAttempts: 4, |
| | 109 | 121 | | baseDelay: TimeSpan.FromMilliseconds(250), |
| | 109 | 122 | | maxDelay: TimeSpan.FromSeconds(2), |
| | 109 | 123 | | CancellationToken.None, |
| | 109 | 124 | | _timeProvider).ConfigureAwait(false); |
| | 93 | 125 | | } |
| | 16 | 126 | | catch (Exception ex) when (ex is not (OperationCanceledException or RecoveryCallbackFailedException)) |
| | | 127 | | { |
| | 8 | 128 | | _logger.LogError(ex, "Ingress failed to process the inbound response message."); |
| | 8 | 129 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | | 130 | | try |
| | | 131 | | { |
| | 8 | 132 | | await _publisher.SetException(ex, correlationId).ConfigureAwait(false); |
| | 6 | 133 | | } |
| | 2 | 134 | | catch (Exception innerEx) |
| | | 135 | | { |
| | 2 | 136 | | _logger.LogError(innerEx, "Ingress failed to publish the exception for the inbound message (original err |
| | | 137 | | |
| | | 138 | | // Both the publish and the SetException escalation failed, so returning normally |
| | | 139 | | // would ack a response that now exists nowhere. Propagate instead: the transport's |
| | | 140 | | // redelivery/dead-letter policy retries the whole pipeline, and the recovery |
| | | 141 | | // registration stays valid for the redelivered attempt. |
| | 2 | 142 | | throw; |
| | | 143 | | } |
| | | 144 | | } |
| | 119 | 145 | | } |
| | | 146 | | |
| | | 147 | | /// <summary>Handles the delivered message.</summary> |
| | | 148 | | public async Task HandleWorkerMessageAsync(string messageJson) |
| | | 149 | | { |
| | 4176 | 150 | | using var activity = AsyncResponseDiagnostics.StartActivity( |
| | 4176 | 151 | | "asyncresponse.ingress.worker", |
| | 4176 | 152 | | ActivityKind.Consumer); |
| | | 153 | | |
| | | 154 | | // Before the parse, so an oversized envelope never becomes a DOM. |
| | 4176 | 155 | | if (RejectIfOversized(messageJson, "worker", activity)) |
| | 2 | 156 | | return; |
| | | 157 | | |
| | | 158 | | WorkerJobEnvelope job; |
| | | 159 | | try |
| | | 160 | | { |
| | | 161 | | // The envelope is the WORST thing in the library to log whole: it carries the job's |
| | | 162 | | // arguments and whatever the context propagators captured (tenant, auth, trace baggage). |
| | | 163 | | // Size only, so a message that fails to even parse still leaves a trace, then the |
| | | 164 | | // routing metadata once it has been read. |
| | 4174 | 165 | | _logger.LogDebug("Ingress received a worker job. Payload: {PayloadLength} UTF-16 code units.", messageJson.L |
| | | 166 | | |
| | 4174 | 167 | | job = JsonSafety.SafeDeserialize<WorkerJobEnvelope>(messageJson) |
| | 4174 | 168 | | ?? throw new InvalidDataException("Worker message deserialized to null."); |
| | | 169 | | |
| | | 170 | | // `required` on WorkerJobEnvelope.Call enforces presence on the wire, not non-null: |
| | | 171 | | // an explicit "call": null parses successfully yet can never be executed, so it is |
| | | 172 | | // the same producer-side contract violation as an unparseable envelope. |
| | 4168 | 173 | | if (job.Call is null) |
| | 2 | 174 | | throw new InvalidDataException("Worker envelope carries a null call description."); |
| | | 175 | | |
| | | 176 | | // Same mechanism one level down: ReflectionCallDto's members are `required` too, so an |
| | | 177 | | // explicit "params": null (or a null element, or a null target name) parses yet can |
| | | 178 | | // never resolve to a callback — it must take this drop-and-ack route, not escape as an |
| | | 179 | | // ArgumentNullException the transport would redeliver forever. |
| | 4166 | 180 | | if (job.Call.ServiceInterfaceFullName is null || job.Call.MethodName is null || job.Call.Params is null) |
| | 6 | 181 | | throw new InvalidDataException("Worker envelope carries a call description with a null member."); |
| | 15898 | 182 | | foreach (var param in job.Call.Params) |
| | | 183 | | { |
| | 3790 | 184 | | if (param is null) |
| | 2 | 185 | | throw new InvalidDataException("Worker envelope carries a call description with a null parameter ent |
| | | 186 | | } |
| | 4158 | 187 | | } |
| | 16 | 188 | | catch (Exception ex) when (ex is InvalidDataException or System.Text.Json.JsonException) |
| | | 189 | | { |
| | | 190 | | // An envelope NO build can ever parse, which is the same class the response path above |
| | | 191 | | // acknowledges rather than throws — and for the same reason: redelivery would retry it |
| | | 192 | | // forever (RabbitMQ's default MaxDeliveryAttempts = 0 has no cap) or burn dead-letter |
| | | 193 | | // attempts on brokers that do. Error log + counter make the drop loud; every occurrence |
| | | 194 | | // is a producer-side contract violation. |
| | | 195 | | // |
| | | 196 | | // This filter must cover ONLY the parse above: the job body can throw the same |
| | | 197 | | // exception types (a durable flow deserializing a persisted input, a handler parsing a |
| | | 198 | | // third-party response), and those must propagate below for the transport to |
| | | 199 | | // redeliver/dead-letter instead of being acknowledged away as a malformed envelope. |
| | | 200 | | // |
| | | 201 | | // Deliberately NOT the unsupported-schema rejection, which stays a throw: that envelope |
| | | 202 | | // is well-formed and a NEWER build can read it, so refusing lets it reach one instead |
| | | 203 | | // of being acknowledged away mid-rolling-deploy. |
| | 16 | 204 | | _logger.LogError( |
| | 16 | 205 | | ex, |
| | 16 | 206 | | "Ingress received a worker envelope it cannot parse; it can never be executed and is acknowledged withou |
| | 16 | 207 | | messageJson.Length); |
| | 16 | 208 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 16 | 209 | | AsyncResponseDiagnostics.RecordWorkerOutcome("rejected"); |
| | 16 | 210 | | return; |
| | | 211 | | } |
| | | 212 | | |
| | | 213 | | try |
| | | 214 | | { |
| | 4158 | 215 | | AsyncResponseDiagnostics.SetCorrelationId(activity, job.CorrelationId); |
| | 4158 | 216 | | AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget); |
| | 4158 | 217 | | AsyncResponseDiagnostics.SetWorker(activity, job.Call); |
| | 4158 | 218 | | _logger.LogDebug( |
| | 4158 | 219 | | "Ingress worker job for {CorrelationId} targets {Service}.{Method}.", |
| | 4158 | 220 | | job.CorrelationId, |
| | 4158 | 221 | | job.Call.ServiceInterfaceFullName, |
| | 4158 | 222 | | job.Call.MethodName); |
| | | 223 | | |
| | | 224 | | // Authorize the target while the envelope is still inert data — BEFORE its propagated |
| | | 225 | | // context is restored. Both halves of this envelope are attacker-controlled to anyone |
| | | 226 | | // who can write to the worker transport: Call names the method to run, Context names |
| | | 227 | | // the ambient identity to run it under. Restoring Context first handed a custom |
| | | 228 | | // authorizer that consults ambient tenant/principal state the message's own answer to |
| | | 229 | | // the question it was about to be asked. ReflectionExtensions.InvokeAsync re-checks |
| | | 230 | | // downstream; this is the ordering, not the only gate. |
| | 4158 | 231 | | ReflectionExtensions.ThrowIfNotAuthorized( |
| | 4158 | 232 | | _authorizer, |
| | 4158 | 233 | | job.Call.ServiceInterfaceFullName ?? string.Empty, |
| | 4158 | 234 | | job.Call.MethodName ?? string.Empty); |
| | | 235 | | |
| | | 236 | | // The job crossed a serialization boundary (broker → ingress): restore any ambient |
| | | 237 | | // context its propagators captured before executing it. |
| | 4156 | 238 | | using (_propagation.Restore(job.Context)) |
| | 4156 | 239 | | await _workerJobExecutor.ExecuteAsync(job).ConfigureAwait(false); |
| | 4098 | 240 | | } |
| | 60 | 241 | | catch (Exception ex) |
| | | 242 | | { |
| | 60 | 243 | | _logger.LogError(ex, "Ingress worker job execution failed."); |
| | 60 | 244 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | | 245 | | |
| | | 246 | | // Propagate: the transport dispatcher owns the retry/dead-letter decision for worker |
| | | 247 | | // jobs (per its AckMode and MaxDeliveryAttempts). Swallowing here acknowledged failed |
| | | 248 | | // jobs as successes, which disabled redelivery entirely and left the waiter to burn |
| | | 249 | | // its full timeout. |
| | 60 | 250 | | throw; |
| | | 251 | | } |
| | 4116 | 252 | | } |
| | | 253 | | } |