| | | 1 | | using Microsoft.Extensions.Logging; |
| | | 2 | | using System.Diagnostics; |
| | | 3 | | using System.Security.Cryptography; |
| | | 4 | | using System.Text; |
| | | 5 | | |
| | | 6 | | namespace AsyncResponse; |
| | | 7 | | |
| | | 8 | | /// <summary> |
| | | 9 | | /// Transport-neutral ingress implementation. Broker/webhook adapters can feed response payloads |
| | | 10 | | /// and worker-job envelopes into this service without depending on a specific response channel. |
| | | 11 | | /// </summary> |
| | 3 | 12 | | internal sealed class AsyncResponseIngress( |
| | 3 | 13 | | IRawAsyncResponsePublisher _rawPublisher, |
| | 3 | 14 | | IAsyncResponsePublisher _publisher, |
| | 3 | 15 | | WorkerJobExecutor _workerJobExecutor, |
| | 3 | 16 | | AsyncResponseContextPropagation _propagation, |
| | 3 | 17 | | ILogger<AsyncResponseIngress> _logger) : IAsyncResponseIngress |
| | | 18 | | { |
| | | 19 | | /// <summary>Handles the delivered message.</summary> |
| | | 20 | | public async Task HandleResponseMessageAsync(string messageJson, string? correlationId) |
| | | 21 | | { |
| | 3 | 22 | | using var activity = AsyncResponseDiagnostics.StartActivity( |
| | 3 | 23 | | "asyncresponse.ingress.response", |
| | 3 | 24 | | ActivityKind.Consumer, |
| | 3 | 25 | | correlationId); |
| | | 26 | | |
| | 3 | 27 | | if (string.IsNullOrWhiteSpace(correlationId)) |
| | | 28 | | { |
| | | 29 | | // Deliberately acknowledged, not thrown: without a correlation id the message can |
| | | 30 | | // never route, so redelivery would retry it forever (RabbitMQ's default |
| | | 31 | | // MaxDeliveryAttempts = 0 has no cap) or burn dead-letter attempts on brokers that do. |
| | | 32 | | // Error-level log + counter make the drop loud — every occurrence is a producer-side |
| | | 33 | | // contract violation. Only metadata is logged: response payloads may carry PII and |
| | | 34 | | // stay out of logs by policy (docs/security.md); the byte length and hash prefix are |
| | | 35 | | // enough to correlate with broker-side capture tooling. |
| | 2 | 36 | | var payloadBytes = Encoding.UTF8.GetBytes(messageJson); |
| | 2 | 37 | | _logger.LogError( |
| | 2 | 38 | | "Ingress received a response message with no correlation id; it cannot be routed and is acknowledged wit |
| | 2 | 39 | | payloadBytes.Length, |
| | 2 | 40 | | Convert.ToHexString(SHA256.HashData(payloadBytes).AsSpan(0, 8))); |
| | 2 | 41 | | AsyncResponseDiagnostics.SetError(activity, "correlation_id_null", "No correlation id on the inbound respons |
| | 2 | 42 | | AsyncResponseDiagnostics.RecordUnroutableResponse(); |
| | 2 | 43 | | return; |
| | | 44 | | } |
| | | 45 | | |
| | | 46 | | try |
| | | 47 | | { |
| | 3 | 48 | | _logger.LogDebug("Ingress received inbound response message: {Message}", messageJson); |
| | | 49 | | |
| | | 50 | | // A transient infrastructure fault (channel store briefly unreachable, recovery-state |
| | | 51 | | // read hiccup, resume-callback dependency blip) must not finalize the waiter on the |
| | | 52 | | // first attempt — that would convert a recoverable response into a permanent business |
| | | 53 | | // failure. Retry briefly in-process before escalating. Parse failures are excluded: |
| | | 54 | | // an unparseable message never becomes parseable, so it escalates immediately. |
| | | 55 | | // Recovery resume callbacks may be re-invoked by these retries, which matches their |
| | | 56 | | // contract — broker redelivery re-invokes them the same way. |
| | 3 | 57 | | await AsyncResponseRetry.ExecuteAsync( |
| | 3 | 58 | | async _ => |
| | 3 | 59 | | { |
| | 3 | 60 | | await _rawPublisher.SetRawResponseJson(messageJson, correlationId).ConfigureAwait(false); |
| | 3 | 61 | | return true; |
| | 3 | 62 | | }, |
| | 3 | 63 | | isTransient: static ex => ex is not (System.Text.Json.JsonException or InvalidDataException), |
| | 3 | 64 | | maxAttempts: 4, |
| | 3 | 65 | | baseDelay: TimeSpan.FromMilliseconds(250), |
| | 3 | 66 | | maxDelay: TimeSpan.FromSeconds(2), |
| | 3 | 67 | | CancellationToken.None).ConfigureAwait(false); |
| | 3 | 68 | | } |
| | 3 | 69 | | catch (Exception ex) |
| | | 70 | | { |
| | 3 | 71 | | _logger.LogError(ex, "Ingress failed to process the inbound response message."); |
| | 3 | 72 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | | 73 | | try |
| | | 74 | | { |
| | 3 | 75 | | await _publisher.SetException(ex, correlationId).ConfigureAwait(false); |
| | 2 | 76 | | } |
| | 3 | 77 | | catch (Exception innerEx) |
| | | 78 | | { |
| | 3 | 79 | | _logger.LogError(innerEx, "Ingress failed to publish the exception for the inbound message (original err |
| | | 80 | | |
| | | 81 | | // Both the publish and the SetException escalation failed, so returning normally |
| | | 82 | | // would ack a response that now exists nowhere. Propagate instead: the transport's |
| | | 83 | | // redelivery/dead-letter policy retries the whole pipeline, and the recovery |
| | | 84 | | // registration stays valid for the redelivered attempt. |
| | 3 | 85 | | throw; |
| | | 86 | | } |
| | 2 | 87 | | } |
| | 3 | 88 | | } |
| | | 89 | | |
| | | 90 | | /// <summary>Handles the delivered message.</summary> |
| | | 91 | | public async Task HandleWorkerMessageAsync(string messageJson) |
| | | 92 | | { |
| | 3 | 93 | | using var activity = AsyncResponseDiagnostics.StartActivity( |
| | 3 | 94 | | "asyncresponse.ingress.worker", |
| | 3 | 95 | | ActivityKind.Consumer); |
| | | 96 | | |
| | | 97 | | try |
| | | 98 | | { |
| | 3 | 99 | | _logger.LogDebug("Ingress received worker job: {Payload}", messageJson); |
| | | 100 | | |
| | 3 | 101 | | var job = JsonSafety.SafeDeserialize<WorkerJobEnvelope>(messageJson) |
| | 3 | 102 | | ?? throw new InvalidDataException("Worker message deserialized to null."); |
| | 3 | 103 | | AsyncResponseDiagnostics.SetCorrelationId(activity, job.CorrelationId); |
| | 3 | 104 | | AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget); |
| | 3 | 105 | | AsyncResponseDiagnostics.SetWorker(activity, job.Call); |
| | | 106 | | |
| | | 107 | | // The job crossed a serialization boundary (broker → ingress): restore any ambient |
| | | 108 | | // context its propagators captured before executing it. |
| | 3 | 109 | | using (_propagation.Restore(job.Context)) |
| | 3 | 110 | | await _workerJobExecutor.ExecuteAsync(job).ConfigureAwait(false); |
| | 3 | 111 | | } |
| | 2 | 112 | | catch (Exception ex) |
| | | 113 | | { |
| | 2 | 114 | | _logger.LogError(ex, "Ingress worker job execution failed."); |
| | 2 | 115 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | | 116 | | |
| | | 117 | | // Propagate: the transport dispatcher owns the retry/dead-letter decision for worker |
| | | 118 | | // jobs (per its AckMode and MaxDeliveryAttempts). Swallowing here acknowledged failed |
| | | 119 | | // jobs as successes, which disabled redelivery entirely and left the waiter to burn |
| | | 120 | | // its full timeout. |
| | 2 | 121 | | throw; |
| | | 122 | | } |
| | 3 | 123 | | } |
| | | 124 | | } |