| | | 1 | | using Microsoft.Extensions.DependencyInjection; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | using Microsoft.Extensions.Options; |
| | | 4 | | using StackExchange.Redis; |
| | | 5 | | using System.Buffers; |
| | | 6 | | using System.Diagnostics; |
| | | 7 | | using System.Text; |
| | | 8 | | using System.Text.Json; |
| | | 9 | | |
| | | 10 | | namespace AsyncResponse.Channels.Redis; |
| | | 11 | | |
| | | 12 | | /// <summary> |
| | | 13 | | /// Redis-backed response channel: |
| | | 14 | | /// <list type="bullet"> |
| | | 15 | | /// <item><description>Publishes responses to Redis pub/sub channels keyed by correlation id.</description></item> |
| | | 16 | | /// <item><description>Subscribes waiters to those channels with per-channel serialized handling.</description></item> |
| | | 17 | | /// <item><description>Persists <see cref="RecoveryState"/> so responses arriving after the waiter |
| | | 18 | | /// died (e.g. a redeploy) are routed through the lost-subscriber dispatcher, which asks the payload's |
| | | 19 | | /// ShouldResumeOnRecovery and invokes the resume or failure callback.</description></item> |
| | | 20 | | /// </list> |
| | | 21 | | /// </summary> |
| | | 22 | | internal sealed class RedisAsyncResponseChannel : IAsyncResponsePublisher, IRawAsyncResponsePublisher, IRecoverableAsync |
| | | 23 | | { |
| | | 24 | | |
| | | 25 | | private readonly ISubscriber _subscriber; |
| | | 26 | | private readonly IRedisChannelSubscriber _channelSubscriber; |
| | | 27 | | private readonly IConnectionMultiplexer _multiplexer; |
| | | 28 | | private readonly IRecoveryStateStore _recoveryStateStore; |
| | | 29 | | private readonly AsyncResponseContextPropagation _propagation; |
| | | 30 | | private readonly LostSubscriberCallbackDispatcher _lostSubscriberDispatcher; |
| | | 31 | | private readonly RedisKeySchema _keys; |
| | | 32 | | private readonly RedisAsyncResponseOptions _options; |
| | | 33 | | private readonly ILogger<RedisAsyncResponseChannel> _logger; |
| | | 34 | | |
| | | 35 | | private readonly SerialExecutorRegistry _executors; |
| | | 36 | | |
| | | 37 | | /// <summary>Creates a Redis-backed async-response channel.</summary> |
| | 3 | 38 | | public RedisAsyncResponseChannel( |
| | 3 | 39 | | IServiceScopeFactory scopeFactory, |
| | 3 | 40 | | IConnectionMultiplexer multiplexer, |
| | 3 | 41 | | IRecoveryStateStore recoveryStateStore, |
| | 3 | 42 | | IOptions<RedisAsyncResponseOptions> options, |
| | 3 | 43 | | AsyncResponseContextPropagation propagation, |
| | 3 | 44 | | ILogger<RedisAsyncResponseChannel> logger, |
| | 3 | 45 | | IRedisChannelSubscriber? channelSubscriber = null) |
| | | 46 | | { |
| | 3 | 47 | | _subscriber = multiplexer.GetSubscriber(); |
| | 3 | 48 | | _channelSubscriber = channelSubscriber ?? new RedisChannelMessageQueueSubscriber(_subscriber); |
| | 3 | 49 | | _multiplexer = multiplexer; |
| | 3 | 50 | | _recoveryStateStore = recoveryStateStore; |
| | 3 | 51 | | _propagation = propagation; |
| | 3 | 52 | | _options = options.Value; |
| | 3 | 53 | | _options.ValidateShared(nameof(RedisAsyncResponseOptions)); |
| | 3 | 54 | | _keys = new RedisKeySchema(_options.KeyPrefix); |
| | 3 | 55 | | _logger = logger; |
| | 3 | 56 | | _lostSubscriberDispatcher = new LostSubscriberCallbackDispatcher(scopeFactory, propagation, logger); |
| | 3 | 57 | | _executors = new SerialExecutorRegistry(logger); |
| | 3 | 58 | | } |
| | | 59 | | |
| | | 60 | | // --------------------------------------------------------------------------------------- |
| | | 61 | | // IAsyncResponseSubscriber / IRecoverableAsyncResponseSubscriber |
| | | 62 | | |
| | | 63 | | /// <inheritdoc/> |
| | | 64 | | public Task<IAsyncResponseWaiter<T>> CreateResponseWaiter<T>( |
| | | 65 | | string correlationId, |
| | | 66 | | Func<T, ValueTask<bool>>? completionPredicate = null, |
| | | 67 | | TimeSpan? timeout = null) where T : IAsyncResponsePayload |
| | 3 | 68 | | => CreateResponseWaiterCore( |
| | 3 | 69 | | correlationId, |
| | 3 | 70 | | resumeCallback: null, |
| | 3 | 71 | | failureCallback: null, |
| | 3 | 72 | | completionPredicate, |
| | 3 | 73 | | timeout); |
| | | 74 | | |
| | | 75 | | /// <inheritdoc/> |
| | | 76 | | public Task<IAsyncResponseWaiter<T>> CreateRecoverableResponseWaiter<T>( |
| | | 77 | | string correlationId, |
| | | 78 | | ReflectionCallDto? resumeCallback = null, |
| | | 79 | | ReflectionCallDto? failureCallback = null, |
| | | 80 | | Func<T, ValueTask<bool>>? completionPredicate = null, |
| | | 81 | | TimeSpan? timeout = null) where T : IAsyncResponsePayload |
| | 3 | 82 | | => CreateResponseWaiterCore( |
| | 3 | 83 | | correlationId, |
| | 3 | 84 | | resumeCallback, |
| | 3 | 85 | | failureCallback, |
| | 3 | 86 | | completionPredicate, |
| | 3 | 87 | | timeout); |
| | | 88 | | |
| | | 89 | | private async Task<IAsyncResponseWaiter<T>> CreateResponseWaiterCore<T>( |
| | | 90 | | string correlationId, |
| | | 91 | | ReflectionCallDto? resumeCallback, |
| | | 92 | | ReflectionCallDto? failureCallback, |
| | | 93 | | Func<T, ValueTask<bool>>? completionPredicate, |
| | | 94 | | TimeSpan? timeout) where T : IAsyncResponsePayload |
| | | 95 | | { |
| | 3 | 96 | | if (string.IsNullOrWhiteSpace(correlationId)) |
| | 3 | 97 | | throw new ArgumentNullException(nameof(correlationId), "CorrelationId must not be empty or whitespace."); |
| | | 98 | | |
| | | 99 | | // Recovery callbacks only make sense if the payload can say whether a late response should |
| | | 100 | | // resume or fail the flow. On this durable channel that decision is real (it survives a |
| | | 101 | | // redeploy), so require the override rather than letting the conservative default silently |
| | | 102 | | // route every recovered response to the failure callback. The in-memory channel, which |
| | | 103 | | // cannot recover across a process restart, is deliberately not subject to this check. |
| | 3 | 104 | | if ((resumeCallback is not null || failureCallback is not null) |
| | 3 | 105 | | && !AsyncResponsePayloadReflection.OverridesShouldResumeOnRecovery(typeof(T))) |
| | | 106 | | { |
| | 3 | 107 | | throw new InvalidOperationException( |
| | 3 | 108 | | $"Payload type '{typeof(T)}' registers lost-subscriber recovery callbacks on the Redis channel " + |
| | 3 | 109 | | $"but does not override {nameof(IAsyncResponsePayload)}.{nameof(IAsyncResponsePayload.ShouldResumeOnReco |
| | 3 | 110 | | "Override it to declare which responses resume the flow (return true) versus fail it (return false); " + |
| | 3 | 111 | | "the durable channel needs this to route a response that arrives after the waiter was lost."); |
| | | 112 | | } |
| | | 113 | | |
| | | 114 | | // default: first envelope completes the wait |
| | 3 | 115 | | completionPredicate ??= _ => new ValueTask<bool>(true); |
| | | 116 | | |
| | | 117 | | // Default timeout aligned with the recovery-state expiry: an infinite wait is never |
| | | 118 | | // meaningful, because once the recovery state expires the correlation id has no recovery |
| | | 119 | | // anyway. Timing out routes the flow through its normal failure handling instead of |
| | | 120 | | // leaving it stuck forever. |
| | 3 | 121 | | timeout ??= _options.DefaultTimeout ?? _options.RecoveryStateExpiry; |
| | | 122 | | // BEFORE any side effect: an unsupported resolved timeout (non-positive, or past the |
| | | 123 | | // ~49.7-day BCL timer ceiling) used to throw only at timer arming — after the |
| | | 124 | | // subscription and recovery state existed, leaking both — and zero used to slip through |
| | | 125 | | // on some channels entirely, insta-timing-out a fully registered waiter. |
| | 3 | 126 | | AsyncResponseChannelOptions.EnsureWaiterTimeoutSupported(timeout.Value); |
| | | 127 | | |
| | 3 | 128 | | var storedCorrelationId = correlationId; |
| | | 129 | | // Capture the subscribe-time ExecutionContext so app AsyncLocals (trace, principal, logging |
| | | 130 | | // scope) flow into the message handler, which runs on a foreign Redis subscriber thread. |
| | 3 | 131 | | var capturedContext = ExecutionContext.Capture(); |
| | 3 | 132 | | var channel = _keys.Channel(correlationId); |
| | | 133 | | |
| | 3 | 134 | | var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.wait", correlationId: correlationId); |
| | 3 | 135 | | activity?.SetTag("asyncresponse.channel", "redis"); |
| | 3 | 136 | | AsyncResponseDiagnostics.SetPayloadType(activity, typeof(T)); |
| | 3 | 137 | | activity?.SetTag("asyncresponse.timeout_seconds", timeout.Value.TotalSeconds); |
| | | 138 | | |
| | 3 | 139 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 3 | 140 | | _logger.LogDebug("Waiting for response on correlationId {CorrelationId} with timeout {Timeout}.", correlatio |
| | | 141 | | |
| | 3 | 142 | | var tcs = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously); |
| | 3 | 143 | | var registrationId = Guid.NewGuid(); |
| | | 144 | | |
| | | 145 | | // Single-use cancellation token implementing the timeout. The timer is armed only |
| | | 146 | | // after subscribe + recovery-state save succeeds, but the callback is registered before |
| | | 147 | | // subscribing so a very fast terminal message can still clean up safely. |
| | 3 | 148 | | var cancellationTokenSource = new CancellationTokenSource(); |
| | 3 | 149 | | CancellationTokenRegistration timeoutRegistration = default; |
| | 3 | 150 | | IRedisChannelSubscription? subscription = null; |
| | 3 | 151 | | var executorRegistered = false; |
| | | 152 | | |
| | | 153 | | // ------------------------------------------------------------------------- |
| | | 154 | | // Local: CleanupOnceAsync |
| | | 155 | | // Ensures unsubscribe, recovery-state delete, timeout disposal, and executor cleanup |
| | | 156 | | // happen once no matter whether completion, timeout, or waiter disposal got there first. |
| | 3 | 157 | | int cleanupStarted = 0; |
| | 3 | 158 | | var cleanupGate = new object(); |
| | 3 | 159 | | Task? cleanupTask = null; |
| | | 160 | | |
| | | 161 | | // Task-latched so EVERY caller completes only when the one real cleanup has finished — |
| | | 162 | | // the previous fire-once int latch let a second caller (a disposing waiter racing the |
| | | 163 | | // timeout) return before the task was settled. |
| | | 164 | | ValueTask CleanupOnceAsync() |
| | | 165 | | { |
| | | 166 | | Task task; |
| | 3 | 167 | | lock (cleanupGate) |
| | | 168 | | { |
| | 3 | 169 | | task = cleanupTask ??= CleanupCoreAsync(); |
| | 3 | 170 | | } |
| | | 171 | | |
| | 3 | 172 | | return task.IsCompletedSuccessfully ? ValueTask.CompletedTask : new ValueTask(task); |
| | | 173 | | } |
| | | 174 | | |
| | | 175 | | // Dispose-path cleanup: DRAINS the per-channel serial executor before settling. A delivery |
| | | 176 | | // may be mid Until-predicate holding a claimed terminal message; the marker work item |
| | | 177 | | // completes only after that in-flight item finished, so by the time cleanup cancels, the |
| | | 178 | | // task is either settled by the delivery or genuinely undelivered — never a cancellation |
| | | 179 | | // stealing a consumed response. Must NOT be called from dispatch code (which runs ON the |
| | | 180 | | // executor): the dispatch-triggered cleanup uses CleanupOnceAsync directly, its task |
| | | 181 | | // already settled. |
| | | 182 | | // |
| | | 183 | | // The drain is bounded by DisposalDrainTimeout — one budget covering marker ADMISSION too |
| | | 184 | | // (a full bounded queue behind a wedged item blocks the enqueue itself). A lapsed budget |
| | | 185 | | // must not fall back to the cleanup's cancel: the wedged delivery holds a message already |
| | | 186 | | // consumed from the stream, and "canceled" would tell a re-attaching caller nothing was |
| | | 187 | | // delivered. It faults the task with the explicit indeterminate contract instead, routing |
| | | 188 | | // durable flows to a fresh idempotent restart. A tombstone-suppressed enqueue is the |
| | | 189 | | // opposite case — the retired executor finished everything it ever admitted, so nothing |
| | | 190 | | // is in flight and the plain cancel is truthful. |
| | | 191 | | async ValueTask DrainThenCleanupAsync() |
| | | 192 | | { |
| | 3 | 193 | | if (Volatile.Read(ref cleanupStarted) == 0 && executorRegistered) |
| | | 194 | | { |
| | 3 | 195 | | var drainTimeout = _options.DisposalDrainTimeout; |
| | 3 | 196 | | var drained = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); |
| | | 197 | | try |
| | | 198 | | { |
| | 3 | 199 | | using var budget = new CancellationTokenSource(drainTimeout); |
| | 3 | 200 | | var accepted = await _executors.EnqueueAsync(channel.ToString()!, () => |
| | 3 | 201 | | { |
| | 3 | 202 | | drained.TrySetResult(); |
| | 3 | 203 | | return Task.CompletedTask; |
| | 3 | 204 | | }, budget.Token).ConfigureAwait(false); |
| | 3 | 205 | | if (accepted) |
| | 3 | 206 | | await drained.Task.WaitAsync(budget.Token).ConfigureAwait(false); |
| | 3 | 207 | | } |
| | 1 | 208 | | catch (Exception drainEx) |
| | | 209 | | { |
| | | 210 | | // Budget lapse — or an unforeseen drain failure: either way the marker never |
| | | 211 | | // ran, so an in-flight delivery cannot be ruled out (only accepted=false |
| | | 212 | | // proves the executor finished everything). Settlement unproven means the |
| | | 213 | | // cleanup's cancel below would be a false "nothing was delivered" — fault |
| | | 214 | | // with the explicit indeterminate contract instead. A TrySetResult from the |
| | | 215 | | // late-finishing dispatch loses against this and is dropped; its cleanup |
| | | 216 | | // call is a no-op behind the latch. |
| | 1 | 217 | | _logger.LogWarning( |
| | 1 | 218 | | "Disposal drain for correlationId {CorrelationId} did not prove settlement within {DrainTimeout} |
| | 1 | 219 | | correlationId, drainTimeout); |
| | 1 | 220 | | AsyncResponseDiagnostics.SetError(activity, "indeterminate_delivery", "Disposal drain did not prove |
| | 1 | 221 | | if (drainEx is not OperationCanceledException) |
| | 1 | 222 | | _logger.LogDebug(drainEx, "Dispatch drain failed for channel {Channel}.", channel.ToString()!); |
| | 1 | 223 | | tcs.TrySetException(new AsyncResponseIndeterminateDeliveryException(correlationId, drainTimeout)); |
| | 1 | 224 | | } |
| | 3 | 225 | | } |
| | | 226 | | |
| | 3 | 227 | | await CleanupOnceAsync().ConfigureAwait(false); |
| | 3 | 228 | | } |
| | | 229 | | |
| | | 230 | | async Task CleanupCoreAsync() |
| | | 231 | | { |
| | 3 | 232 | | Interlocked.Exchange(ref cleanupStarted, 1); |
| | | 233 | | |
| | | 234 | | // A waiter disposed before any terminal signal must not leave ResponseTask pending |
| | | 235 | | // forever for callers that hold it directly — the timeout dies with this cleanup, so |
| | | 236 | | // nothing else could ever complete the task. Cancellation is a no-op after a normal |
| | | 237 | | // completion, timeout, or fault (and after a delivery drained by DrainThenCleanupAsync). |
| | 3 | 238 | | tcs.TrySetCanceled(); |
| | | 239 | | |
| | | 240 | | try |
| | | 241 | | { |
| | | 242 | | try |
| | | 243 | | { |
| | | 244 | | // Delete the recovery state BEFORE unsubscribing. In the reverse order a publish |
| | | 245 | | // landing in the window sees "no subscriber, state present" and fires a spurious |
| | | 246 | | // recovery callback for a wait that already reached a terminal state. In this |
| | | 247 | | // order the window shows a subscriber that drops the message — a late or duplicate |
| | | 248 | | // terminal message is droppable; a resurrected recovery callback is not. |
| | 3 | 249 | | await _recoveryStateStore.TryDeleteAsync(correlationId, registrationId).ConfigureAwait(false); |
| | 3 | 250 | | } |
| | 3 | 251 | | catch (Exception ex) |
| | | 252 | | { |
| | | 253 | | // Best-effort: the state expires on its own, and a transient store failure must |
| | | 254 | | // not skip the unsubscribe and executor teardown below. |
| | 3 | 255 | | _logger.LogError(ex, "Failed to delete recovery state for correlationId {CorrelationId}.", correlati |
| | 3 | 256 | | } |
| | | 257 | | |
| | | 258 | | try |
| | | 259 | | { |
| | | 260 | | // Bounded like the drain: this latched core is what a disposing waiter awaits |
| | | 261 | | // when terminal delivery started cleanup first, so an unbudgeted unsubscribe |
| | | 262 | | // would let a wedged client library hold DisposeAsync hostage past |
| | | 263 | | // DisposalDrainTimeout. The quiet wrapper logs its own failure — including |
| | | 264 | | // one that completes AFTER this wait was abandoned, which previously died as |
| | | 265 | | // a TaskScheduler.UnobservedTaskException nobody logged. |
| | 3 | 266 | | if (subscription is not null) |
| | 3 | 267 | | await UnsubscribeQuietlyAsync(subscription).WaitAsync(_options.DisposalDrainTimeout).ConfigureAw |
| | 3 | 268 | | } |
| | 1 | 269 | | catch (TimeoutException) |
| | | 270 | | { |
| | 1 | 271 | | _logger.LogError( |
| | 1 | 272 | | "Unsubscribe for channel {Channel} did not finish within {DisposalDrainTimeout}; abandoning the |
| | 1 | 273 | | channel.ToString()!, _options.DisposalDrainTimeout); |
| | 1 | 274 | | } |
| | | 275 | | } |
| | | 276 | | finally |
| | | 277 | | { |
| | | 278 | | // Purely local teardown runs no matter which network call above failed — the |
| | | 279 | | // cleanup latch is already set, so anything skipped here would leak until process |
| | | 280 | | // exit. |
| | 3 | 281 | | if (executorRegistered) |
| | 3 | 282 | | _executors.OnSubscriptionRetired(channel.ToString()!); |
| | | 283 | | |
| | | 284 | | // Schedule the disposal on the thread pool; do not await directly to prevent |
| | | 285 | | // deadlocks with work currently running on the executor. |
| | 3 | 286 | | _ = Task.Run(async () => |
| | 3 | 287 | | { |
| | 3 | 288 | | try |
| | 3 | 289 | | { |
| | 3 | 290 | | await _executors.RemoveAsync(channel.ToString()!).ConfigureAwait(false); |
| | 3 | 291 | | } |
| | 1 | 292 | | catch (Exception ex) |
| | 3 | 293 | | { |
| | 1 | 294 | | _logger.LogError(ex, "Failed to retire the executor for channel {Channel}.", channel.ToString()! |
| | 1 | 295 | | } |
| | 3 | 296 | | }); |
| | | 297 | | |
| | 3 | 298 | | await timeoutRegistration.DisposeAsync().ConfigureAwait(false); |
| | 3 | 299 | | cancellationTokenSource.Dispose(); |
| | 3 | 300 | | activity?.Dispose(); |
| | | 301 | | } |
| | | 302 | | } |
| | | 303 | | |
| | | 304 | | // Never faults: the unsubscribe outcome is logged HERE, so a teardown outliving the |
| | | 305 | | // bounded wait above still records its failure instead of surfacing as an unobserved |
| | | 306 | | // task exception. |
| | | 307 | | async Task UnsubscribeQuietlyAsync(IRedisChannelSubscription liveSubscription) |
| | | 308 | | { |
| | | 309 | | try |
| | | 310 | | { |
| | 3 | 311 | | await liveSubscription.DisposeAsync().ConfigureAwait(false); |
| | 3 | 312 | | _logger.LogDebug("Unsubscribed from channel {Channel}.", channel.ToString()!); |
| | 3 | 313 | | } |
| | 3 | 314 | | catch (Exception ex) |
| | | 315 | | { |
| | 3 | 316 | | _logger.LogError(ex, "Error during unsubscribe-once for channel {Channel}.", channel.ToString()!); |
| | 3 | 317 | | } |
| | 3 | 318 | | } |
| | | 319 | | |
| | | 320 | | // ------------------------------------------------------------------------- |
| | | 321 | | // Local: ProcessRedisMessageAsync |
| | | 322 | | // Deserializes and handles a single incoming envelope, completes the TCS when terminal. |
| | | 323 | | async Task ProcessRedisMessageAsync(RedisChannel messageChannel, RedisValue messageValue) |
| | | 324 | | { |
| | 3 | 325 | | _logger.LogDebug("Received message on channel {Channel}.", messageChannel.ToString()!); |
| | | 326 | | |
| | 3 | 327 | | bool finished = false; |
| | | 328 | | try |
| | | 329 | | { |
| | 3 | 330 | | var envelope = JsonSerializer.Deserialize(messageValue.ToString(), AsyncResponseEnvelopeJson.TypeInfo<T> |
| | | 331 | | |
| | 3 | 332 | | if (envelope == null) |
| | | 333 | | { |
| | 3 | 334 | | _logger.LogError("Failed to deserialize envelope for correlationId {CorrelationId}.", correlationId) |
| | | 335 | | |
| | 2 | 336 | | finished = true; |
| | 2 | 337 | | var deserializationError = new JsonException($"Failed to deserialize envelope for correlationId {cor |
| | 2 | 338 | | AsyncResponseDiagnostics.SetError(activity, "deserialize_failure", deserializationError.Message); |
| | 2 | 339 | | if (!tcs.TrySetException(deserializationError)) |
| | 2 | 340 | | _logger.LogWarning(deserializationError, "TaskCompletionSource already completed for correlation |
| | | 341 | | } |
| | 3 | 342 | | else if (!AsyncResponseEnvelopeSchema.IsReadable(envelope.SchemaVersion)) |
| | | 343 | | { |
| | 3 | 344 | | finished = true; |
| | 3 | 345 | | var schemaError = new InvalidOperationException( |
| | 3 | 346 | | $"Response envelope for correlationId {correlationId} has schema version {envelope.SchemaVersion |
| | 3 | 347 | | $"which this build does not support (current: {AsyncResponseEnvelopeSchema.Current})."); |
| | 2 | 348 | | AsyncResponseDiagnostics.SetError(activity, "schema_mismatch", schemaError.Message); |
| | 2 | 349 | | if (!tcs.TrySetException(schemaError)) |
| | 2 | 350 | | _logger.LogWarning(schemaError, "TaskCompletionSource already completed for correlationId {Corre |
| | | 351 | | } |
| | 3 | 352 | | else if (!envelope.Success) |
| | | 353 | | { |
| | 3 | 354 | | finished = true; |
| | 3 | 355 | | var remoteFailure = new Exception(envelope.ExceptionMessage ?? "Unknown error during asynchronous pr |
| | 3 | 356 | | if (!string.IsNullOrEmpty(envelope.ExceptionStackTrace)) |
| | | 357 | | { |
| | | 358 | | // Cap on receive too: the publish-side cap only bounds traces we emit, not what |
| | | 359 | | // a remote we do not control can push at us. |
| | 3 | 360 | | remoteFailure.Data["RemoteStackTrace"] = RemoteStackTrace.Cap(envelope.ExceptionStackTrace, _opt |
| | | 361 | | } |
| | | 362 | | |
| | 3 | 363 | | _logger.LogWarning("Received error response for correlationId {CorrelationId}: {ErrorMessage}", corr |
| | 3 | 364 | | AsyncResponseDiagnostics.SetError(activity, "remote_failure", remoteFailure.Message); |
| | 3 | 365 | | if (!tcs.TrySetException(remoteFailure)) |
| | 3 | 366 | | _logger.LogWarning(remoteFailure, "TaskCompletionSource already completed for correlationId {Cor |
| | | 367 | | } |
| | | 368 | | else |
| | | 369 | | { |
| | 3 | 370 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 3 | 371 | | _logger.LogDebug("Received response for correlationId {CorrelationId}.", correlationId); |
| | | 372 | | |
| | 3 | 373 | | finished = await completionPredicate(envelope.Payload!).ConfigureAwait(false); |
| | | 374 | | |
| | 3 | 375 | | if (finished && !tcs.TrySetResult(envelope.Payload!)) |
| | 3 | 376 | | _logger.LogWarning("TaskCompletionSource already completed for correlationId {CorrelationId}.", |
| | | 377 | | } |
| | 3 | 378 | | } |
| | 3 | 379 | | catch (Exception ex) |
| | | 380 | | { |
| | 3 | 381 | | _logger.LogError(ex, "Error processing message on channel {Channel} for correlationId {CorrelationId}.", |
| | | 382 | | |
| | 2 | 383 | | finished = true; |
| | 2 | 384 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 2 | 385 | | if (!tcs.TrySetException(ex)) |
| | 2 | 386 | | _logger.LogWarning(ex, "TaskCompletionSource already completed for correlationId {CorrelationId}.", |
| | 3 | 387 | | } |
| | | 388 | | finally |
| | | 389 | | { |
| | | 390 | | // Unsubscription also happens on dispose, but doing it immediately after the |
| | | 391 | | // terminal message releases resources sooner. |
| | 3 | 392 | | if (finished) |
| | 3 | 393 | | await CleanupOnceAsync().ConfigureAwait(false); |
| | | 394 | | } |
| | | 395 | | } |
| | | 396 | | |
| | | 397 | | // ------------------------------------------------------------------------- |
| | | 398 | | // Local: HandleMessageAsync |
| | | 399 | | // Receives pub/sub messages from the async subscription and enqueues them on the |
| | | 400 | | // per-channel executor, awaiting admission so executor backpressure reaches the |
| | | 401 | | // subscription's message loop instead of blocking a Redis reader thread. |
| | | 402 | | Task HandleMessageAsync(RedisChannel messageChannel, RedisValue messageValue) |
| | | 403 | | { |
| | | 404 | | // The registry coordinates create/enqueue/retire under one lock, so the message is never |
| | | 405 | | // enqueued onto an executor that is concurrently being torn down (no lost messages) and a |
| | | 406 | | // correlation-id reused mid-drain never produces two live executors for one channel. |
| | 3 | 407 | | var enqueue = _executors.EnqueueAsync( |
| | 3 | 408 | | messageChannel.ToString()!, |
| | 3 | 409 | | () => ProcessUnderCapturedContextAsync(messageChannel, messageValue)); |
| | 3 | 410 | | return enqueue.IsCompletedSuccessfully ? Task.CompletedTask : enqueue.AsTask(); |
| | | 411 | | } |
| | | 412 | | |
| | | 413 | | // ------------------------------------------------------------------------- |
| | | 414 | | // Local: ProcessUnderCapturedContextAsync |
| | | 415 | | // Restores the waiter's subscribe-time ExecutionContext (app AsyncLocals: trace, principal, |
| | | 416 | | // logging scope) plus the correlation id before processing — the Redis subscriber callback |
| | | 417 | | // runs on a foreign thread-pool thread that never had them. |
| | | 418 | | Task ProcessUnderCapturedContextAsync(RedisChannel messageChannel, RedisValue messageValue) |
| | | 419 | | { |
| | | 420 | | async Task ProcessAsync() |
| | | 421 | | { |
| | 3 | 422 | | using var correlationScope = AsyncResponseContext.PushCorrelationId(storedCorrelationId); |
| | 3 | 423 | | await ProcessRedisMessageAsync(messageChannel, messageValue).ConfigureAwait(false); |
| | 3 | 424 | | } |
| | | 425 | | |
| | 3 | 426 | | if (capturedContext is null) |
| | 3 | 427 | | return ProcessAsync(); |
| | | 428 | | |
| | 3 | 429 | | Task? task = null; |
| | 3 | 430 | | ExecutionContext.Run(capturedContext, _ => task = ProcessAsync(), null); |
| | 3 | 431 | | return task!; |
| | | 432 | | } |
| | | 433 | | |
| | 3 | 434 | | timeoutRegistration = cancellationTokenSource.Token.Register(() => |
| | 3 | 435 | | { |
| | 3 | 436 | | _ = Task.Run(async () => |
| | 3 | 437 | | { |
| | 3 | 438 | | _logger.LogWarning("Timed out waiting for response for correlationId {CorrelationId}.", correlationId); |
| | 3 | 439 | | AsyncResponseDiagnostics.SetError(activity, "timeout", $"Timed out waiting for response for correlationI |
| | 3 | 440 | | AsyncResponseDiagnostics.RecordWaiterTimeout("redis"); |
| | 3 | 441 | | tcs.TrySetException(new TimeoutException($"Timed out waiting for response for correlationId {correlation |
| | 3 | 442 | | await DrainThenCleanupAsync().ConfigureAwait(false); |
| | 3 | 443 | | }); |
| | 3 | 444 | | }); |
| | | 445 | | |
| | | 446 | | try |
| | | 447 | | { |
| | | 448 | | // Register the executor channel BEFORE the server-side SUBSCRIBE completes: the |
| | | 449 | | // subscriber attaches its message pump inside SubscribeAsync, so deliveries can start |
| | | 450 | | // before it returns, and on a correlation id reused within the tombstone lifetime the |
| | | 451 | | // registry would silently drop them as retirement stragglers until this registration |
| | | 452 | | // is visible. The subscribe-failure path below retires it again. |
| | 3 | 453 | | _executors.OnSubscriptionRegistered(channel.ToString()!); |
| | 3 | 454 | | executorRegistered = true; |
| | 3 | 455 | | subscription = await _channelSubscriber.SubscribeAsync(channel, HandleMessageAsync).ConfigureAwait(false); |
| | 3 | 456 | | var recoveryState = new RecoveryState |
| | 3 | 457 | | { |
| | 3 | 458 | | RegistrationId = registrationId, |
| | 3 | 459 | | ResumeCallback = resumeCallback, |
| | 3 | 460 | | FailureCallback = failureCallback, |
| | 3 | 461 | | CorrelationId = correlationId, |
| | 3 | 462 | | PayloadTypeFullName = typeof(T).FullName, |
| | 3 | 463 | | RegisteredAtUtc = DateTime.UtcNow, |
| | 3 | 464 | | Context = _propagation.Capture() |
| | 3 | 465 | | }; |
| | 3 | 466 | | await _recoveryStateStore.SaveAsync(correlationId, recoveryState, _options.RecoveryStateExpiry).ConfigureAwa |
| | 3 | 467 | | _logger.LogDebug("Subscribed to channel {Channel} for correlationId {CorrelationId}.", channel.ToString()!, |
| | 3 | 468 | | } |
| | 3 | 469 | | catch (Exception ex) |
| | | 470 | | { |
| | 3 | 471 | | _logger.LogError(ex, "Failed to subscribe to channel {Channel} for correlationId {CorrelationId}.", channel. |
| | 2 | 472 | | AsyncResponseDiagnostics.SetError(activity, "subscribe_failure", ex.Message); |
| | 2 | 473 | | await DrainThenCleanupAsync().ConfigureAwait(false); |
| | | 474 | | |
| | | 475 | | // Rethrow instead of returning a pre-faulted waiter: the builder's contract is that |
| | | 476 | | // the trigger runs only once the subscription AND recovery state exist. A returned |
| | | 477 | | // waiter would still let the trigger fire the remote operation with no registration |
| | | 478 | | // left to receive (or recover) its response. Cleanup leaves nothing behind and cancels |
| | | 479 | | // the response task rather than faulting it, so no unobserved fault lingers. |
| | 3 | 480 | | throw; |
| | | 481 | | } |
| | | 482 | | |
| | | 483 | | try |
| | | 484 | | { |
| | 3 | 485 | | if (Volatile.Read(ref cleanupStarted) == 0) |
| | 3 | 486 | | cancellationTokenSource.CancelAfter(timeout.Value); |
| | 3 | 487 | | } |
| | 1 | 488 | | catch (ObjectDisposedException) |
| | | 489 | | { |
| | | 490 | | // A response completed and cleaned up between the check and CancelAfter. |
| | 1 | 491 | | } |
| | | 492 | | |
| | 3 | 493 | | return new RedisAsyncResponseWaiter<T>(tcs.Task, DrainThenCleanupAsync); |
| | 3 | 494 | | } |
| | | 495 | | |
| | | 496 | | // --------------------------------------------------------------------------------------- |
| | | 497 | | // IAsyncResponsePublisher |
| | | 498 | | |
| | | 499 | | /// <inheritdoc/> |
| | | 500 | | public Task SetResponse<T>(T response, string correlationId, CancellationToken cancellationToken = default) where T |
| | 3 | 501 | | => SetResponseCore(response, correlationId, cancellationToken); |
| | | 502 | | |
| | | 503 | | Task IRawAsyncResponsePublisher.SetRawResponse(object? response, string correlationId, CancellationToken cancellatio |
| | 3 | 504 | | => SetResponseCore(response, correlationId, cancellationToken); |
| | | 505 | | |
| | | 506 | | Task IRawAsyncResponsePublisher.SetRawResponseJson(string responseJson, string correlationId, CancellationToken canc |
| | 3 | 507 | | => SetRawResponseJsonCore(responseJson, correlationId, cancellationToken); |
| | | 508 | | |
| | | 509 | | // Intentionally duplicated with SetRawResponseJsonCore: this publish method is a latency hot |
| | | 510 | | // path, and earlier shared helper/delegate refactors regressed throughput in benchmarks. |
| | | 511 | | // Keep the typed Redis path inline unless a benchmark run proves a refactor is free. |
| | | 512 | | private async Task SetResponseCore<T>(T response, string correlationId, CancellationToken cancellationToken) |
| | | 513 | | { |
| | 3 | 514 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.set_response", ActivityKind.Producer) |
| | 3 | 515 | | activity?.SetTag("asyncresponse.channel", "redis"); |
| | 3 | 516 | | AsyncResponseDiagnostics.SetPayloadType(activity, typeof(T)); |
| | | 517 | | |
| | | 518 | | // When no correlation id is provided, fall back to the ambient context. |
| | 3 | 519 | | AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId); |
| | | 520 | | |
| | 3 | 521 | | if (string.IsNullOrWhiteSpace(correlationId)) |
| | | 522 | | { |
| | 3 | 523 | | _logger.LogWarning("CorrelationId is null; cannot publish the response."); |
| | 2 | 524 | | AsyncResponseDiagnostics.SetError(activity, "correlation_id_null", "CorrelationId is null; cannot publish th |
| | 3 | 525 | | return; |
| | | 526 | | } |
| | | 527 | | |
| | 3 | 528 | | var channel = _keys.Channel(correlationId); |
| | | 529 | | try |
| | | 530 | | { |
| | 3 | 531 | | var envelope = new AsyncResponseEnvelope<T> |
| | 3 | 532 | | { |
| | 3 | 533 | | Success = true, |
| | 3 | 534 | | Payload = response |
| | 3 | 535 | | }; |
| | 3 | 536 | | var json = AsyncResponseEnvelopeJson.Serialize(envelope); |
| | 3 | 537 | | long numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false); |
| | 3 | 538 | | activity?.SetTag("asyncresponse.subscribers", numSubscribers); |
| | | 539 | | |
| | 3 | 540 | | if (numSubscribers == 0) |
| | | 541 | | { |
| | | 542 | | // Nobody was listening (the waiter died, e.g. with a redeploy): hand the response |
| | | 543 | | // over to the lost-subscriber dispatcher, which asks the payload whether to resume |
| | | 544 | | // the flow or fail it, and invokes the matching callback. |
| | 3 | 545 | | var dispatchResult = await _lostSubscriberDispatcher |
| | 3 | 546 | | .DispatchLostResponses( |
| | 3 | 547 | | _recoveryStateStore, |
| | 3 | 548 | | correlationId, |
| | 3 | 549 | | response, |
| | 3 | 550 | | channel.ToString()!, |
| | 3 | 551 | | cancellationToken, |
| | 3 | 552 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 3 | 553 | | .ConfigureAwait(false); |
| | 3 | 554 | | if (dispatchResult.RetryLive) |
| | | 555 | | { |
| | | 556 | | // A waiter subscribed between the publish and the recovery-state read — |
| | | 557 | | // re-publish live instead of consuming its registration; only a second miss |
| | | 558 | | // consumes it. |
| | 3 | 559 | | numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false); |
| | 2 | 560 | | activity?.SetTag("asyncresponse.subscribers", numSubscribers); |
| | 2 | 561 | | if (numSubscribers > 0) |
| | 2 | 562 | | return; |
| | | 563 | | |
| | 2 | 564 | | dispatchResult = await _lostSubscriberDispatcher |
| | 2 | 565 | | .DispatchLostResponses(_recoveryStateStore, correlationId, response, channel.ToString()!, cancel |
| | 2 | 566 | | .ConfigureAwait(false); |
| | | 567 | | } |
| | | 568 | | |
| | 3 | 569 | | AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, dispatchResult.ShouldResume); |
| | 3 | 570 | | AsyncResponseDiagnostics.RecordLostSubscriber("response", dispatchResult.ShouldResume, dispatchResult.Ca |
| | 3 | 571 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked); |
| | | 572 | | |
| | 3 | 573 | | await _executors.RemoveAsync(channel.ToString()!).ConfigureAwait(false); |
| | | 574 | | } |
| | | 575 | | else |
| | | 576 | | { |
| | 3 | 577 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 3 | 578 | | _logger.LogDebug("Published response for correlationId {CorrelationId} on channel {Channel}. Payload |
| | | 579 | | } |
| | 3 | 580 | | } |
| | 3 | 581 | | catch (Exception ex) |
| | | 582 | | { |
| | 3 | 583 | | _logger.LogError(ex, "Failed to publish response for correlationId {CorrelationId} on channel {Channel}.", c |
| | 2 | 584 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 3 | 585 | | throw; |
| | | 586 | | } |
| | 3 | 587 | | } |
| | | 588 | | |
| | | 589 | | // Intentionally duplicated with SetResponseCore: raw ingress uses pre-serialized payload JSON |
| | | 590 | | // and a different lost-subscriber materialization path, so avoiding shared indirection matters. |
| | | 591 | | private async Task SetRawResponseJsonCore(string responseJson, string correlationId, CancellationToken cancellationT |
| | | 592 | | { |
| | 3 | 593 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.ingress.raw_response", ActivityKind.P |
| | 3 | 594 | | activity?.SetTag("asyncresponse.channel", "redis"); |
| | | 595 | | |
| | 3 | 596 | | AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId); |
| | | 597 | | |
| | 3 | 598 | | if (string.IsNullOrWhiteSpace(correlationId)) |
| | | 599 | | { |
| | 3 | 600 | | _logger.LogWarning("CorrelationId is null; cannot publish the raw response."); |
| | 2 | 601 | | AsyncResponseDiagnostics.SetError(activity, "correlation_id_null", "CorrelationId is null; cannot publish th |
| | 3 | 602 | | return; |
| | | 603 | | } |
| | | 604 | | |
| | 3 | 605 | | var channel = _keys.Channel(correlationId); |
| | | 606 | | try |
| | | 607 | | { |
| | 3 | 608 | | var json = SerializeRawSuccessEnvelope(responseJson); |
| | 3 | 609 | | long numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false); |
| | 3 | 610 | | activity?.SetTag("asyncresponse.subscribers", numSubscribers); |
| | | 611 | | |
| | 3 | 612 | | if (numSubscribers == 0) |
| | | 613 | | { |
| | 3 | 614 | | var response = new RawJsonResponse(responseJson).DeserializeUntyped(); |
| | | 615 | | |
| | 2 | 616 | | var dispatchResult = await _lostSubscriberDispatcher |
| | 2 | 617 | | .DispatchLostResponses( |
| | 2 | 618 | | _recoveryStateStore, |
| | 2 | 619 | | correlationId, |
| | 2 | 620 | | response, |
| | 2 | 621 | | channel.ToString()!, |
| | 2 | 622 | | cancellationToken, |
| | 3 | 623 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 2 | 624 | | .ConfigureAwait(false); |
| | 2 | 625 | | if (dispatchResult.RetryLive) |
| | | 626 | | { |
| | | 627 | | // A waiter subscribed between the publish and the recovery-state read — |
| | | 628 | | // re-publish live instead of consuming its registration; only a second miss |
| | | 629 | | // consumes it. |
| | 2 | 630 | | numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false); |
| | 2 | 631 | | activity?.SetTag("asyncresponse.subscribers", numSubscribers); |
| | 2 | 632 | | if (numSubscribers > 0) |
| | 2 | 633 | | return; |
| | | 634 | | |
| | 2 | 635 | | dispatchResult = await _lostSubscriberDispatcher |
| | 2 | 636 | | .DispatchLostResponses(_recoveryStateStore, correlationId, response, channel.ToString()!, cancel |
| | 2 | 637 | | .ConfigureAwait(false); |
| | | 638 | | } |
| | | 639 | | |
| | 2 | 640 | | AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, dispatchResult.ShouldResume); |
| | 2 | 641 | | AsyncResponseDiagnostics.RecordLostSubscriber("response", dispatchResult.ShouldResume, dispatchResult.Ca |
| | 2 | 642 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked); |
| | | 643 | | |
| | 2 | 644 | | await _executors.RemoveAsync(channel.ToString()!).ConfigureAwait(false); |
| | 3 | 645 | | } |
| | | 646 | | else |
| | | 647 | | { |
| | 3 | 648 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 3 | 649 | | _logger.LogDebug("Published raw response for correlationId {CorrelationId} on channel {Channel}. Sub |
| | | 650 | | } |
| | 3 | 651 | | } |
| | 3 | 652 | | catch (Exception ex) |
| | | 653 | | { |
| | 3 | 654 | | _logger.LogError(ex, "Failed to publish raw response for correlationId {CorrelationId} on channel {Channel}. |
| | 3 | 655 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 3 | 656 | | throw; |
| | | 657 | | } |
| | 3 | 658 | | } |
| | | 659 | | |
| | | 660 | | /// <inheritdoc/> |
| | | 661 | | public async Task SetException(Exception exception, string correlationId, CancellationToken cancellationToken = defa |
| | | 662 | | { |
| | 3 | 663 | | ArgumentNullException.ThrowIfNull(exception); |
| | | 664 | | |
| | 3 | 665 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.set_exception", ActivityKind.Producer |
| | 3 | 666 | | activity?.SetTag("asyncresponse.channel", "redis"); |
| | 3 | 667 | | activity?.SetTag("asyncresponse.exception_type", exception.GetType().FullName ?? exception.GetType().Name); |
| | | 668 | | |
| | 3 | 669 | | AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId); |
| | | 670 | | |
| | 3 | 671 | | if (string.IsNullOrWhiteSpace(correlationId)) |
| | | 672 | | { |
| | 3 | 673 | | _logger.LogWarning("CorrelationId is null; cannot publish the exception. Exception: {ExceptionMessage}", exc |
| | 2 | 674 | | AsyncResponseDiagnostics.SetError(activity, "correlation_id_null", "CorrelationId is null; cannot publish th |
| | 3 | 675 | | return; |
| | | 676 | | } |
| | | 677 | | |
| | 3 | 678 | | var channel = _keys.Channel(correlationId); |
| | | 679 | | try |
| | | 680 | | { |
| | 3 | 681 | | var envelope = new AsyncResponseEnvelope<object> |
| | 3 | 682 | | { |
| | 3 | 683 | | Success = false, |
| | 3 | 684 | | ExceptionMessage = exception.Message, |
| | 3 | 685 | | ExceptionStackTrace = RemoteStackTrace.ForWire(exception.StackTrace, _options.IncludeRemoteStackTrace, _ |
| | 3 | 686 | | Payload = null |
| | 3 | 687 | | }; |
| | 3 | 688 | | var json = AsyncResponseEnvelopeJson.Serialize(envelope); |
| | 3 | 689 | | long numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false); |
| | 3 | 690 | | activity?.SetTag("asyncresponse.subscribers", numSubscribers); |
| | | 691 | | |
| | 3 | 692 | | if (numSubscribers == 0) |
| | | 693 | | { |
| | | 694 | | // Nobody was listening: exception envelopes always go to the failure callback. |
| | 3 | 695 | | var dispatchResult = await _lostSubscriberDispatcher |
| | 3 | 696 | | .DispatchLostExceptions( |
| | 3 | 697 | | _recoveryStateStore, |
| | 3 | 698 | | correlationId, |
| | 3 | 699 | | exception, |
| | 3 | 700 | | channel.ToString()!, |
| | 3 | 701 | | cancellationToken, |
| | 3 | 702 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 3 | 703 | | .ConfigureAwait(false); |
| | 3 | 704 | | if (dispatchResult.RetryLive) |
| | | 705 | | { |
| | | 706 | | // A waiter subscribed between the publish and the recovery-state read — |
| | | 707 | | // re-publish live instead of consuming its registration; only a second miss |
| | | 708 | | // consumes it. |
| | 3 | 709 | | numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false); |
| | 2 | 710 | | activity?.SetTag("asyncresponse.subscribers", numSubscribers); |
| | 2 | 711 | | if (numSubscribers > 0) |
| | 2 | 712 | | return; |
| | | 713 | | |
| | 2 | 714 | | dispatchResult = await _lostSubscriberDispatcher |
| | 2 | 715 | | .DispatchLostExceptions(_recoveryStateStore, correlationId, exception, channel.ToString()!, canc |
| | 2 | 716 | | .ConfigureAwait(false); |
| | | 717 | | } |
| | | 718 | | |
| | 3 | 719 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked); |
| | 3 | 720 | | AsyncResponseDiagnostics.RecordLostSubscriber("exception", shouldResume: false, dispatchResult.CallbackI |
| | | 721 | | |
| | 3 | 722 | | await _executors.RemoveAsync(channel.ToString()!).ConfigureAwait(false); |
| | | 723 | | } |
| | 3 | 724 | | else if (_logger.IsEnabled(LogLevel.Debug)) |
| | | 725 | | { |
| | 3 | 726 | | _logger.LogDebug("Published exception response for correlationId {CorrelationId} on channel {Channel}. S |
| | | 727 | | } |
| | 3 | 728 | | } |
| | 3 | 729 | | catch (Exception ex) |
| | | 730 | | { |
| | 3 | 731 | | _logger.LogError(ex, "Failed to publish exception response for correlationId {CorrelationId} on channel {Cha |
| | 3 | 732 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 3 | 733 | | throw; |
| | | 734 | | } |
| | 3 | 735 | | } |
| | | 736 | | |
| | | 737 | | // --------------------------------------------------------------------------------------- |
| | | 738 | | // IActiveSubscriberProbe |
| | | 739 | | |
| | | 740 | | /// <inheritdoc/> |
| | | 741 | | public ValueTask<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken = defau |
| | | 742 | | { |
| | 3 | 743 | | if (string.IsNullOrWhiteSpace(correlationId)) |
| | 3 | 744 | | return new ValueTask<long>(0L); |
| | | 745 | | |
| | 3 | 746 | | var channel = _keys.Channel(correlationId); |
| | | 747 | | |
| | | 748 | | // Subscriptions live on whichever node the client subscribed through, so the live count is |
| | | 749 | | // the maximum reported across all connected endpoints. |
| | 3 | 750 | | long subscribers = 0; |
| | 3 | 751 | | foreach (var endPoint in _multiplexer.GetEndPoints()) |
| | | 752 | | { |
| | 3 | 753 | | var server = _multiplexer.GetServer(endPoint); |
| | 3 | 754 | | if (!server.IsConnected) |
| | | 755 | | continue; |
| | | 756 | | |
| | | 757 | | try |
| | | 758 | | { |
| | 3 | 759 | | subscribers = Math.Max(subscribers, server.SubscriptionSubscriberCount(channel)); |
| | 3 | 760 | | } |
| | 3 | 761 | | catch (Exception ex) |
| | | 762 | | { |
| | 3 | 763 | | _logger.LogDebug(ex, "Failed to read subscriber count for channel {Channel}.", channel.ToString()!); |
| | 3 | 764 | | } |
| | | 765 | | } |
| | | 766 | | |
| | 3 | 767 | | return new ValueTask<long>(subscribers); |
| | | 768 | | } |
| | | 769 | | |
| | | 770 | | /// <summary> |
| | | 771 | | /// Re-probes waiter liveness for the lost-subscriber dispatcher's snapshot-race re-check, |
| | | 772 | | /// using the same PUBSUB NUMSUB-based probe the watchdog uses. |
| | | 773 | | /// </summary> |
| | | 774 | | private async ValueTask<bool> HasLiveSubscriberAsync(string correlationId, CancellationToken cancellationToken) |
| | 3 | 775 | | => await CountActiveSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false) > 0; |
| | | 776 | | |
| | | 777 | | private static string SerializeRawSuccessEnvelope(string payloadJson) |
| | | 778 | | { |
| | 3 | 779 | | JsonSafety.ThrowIfClearlyNotJson(payloadJson); |
| | | 780 | | |
| | 3 | 781 | | var buffer = new ArrayBufferWriter<byte>(); |
| | 3 | 782 | | using (var writer = new Utf8JsonWriter(buffer)) |
| | | 783 | | { |
| | 3 | 784 | | writer.WriteStartObject(); |
| | 3 | 785 | | writer.WriteNumber("SchemaVersion", AsyncResponseEnvelopeSchema.Current); |
| | 3 | 786 | | writer.WriteBoolean("Success", true); |
| | 3 | 787 | | writer.WritePropertyName("Payload"); |
| | 3 | 788 | | writer.WriteRawValue(payloadJson); |
| | 3 | 789 | | writer.WriteNull("ExceptionMessage"); |
| | 3 | 790 | | writer.WriteNull("ExceptionStackTrace"); |
| | 3 | 791 | | writer.WriteEndObject(); |
| | 3 | 792 | | } |
| | | 793 | | |
| | 3 | 794 | | return Encoding.UTF8.GetString(buffer.WrittenSpan); |
| | | 795 | | } |
| | | 796 | | } |