| | | 1 | | using Microsoft.Extensions.DependencyInjection; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | using System.Diagnostics; |
| | | 4 | | |
| | | 5 | | namespace AsyncResponse; |
| | | 6 | | |
| | | 7 | | /// <summary> |
| | | 8 | | /// Executes <see cref="WorkerJobEnvelope"/>s: restores the correlation context and invokes the |
| | | 9 | | /// described service method through the DI container. Shared by the broker ingress |
| | | 10 | | /// (<see cref="IAsyncResponseIngress.HandleWorkerMessageAsync"/>) and the in-process worker |
| | | 11 | | /// transport, so every transport executes jobs identically. |
| | | 12 | | /// </summary> |
| | 2660 | 13 | | internal sealed class WorkerJobExecutor( |
| | 2660 | 14 | | IServiceScopeFactory _scopeFactory, |
| | 2660 | 15 | | ILogger<WorkerJobExecutor> _logger, |
| | 2660 | 16 | | IWorkerTransport? _workerTransport = null, |
| | 2660 | 17 | | TimeProvider? _timeProvider = null) |
| | | 18 | | { |
| | | 19 | | /// <summary> |
| | | 20 | | /// Tolerance for early delivery of a due-time-stamped job. Broker delay resolution is one |
| | | 21 | | /// second at best (SQS DelaySeconds, visibility timestamps), so re-publishing for a |
| | | 22 | | /// sub-second remainder would spin a delivery loop that can never catch the instant. |
| | | 23 | | /// </summary> |
| | 4 | 24 | | private static readonly TimeSpan NotBeforeTolerance = TimeSpan.FromSeconds(1); |
| | | 25 | | |
| | | 26 | | /// <summary> |
| | | 27 | | /// Minimum shrink of the remaining delay between two hops of the re-publish chain for the |
| | | 28 | | /// chain to count as making progress. A real hop shrinks the remainder by at least the |
| | | 29 | | /// broker's ~1s delay resolution; a hop redelivered with the SAME remainder means the gating |
| | | 30 | | /// clock disagrees with the stamping clock (skew) and re-publishing would loop forever. |
| | | 31 | | /// </summary> |
| | 4 | 32 | | private static readonly TimeSpan RedelayProgressEpsilon = TimeSpan.FromMilliseconds(500); |
| | | 33 | | |
| | | 34 | | /// <summary> |
| | | 35 | | /// Consecutive no-progress hops required before the stall fallback executes the job early. |
| | | 36 | | /// A single early redelivery can be a transient anomaly (an unhonored delay, a redrive |
| | | 37 | | /// surfacing the message) — executing on one sample would break the due-time contract by the |
| | | 38 | | /// whole remainder. Genuine skew stalls EVERY hop, so requiring a second consecutive stall |
| | | 39 | | /// keeps the anti-livelock property while a lone anomaly just re-publishes once more. |
| | | 40 | | /// </summary> |
| | | 41 | | private const int RedelayStallExecuteThreshold = 2; |
| | | 42 | | |
| | | 43 | | /// <summary> |
| | | 44 | | /// Executes the job. Exceptions propagate to the caller — transports decide whether to log, |
| | | 45 | | /// retry, or dead-letter. |
| | | 46 | | /// </summary> |
| | | 47 | | public async Task ExecuteAsync(WorkerJobEnvelope job) |
| | | 48 | | { |
| | 6320 | 49 | | ArgumentNullException.ThrowIfNull(job); |
| | | 50 | | |
| | | 51 | | // Armed only on the skew-proven early-execution path below; disposed with the invocation. |
| | 6318 | 52 | | IDisposable? forcedEarly = null; |
| | | 53 | | |
| | | 54 | | // Reject a job stamped with an unsupported schema rather than invoke a possibly-incompatible |
| | | 55 | | // method shape. Throwing routes the job through the transport's normal |
| | | 56 | | // failure/dead-letter handling. This is the single choke point every transport shares. |
| | 6318 | 57 | | if (!WorkerJobEnvelopeSchema.IsReadable(job.SchemaVersion)) |
| | | 58 | | { |
| | 6 | 59 | | _logger.LogWarning( |
| | 6 | 60 | | "Worker job for correlationId {CorrelationId} has unsupported schema version {SchemaVersion} (current: { |
| | 6 | 61 | | job.CorrelationId, job.SchemaVersion, WorkerJobEnvelopeSchema.Current); |
| | 6 | 62 | | AsyncResponseDiagnostics.RecordWorkerOutcome("rejected"); |
| | 6 | 63 | | throw new InvalidOperationException( |
| | 6 | 64 | | $"Worker job schema version {job.SchemaVersion} is not supported by this build " + |
| | 6 | 65 | | $"(current: {WorkerJobEnvelopeSchema.Current}) and cannot be executed safely."); |
| | | 66 | | } |
| | | 67 | | |
| | | 68 | | // The same portable-id contract the publishers enforce, applied to an id that arrived over |
| | | 69 | | // a broker. It has to happen HERE, before the redelay hop and before any handler runs: the |
| | | 70 | | // handler's implicit response publish would throw on this id, so the job would fail AFTER |
| | | 71 | | // its side effects and be redelivered to run them again. A null or blank id is left alone — |
| | | 72 | | // that is a fire-and-forget job, which has no response to publish. |
| | | 73 | | // |
| | | 74 | | // Drop, never throw — the same answer the ingress gives the identical id class on the |
| | | 75 | | // response path: the id can never become portable, so throwing turns the job into a |
| | | 76 | | // poison message that redelivers forever (RabbitMQ's default MaxDeliveryAttempts = 0 has |
| | | 77 | | // no cap) or burns dead-letter attempts on brokers that do. Returning cleanly lets the |
| | | 78 | | // transport ACK; the Error log + counter make the drop loud. |
| | 6312 | 79 | | if (!string.IsNullOrWhiteSpace(job.CorrelationId) |
| | 6312 | 80 | | && AsyncResponseChannelOptions.CorrelationIdNotPortable(job.CorrelationId) is { } rejection) |
| | | 81 | | { |
| | 8 | 82 | | _logger.LogError( |
| | 8 | 83 | | "Worker job carries a correlation id outside the portable contract; it cannot be executed and is acknowl |
| | 8 | 84 | | rejection); |
| | 8 | 85 | | AsyncResponseDiagnostics.RecordWorkerOutcome("rejected"); |
| | 8 | 86 | | return; |
| | | 87 | | } |
| | | 88 | | |
| | | 89 | | // Due-time guard, the shared half of delayed delivery (see IDelayedWorkerTransport): a job |
| | | 90 | | // delivered before its stamped due time — a chunked hop on a transport whose per-publish |
| | | 91 | | // delay is capped, or plain broker imprecision — is re-published for the remainder instead |
| | | 92 | | // of executed. Every transport funnels through here, so the chunk chain needs no |
| | | 93 | | // per-transport code. |
| | 6304 | 94 | | if (job.NotBeforeUtc is { } notBeforeUtc) |
| | | 95 | | { |
| | 109 | 96 | | var remaining = notBeforeUtc - (_timeProvider ?? TimeProvider.System).GetUtcNow().UtcDateTime; |
| | 109 | 97 | | if (remaining > NotBeforeTolerance) |
| | | 98 | | { |
| | | 99 | | // MaxPublishDelay <= zero: the capability is unavailable in the current |
| | | 100 | | // configuration (an SQS FIFO worker queue) — same as not implementing it. |
| | 24 | 101 | | if (_workerTransport is not IDelayedWorkerTransport delayedTransport |
| | 24 | 102 | | || delayedTransport.MaxPublishDelay <= TimeSpan.Zero) |
| | | 103 | | { |
| | | 104 | | // The job was published by a delayed-capable producer, but THIS consumer's |
| | | 105 | | // transport cannot re-delay it. Executing early would silently break the due |
| | | 106 | | // time; throwing routes it through normal retry/DLQ where it is visible. |
| | 0 | 107 | | throw new InvalidOperationException( |
| | 0 | 108 | | $"Worker job for correlationId {job.CorrelationId} is due at {notBeforeUtc:O} ({remaining} from |
| | 0 | 109 | | $"registered worker transport ({_workerTransport?.GetType().Name ?? "none"}) does not support de |
| | | 110 | | } |
| | | 111 | | |
| | | 112 | | // Progress check: on transports whose due time is gated by a different clock than |
| | | 113 | | // the one that stamped it (client-computed available_at / ScheduledEnqueueTime vs |
| | | 114 | | // the broker's own clock), a consumer running behind that clock is handed the job |
| | | 115 | | // back immediately and would re-publish the same remainder forever — each hop a |
| | | 116 | | // fresh message id, so no delivery counter ever reaches a DLQ. Executing early by |
| | | 117 | | // the skew beats never executing — but only after consecutive stalls prove the |
| | | 118 | | // skew is persistent, so a single anomalous early delivery cannot fire the job |
| | | 119 | | // arbitrarily ahead of its due time. |
| | | 120 | | // Both stall fields are wire values a foreign producer controls. The library only |
| | | 121 | | // ever stamps a strictly positive remainder, so a negative LastRedelayRemaining is |
| | | 122 | | // invalid (and TimeSpan.MinValue would overflow the checked subtraction below); |
| | | 123 | | // clamping the counter into [0, threshold] keeps a hostile int.MaxValue from |
| | | 124 | | // wrapping negative and disarming the stall fallback forever. |
| | 24 | 125 | | var stalled = job.LastRedelayRemaining is { } lastRemaining |
| | 24 | 126 | | && lastRemaining >= TimeSpan.Zero |
| | 24 | 127 | | && remaining >= lastRemaining - RedelayProgressEpsilon; |
| | 24 | 128 | | job.RedelayStallCount = stalled ? Math.Clamp(job.RedelayStallCount, 0, RedelayStallExecuteThreshold) + 1 |
| | | 129 | | |
| | 24 | 130 | | if (stalled && job.RedelayStallCount >= RedelayStallExecuteThreshold) |
| | | 131 | | { |
| | | 132 | | // The proof dies with this envelope. Anything the execution below re-publishes |
| | | 133 | | // is a NEW message whose stall counters start at zero, so a durable timer that |
| | | 134 | | // suspends again would rebuild the same proof from scratch on every lap and |
| | | 135 | | // never finish. The marker lets such a step wait out its remainder in process |
| | | 136 | | // instead — see WorkerJobSkewScope. |
| | 6 | 137 | | forcedEarly = WorkerJobSkewScope.Enter(); |
| | | 138 | | |
| | 6 | 139 | | _logger.LogWarning( |
| | 6 | 140 | | "Worker job {Target}.{Method} was redelivered {Remaining} before its due time {NotBeforeUtc} wit |
| | 6 | 141 | | "the publishing and delivery-gating clocks disagree (clock skew). Executing it now instead of re |
| | 6 | 142 | | job.Call.ServiceInterfaceFullName, job.Call.MethodName, remaining, notBeforeUtc, job.RedelayStal |
| | | 143 | | // No outcome recorded here: the execution below records exactly one outcome |
| | | 144 | | // ("executed"/"failed") for this delivery, like every other path. |
| | | 145 | | } |
| | | 146 | | else |
| | | 147 | | { |
| | 18 | 148 | | _logger.LogDebug( |
| | 18 | 149 | | "Worker job {Target}.{Method} delivered {Remaining} before its due time {NotBeforeUtc}; re-publi |
| | 18 | 150 | | job.Call.ServiceInterfaceFullName, job.Call.MethodName, remaining, notBeforeUtc); |
| | 18 | 151 | | AsyncResponseDiagnostics.RecordWorkerOutcome("redelayed"); |
| | | 152 | | |
| | 18 | 153 | | job.LastRedelayRemaining = remaining; |
| | 18 | 154 | | var hop = remaining <= delayedTransport.MaxPublishDelay ? remaining : delayedTransport.MaxPublishDel |
| | 18 | 155 | | await delayedTransport.PublishAsync(job, hop).ConfigureAwait(false); |
| | 18 | 156 | | return; |
| | | 157 | | } |
| | | 158 | | } |
| | | 159 | | } |
| | | 160 | | |
| | 6286 | 161 | | using var activity = AsyncResponseDiagnostics.StartActivity( |
| | 6286 | 162 | | "asyncresponse.worker.execute", |
| | 6286 | 163 | | ActivityKind.Consumer, |
| | 6286 | 164 | | job.CorrelationId); |
| | 6286 | 165 | | AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget); |
| | 6286 | 166 | | AsyncResponseDiagnostics.SetWorker(activity, job.Call); |
| | | 167 | | |
| | 6286 | 168 | | _logger.LogDebug("Executing worker job {Target}.{Method} (correlationId: {CorrelationId}, replyTarget: {ReplyTar |
| | | 169 | | |
| | | 170 | | try |
| | | 171 | | { |
| | | 172 | | // Scope the restored ambient context so one job cannot inherit or leak another job's |
| | | 173 | | // correlation id or reply target. |
| | 6286 | 174 | | using var asyncResponseScope = AsyncResponseContext.PushContext(job.CorrelationId, job.ReplyTarget); |
| | | 175 | | |
| | | 176 | | // The executing job itself, for the one handler that needs it: a durable-flow |
| | | 177 | | // execution records the job's identity with its lease, and re-publishes THIS job when |
| | | 178 | | // the broker redelivers it under a handler that is still running. Entered in this |
| | | 179 | | // frame — the one that awaits the invocation — because an AsyncLocal written inside a |
| | | 180 | | // callee never flows back here, and entered even for a job without an id so a job the |
| | | 181 | | // in-memory transport runs under its enqueuer's captured context never reads as the |
| | | 182 | | // job that published it. |
| | 6286 | 183 | | using var jobScope = WorkerJobScope.Enter(job); |
| | | 184 | | |
| | 6286 | 185 | | var invocation = ReflectionExtensions.ResolveCallback( |
| | 6286 | 186 | | job.Call, |
| | 6286 | 187 | | payload: null, |
| | 6286 | 188 | | exception: null, |
| | 6286 | 189 | | correlationId: job.CorrelationId); |
| | | 190 | | |
| | 6286 | 191 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 6270 | 192 | | await scope.ServiceProvider.InvokeAsync(invocation).ConfigureAwait(false); |
| | | 193 | | |
| | 6102 | 194 | | _logger.LogDebug("Executed worker job {Target}.{Method} successfully.", job.Call.ServiceInterfaceFullName, j |
| | 6102 | 195 | | AsyncResponseDiagnostics.RecordWorkerOutcome("executed"); |
| | 6102 | 196 | | } |
| | 170 | 197 | | catch (Exception ex) |
| | | 198 | | { |
| | 170 | 199 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 170 | 200 | | AsyncResponseDiagnostics.RecordWorkerOutcome("failed"); |
| | 170 | 201 | | throw; |
| | | 202 | | } |
| | | 203 | | finally |
| | | 204 | | { |
| | 6272 | 205 | | forcedEarly?.Dispose(); |
| | | 206 | | } |
| | 6128 | 207 | | } |
| | | 208 | | } |