| | | 1 | | using Microsoft.Extensions.DependencyInjection; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | using System.Buffers; |
| | | 4 | | using System.Collections.Concurrent; |
| | | 5 | | using System.Diagnostics; |
| | | 6 | | using System.Runtime.CompilerServices; |
| | | 7 | | using System.Text; |
| | | 8 | | using System.Text.Json; |
| | | 9 | | using System.Threading.Channels; |
| | | 10 | | |
| | | 11 | | namespace AsyncResponse.Channels; |
| | | 12 | | |
| | | 13 | | // Shared source for the database-backed response channels (PostgreSQL, SQL Server, MongoDB), |
| | | 14 | | // mirroring the DurableFlows shared-store pattern: each channel csproj pulls this file in via |
| | | 15 | | // <Compile Include="..\Shared\DbChannelShared.cs" />, so the base class compiles INTO each |
| | | 16 | | // provider assembly against that provider's concrete seam types. The seam is bound per project |
| | | 17 | | // with three global using aliases (declared at the top of the provider's channel file): |
| | | 18 | | // |
| | | 19 | | // DbChannelStore -> the provider's store/SQL adapter (e.g. PostgreSqlChannelSql) |
| | | 20 | | // DbChannelMessage -> the provider's channel-message record (e.g. PostgreSqlChannelMessage) |
| | | 21 | | // DbChannelOptions -> the provider's options class (e.g. PostgreSqlAsyncResponseChannelOptions) |
| | | 22 | | // |
| | | 23 | | // Because the aliases resolve to concrete sealed types at compile time, store calls on the |
| | | 24 | | // per-message paths stay direct (no interface dispatch, no delegate indirection) — see the |
| | | 25 | | // benchmark note in RedisAsyncResponseChannel.SetResponseCore for why that matters. The only |
| | | 26 | | // virtual seams are the four hooks below, which cover exactly what the three providers genuinely |
| | | 27 | | // do differently: the channel-name format, the sweep cadence, the optional wake listener, and the |
| | | 28 | | // provider waiter type. |
| | | 29 | | |
| | | 30 | | /// <summary> |
| | | 31 | | /// Provider-agnostic machinery for the database-backed response channels: waiter registration and |
| | | 32 | | /// recovery-state bookkeeping, publish with delivery confirmation, the signal-driven dispatch |
| | | 33 | | /// sweep, the subscriber heartbeat, and subscription lifecycle/cleanup. Derived channels supply |
| | | 34 | | /// the wake mechanism (LISTEN/NOTIFY, adaptive polling, change streams), the channel-name format, |
| | | 35 | | /// and the provider waiter type via the protected hooks. |
| | | 36 | | /// </summary> |
| | | 37 | | internal abstract class DbAsyncResponseChannelBase : |
| | | 38 | | IAsyncResponsePublisher, |
| | | 39 | | IRawAsyncResponsePublisher, |
| | | 40 | | IRecoverableAsyncResponseSubscriber, |
| | | 41 | | IActiveSubscriberProbe, |
| | | 42 | | IAsyncDisposable |
| | | 43 | | { |
| | 419 | 44 | | private protected readonly ConcurrentDictionary<string, ConcurrentDictionary<Guid, IDbSubscription>> _subscriptions |
| | | 45 | | |
| | | 46 | | // A signal carries the correlation id to scan (targeted), or null to scan every subscribed |
| | | 47 | | // correlation id (the periodic sweep that is the missed-wake safety net). |
| | 419 | 48 | | private readonly Channel<string?> _signals = Channel.CreateBounded<string?>(new BoundedChannelOptions(1024) |
| | 419 | 49 | | { |
| | 419 | 50 | | SingleReader = true, |
| | 419 | 51 | | SingleWriter = false, |
| | 419 | 52 | | FullMode = BoundedChannelFullMode.DropOldest |
| | 419 | 53 | | }); |
| | | 54 | | |
| | | 55 | | // Maps a just-published message id to a completion the local dispatch loop trips the instant it |
| | | 56 | | // delivers the message to a live waiter. Same-process delivery (the overwhelmingly common case) |
| | | 57 | | // is confirmed without polling the database; cross-process delivery falls back to polling acked_at. |
| | 419 | 58 | | private readonly ConcurrentDictionary<Guid, TaskCompletionSource<bool>> _pendingConfirmations = new(); |
| | | 59 | | |
| | | 60 | | private protected readonly DbChannelStore _store; |
| | | 61 | | private readonly IRecoveryStateStore _recoveryStateStore; |
| | | 62 | | private readonly AsyncResponseContextPropagation _propagation; |
| | | 63 | | private readonly LostSubscriberCallbackDispatcher _lostSubscriberDispatcher; |
| | | 64 | | private protected readonly DbChannelOptions _options; |
| | | 65 | | private protected readonly ILogger _logger; |
| | | 66 | | private readonly SerialExecutorRegistry _executors; |
| | 419 | 67 | | private readonly string _instanceId = $"{Environment.MachineName}-{Environment.ProcessId}-{Guid.NewGuid():N}"; |
| | | 68 | | |
| | | 69 | | // Provider text used in diagnostics. The emitted strings must stay byte-identical to the |
| | | 70 | | // pre-consolidation per-provider channels — tests and dashboards match on them. |
| | | 71 | | private readonly string _channelTypeName; |
| | | 72 | | private readonly string _providerName; |
| | | 73 | | private readonly string _activityTag; |
| | | 74 | | private readonly string _subscriberRecordNoun; |
| | | 75 | | private readonly string _localDispatchRetryHint; |
| | | 76 | | |
| | 419 | 77 | | private readonly object _listenerGate = new(); |
| | | 78 | | private protected CancellationTokenSource? _listenerCts; |
| | | 79 | | private protected Task? _listenTask; |
| | | 80 | | private protected Task? _dispatchTask; |
| | | 81 | | private protected Task? _heartbeatTask; |
| | | 82 | | private bool _disposed; |
| | | 83 | | |
| | | 84 | | /// <summary>Creates the shared machinery for a database-backed async-response channel.</summary> |
| | 419 | 85 | | protected DbAsyncResponseChannelBase( |
| | 419 | 86 | | IServiceScopeFactory scopeFactory, |
| | 419 | 87 | | DbChannelStore store, |
| | 419 | 88 | | IRecoveryStateStore recoveryStateStore, |
| | 419 | 89 | | DbChannelOptions options, |
| | 419 | 90 | | AsyncResponseContextPropagation propagation, |
| | 419 | 91 | | ILogger logger, |
| | 419 | 92 | | string channelTypeName, |
| | 419 | 93 | | string providerName, |
| | 419 | 94 | | string activityTag, |
| | 419 | 95 | | string subscriberRecordNoun, |
| | 419 | 96 | | string localDispatchRetryHint, |
| | 419 | 97 | | TimeProvider? timeProvider = null) |
| | | 98 | | { |
| | 419 | 99 | | _store = store; |
| | 419 | 100 | | _recoveryStateStore = recoveryStateStore; |
| | 419 | 101 | | _propagation = propagation; |
| | 419 | 102 | | _options = options; |
| | 419 | 103 | | _options.Validate(); |
| | 419 | 104 | | _logger = logger; |
| | 419 | 105 | | _channelTypeName = channelTypeName; |
| | 419 | 106 | | _providerName = providerName; |
| | 419 | 107 | | _activityTag = activityTag; |
| | 419 | 108 | | _subscriberRecordNoun = subscriberRecordNoun; |
| | 419 | 109 | | _localDispatchRetryHint = localDispatchRetryHint; |
| | 419 | 110 | | _timeProvider = timeProvider ?? TimeProvider.System; |
| | 419 | 111 | | _lostSubscriberDispatcher = new LostSubscriberCallbackDispatcher(scopeFactory, propagation, logger, _timeProvide |
| | 419 | 112 | | _executors = new SerialExecutorRegistry(logger, timeProvider: _timeProvider); |
| | 419 | 113 | | } |
| | | 114 | | |
| | | 115 | | /// <summary> |
| | | 116 | | /// The engine's clock. Waiter timeouts and the delivery-confirmation wait arm on it rather |
| | | 117 | | /// than on the wall clock, so AsyncResponse.Testing's virtual clock can fire production-sized |
| | | 118 | | /// timeouts instantly here exactly as it already does on the in-memory channel — previously |
| | | 119 | | /// these were the only channels whose timeout paths a virtual-clock test could not reach. |
| | | 120 | | /// </summary> |
| | | 121 | | private protected readonly TimeProvider _timeProvider; |
| | | 122 | | |
| | | 123 | | /// <summary> |
| | | 124 | | /// The per-correlation channel name used as the serial-executor key and the lost-subscriber |
| | | 125 | | /// channel label. Formats differ per provider (notification channel, schema.table, collection). |
| | | 126 | | /// </summary> |
| | | 127 | | protected abstract string ChannelName(string correlationId); |
| | | 128 | | |
| | | 129 | | /// <summary> |
| | | 130 | | /// The dispatch sweep cadence. Fixed (<c>ListenerPollInterval</c>) for the providers with a push |
| | | 131 | | /// wake; adaptive (active/idle) for SQL Server where the sweep IS the cross-process wake. |
| | | 132 | | /// </summary> |
| | | 133 | | protected abstract TimeSpan CurrentPollInterval(); |
| | | 134 | | |
| | | 135 | | /// <summary> |
| | | 136 | | /// Starts the provider's wake listener loop (LISTEN/NOTIFY, change stream), or returns |
| | | 137 | | /// <c>null</c> when the provider has none and relies on the dispatch sweep alone. |
| | | 138 | | /// </summary> |
| | 364 | 139 | | protected virtual Task? StartWakeListener(CancellationToken cancellationToken) => null; |
| | | 140 | | |
| | | 141 | | /// <summary>Wraps the response task in the provider's waiter type.</summary> |
| | | 142 | | protected abstract IAsyncResponseWaiter<T> CreateWaiter<T>(Task<T> responseTask, Func<ValueTask> cleanupAsync) |
| | | 143 | | where T : IAsyncResponsePayload; |
| | | 144 | | |
| | | 145 | | /// <inheritdoc /> |
| | | 146 | | public Task<IAsyncResponseWaiter<T>> CreateResponseWaiter<T>( |
| | | 147 | | string correlationId, |
| | | 148 | | Func<T, ValueTask<bool>>? completionPredicate = null, |
| | | 149 | | TimeSpan? timeout = null) where T : IAsyncResponsePayload |
| | 242 | 150 | | => CreateResponseWaiterCore(correlationId, null, null, completionPredicate, timeout); |
| | | 151 | | |
| | | 152 | | /// <inheritdoc /> |
| | | 153 | | public Task<IAsyncResponseWaiter<T>> CreateRecoverableResponseWaiter<T>( |
| | | 154 | | string correlationId, |
| | | 155 | | ReflectionCallDto? resumeCallback = null, |
| | | 156 | | ReflectionCallDto? failureCallback = null, |
| | | 157 | | Func<T, ValueTask<bool>>? completionPredicate = null, |
| | | 158 | | TimeSpan? timeout = null) where T : IAsyncResponsePayload |
| | 228 | 159 | | => CreateResponseWaiterCore(correlationId, resumeCallback, failureCallback, completionPredicate, timeout); |
| | | 160 | | |
| | | 161 | | private async Task<IAsyncResponseWaiter<T>> CreateResponseWaiterCore<T>( |
| | | 162 | | string correlationId, |
| | | 163 | | ReflectionCallDto? resumeCallback, |
| | | 164 | | ReflectionCallDto? failureCallback, |
| | | 165 | | Func<T, ValueTask<bool>>? completionPredicate, |
| | | 166 | | TimeSpan? timeout) where T : IAsyncResponsePayload |
| | | 167 | | { |
| | 470 | 168 | | CorrelationIdGuard.ThrowIfUnusable(correlationId); |
| | | 169 | | |
| | 465 | 170 | | if ((resumeCallback is not null || failureCallback is not null) |
| | 465 | 171 | | && !AsyncResponsePayloadReflection.OverridesOnRecovery(typeof(T))) |
| | | 172 | | { |
| | 1 | 173 | | throw new InvalidOperationException( |
| | 1 | 174 | | $"Payload type '{typeof(T)}' registers lost-subscriber recovery callbacks on the {_providerName} channel |
| | 1 | 175 | | $"but does not override {nameof(IAsyncResponsePayload)}.{nameof(IAsyncResponsePayload.OnRecovery)}(). " |
| | 1 | 176 | | "Override it to declare what each response does to the flow — RecoveryAction.Resume, " + |
| | 1 | 177 | | "RecoveryAction.Fail, or RecoveryAction.KeepWaiting for non-terminal checkpoints; the durable " + |
| | 1 | 178 | | "channel needs this to route a response that arrives after the waiter was lost."); |
| | | 179 | | } |
| | | 180 | | |
| | 674 | 181 | | completionPredicate ??= _ => new ValueTask<bool>(true); |
| | 464 | 182 | | timeout ??= _options.DefaultTimeout ?? _options.RecoveryStateExpiry; |
| | | 183 | | // BEFORE any side effect: an unsupported resolved timeout (non-positive, or past the |
| | | 184 | | // ~49.7-day BCL timer ceiling) used to throw only at timer arming — after the |
| | | 185 | | // subscription and recovery state existed, leaking both — and zero used to slip through |
| | | 186 | | // on some channels entirely, insta-timing-out a fully registered waiter. |
| | 464 | 187 | | AsyncResponseChannelOptions.EnsureWaiterTimeoutSupported(timeout.Value); |
| | | 188 | | |
| | | 189 | | // Refuse BEFORE any store round trip: EnsureCreatedAsync now validates manually managed |
| | | 190 | | // schemas over the network, and a disposed channel must fail with ObjectDisposedException, |
| | | 191 | | // not with whatever that connection attempt throws. EnsureListenerStarted below re-checks |
| | | 192 | | // under the gate, so a dispose racing this early check still cannot start listeners. |
| | 461 | 193 | | ThrowIfDisposed(); |
| | 458 | 194 | | await _store.EnsureCreatedAsync().ConfigureAwait(false); |
| | 456 | 195 | | EnsureListenerStarted(); |
| | | 196 | | |
| | | 197 | | // Watermark from the database server's clock, not the app clock: the dispatch loop filters |
| | | 198 | | // pending messages with created_at >= started, and mixing an app-side timestamp with the |
| | | 199 | | // server-stamped created_at would silently drop live deliveries under clock skew. The |
| | | 200 | | // same round trip draws this subscription's position in the store's monotonic ack |
| | | 201 | | // sequence — the exact ordering IsWithinWatermark uses to separate "acked before this |
| | | 202 | | // waiter existed" (history) from "acked to a group including this waiter" (fan-out), |
| | | 203 | | // which no pair of same-tick timestamps can distinguish. |
| | 456 | 204 | | var (startedAtUtc, startedSeq) = await _store.GetSubscriptionStartAsync(CancellationToken.None).ConfigureAwait(f |
| | | 205 | | |
| | 456 | 206 | | var storedCorrelationId = correlationId; |
| | 456 | 207 | | var capturedContext = ExecutionContext.Capture(); |
| | | 208 | | |
| | 456 | 209 | | var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.wait", correlationId: correlationId); |
| | 456 | 210 | | activity?.SetTag("asyncresponse.channel", _activityTag); |
| | 456 | 211 | | AsyncResponseDiagnostics.SetPayloadType(activity, typeof(T)); |
| | 456 | 212 | | activity?.SetTag("asyncresponse.timeout_seconds", timeout.Value.TotalSeconds); |
| | | 213 | | |
| | 456 | 214 | | var tcs = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously); |
| | 456 | 215 | | var registrationId = Guid.NewGuid(); |
| | 456 | 216 | | var subscription = new DbSubscription<T>( |
| | 456 | 217 | | this, |
| | 456 | 218 | | correlationId, |
| | 456 | 219 | | registrationId, |
| | 456 | 220 | | startedAtUtc, |
| | 456 | 221 | | startedSeq, |
| | 456 | 222 | | completionPredicate, |
| | 456 | 223 | | tcs, |
| | 456 | 224 | | activity); |
| | | 225 | | |
| | | 226 | | // Clock-injected: CancelAfter on a default CTS is bound to the system clock, so a virtual |
| | | 227 | | // clock could never fire a production-sized waiter timeout on this channel. |
| | 456 | 228 | | var timeoutCts = new CancellationTokenSource(Timeout.InfiniteTimeSpan, _timeProvider); |
| | 456 | 229 | | CancellationTokenRegistration timeoutRegistration = default; |
| | 912 | 230 | | subscription.TimeoutRegistration = () => timeoutRegistration.DisposeAsync(); |
| | 456 | 231 | | subscription.TimeoutCancellation = timeoutCts; |
| | | 232 | | |
| | 456 | 233 | | timeoutRegistration = timeoutCts.Token.Register( |
| | 456 | 234 | | OnWaiterTimeout, |
| | 456 | 235 | | new WaiterTimeoutState<T>(this, subscription, activity, correlationId)); |
| | | 236 | | |
| | | 237 | | // Wire the captured-context delegate before the subscription becomes discoverable, so a |
| | | 238 | | // response already stored for this correlation id is processed with the caller's context. |
| | | 239 | | Task ProcessUnderCapturedContextAsync(DbChannelMessage message) |
| | | 240 | | { |
| | | 241 | | async Task Process() |
| | | 242 | | { |
| | 557 | 243 | | using var correlationScope = AsyncResponseContext.PushCorrelationId(storedCorrelationId); |
| | 557 | 244 | | await subscription.ProcessAsync(message).ConfigureAwait(false); |
| | 557 | 245 | | } |
| | | 246 | | |
| | 557 | 247 | | if (capturedContext is null) |
| | 0 | 248 | | return Process(); |
| | | 249 | | |
| | 557 | 250 | | Task? task = null; |
| | 1114 | 251 | | ExecutionContext.Run(capturedContext, _ => task = Process(), null); |
| | 557 | 252 | | return task!; |
| | | 253 | | } |
| | | 254 | | |
| | 456 | 255 | | subscription.ProcessUnderContextAsync = ProcessUnderCapturedContextAsync; |
| | | 256 | | |
| | | 257 | | try |
| | | 258 | | { |
| | 456 | 259 | | var recoveryState = new RecoveryState |
| | 456 | 260 | | { |
| | 456 | 261 | | RegistrationId = registrationId, |
| | 456 | 262 | | ResumeCallback = resumeCallback, |
| | 456 | 263 | | FailureCallback = failureCallback, |
| | 456 | 264 | | CorrelationId = correlationId, |
| | 456 | 265 | | PayloadTypeFullName = typeof(T).FullName, |
| | 456 | 266 | | // The SERVER-stamped subscription start, not the app clock. The watchdog judges |
| | 456 | 267 | | // staleness as "utcNow - RegisteredAtUtc" from whichever host scans, so an |
| | 456 | 268 | | // app-clock stamp made a skewed host's registrations either never age (skew |
| | 456 | 269 | | // ahead: a genuinely stuck flow stays invisible) or age instantly (skew behind: |
| | 456 | 270 | | // healthy waits page the operator). This is the same clock the delivery watermark |
| | 456 | 271 | | // above is drawn from, and for the same reason. |
| | 456 | 272 | | RegisteredAtUtc = startedAtUtc.UtcDateTime, |
| | 456 | 273 | | Context = _propagation.Capture() |
| | 456 | 274 | | }; |
| | | 275 | | // Subscriber record BEFORE recovery state: "recovery state visible ⇒ subscription |
| | | 276 | | // visible" is the invariant the lost-subscriber dispatcher's live re-check relies on. |
| | | 277 | | // In the reverse order a publisher could see the state, see no subscriber, and consume |
| | | 278 | | // the registration while this waiter is milliseconds from being live. |
| | 456 | 279 | | await _store.UpsertSubscriberAsync(correlationId, registrationId, _instanceId, _options.SubscriberHeartbeatT |
| | 456 | 280 | | await _recoveryStateStore.SaveAsync(correlationId, recoveryState, _options.RecoveryStateExpiry).ConfigureAwa |
| | | 281 | | |
| | 456 | 282 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 90 | 283 | | _logger.LogDebug("Waiting for {Provider} response on correlationId {CorrelationId} with timeout {Timeout |
| | 456 | 284 | | } |
| | 0 | 285 | | catch (Exception ex) |
| | | 286 | | { |
| | 0 | 287 | | _logger.LogError(ex, "Failed to create {Provider} waiter for correlationId {CorrelationId}.", _providerName, |
| | 0 | 288 | | AsyncResponseDiagnostics.SetError(activity, "subscribe_failure", ex.Message); |
| | 0 | 289 | | await subscription.DrainThenCleanupAsync(deleteRecoveryState: true).ConfigureAwait(false); |
| | | 290 | | |
| | | 291 | | // Rethrow instead of returning a pre-faulted waiter: the builder's contract is that |
| | | 292 | | // the trigger runs only once the subscription AND recovery state exist. A returned |
| | | 293 | | // waiter would still let the trigger fire the remote operation with no registration |
| | | 294 | | // left to receive (or recover) its response. Cleanup cancels the response task, so no |
| | | 295 | | // pending task is left behind. |
| | 0 | 296 | | throw; |
| | | 297 | | } |
| | | 298 | | |
| | | 299 | | // Publish the subscription only once it is fully armed (heartbeat + context delegate), |
| | | 300 | | // then signal a scan targeted at this correlation id so any already-stored response is |
| | | 301 | | // delivered promptly without a full sweep. |
| | 456 | 302 | | AddSubscription(correlationId, subscription); |
| | 456 | 303 | | SignalDispatcher(correlationId); |
| | | 304 | | |
| | | 305 | | // Arm the waiter timeout only AFTER the subscription is discoverable (Redis/NATS parity): |
| | | 306 | | // a timer that fired before AddSubscription would run cleanup against a map that does not |
| | | 307 | | // hold the entry yet, and the insert above would then pin a permanently-dropped |
| | | 308 | | // subscription (plus its executor registration) that nothing can ever remove again. |
| | | 309 | | try |
| | | 310 | | { |
| | 456 | 311 | | if (!subscription.CleanupStarted) |
| | 456 | 312 | | timeoutCts.CancelAfter(timeout.Value); |
| | 456 | 313 | | } |
| | 0 | 314 | | catch (ObjectDisposedException) |
| | | 315 | | { |
| | | 316 | | // A response completed and cleaned up between the check and CancelAfter. |
| | 0 | 317 | | } |
| | | 318 | | |
| | 915 | 319 | | return CreateWaiter<T>(tcs.Task, () => subscription.DrainThenCleanupAsync(deleteRecoveryState: true)); |
| | 456 | 320 | | } |
| | | 321 | | |
| | | 322 | | /// <inheritdoc /> |
| | | 323 | | public Task SetResponse<T>(T response, string correlationId, CancellationToken cancellationToken = default) where T |
| | 560 | 324 | | => SetResponseCore(response, correlationId, cancellationToken); |
| | | 325 | | |
| | | 326 | | Task IRawAsyncResponsePublisher.SetRawResponse(object? response, string correlationId, CancellationToken cancellatio |
| | 1 | 327 | | => SetResponseCore(response, correlationId, cancellationToken); |
| | | 328 | | |
| | | 329 | | Task IRawAsyncResponsePublisher.SetRawResponseJson(string responseJson, string correlationId, CancellationToken canc |
| | 23 | 330 | | => SetRawResponseJsonCore(responseJson, correlationId, cancellationToken); |
| | | 331 | | |
| | | 332 | | private async Task SetResponseCore<T>(T response, string correlationId, CancellationToken cancellationToken) |
| | | 333 | | { |
| | 561 | 334 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.set_response", ActivityKind.Producer) |
| | 561 | 335 | | activity?.SetTag("asyncresponse.channel", _activityTag); |
| | 561 | 336 | | AsyncResponseDiagnostics.SetPayloadType(activity, typeof(T)); |
| | 561 | 337 | | AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId); |
| | | 338 | | |
| | 561 | 339 | | if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the response")) |
| | 2 | 340 | | return; |
| | | 341 | | |
| | | 342 | | try |
| | | 343 | | { |
| | 556 | 344 | | var envelope = new AsyncResponseEnvelope<T> { Success = true, Payload = response }; |
| | 556 | 345 | | await PublishResponseWithRecoveryAsync( |
| | 556 | 346 | | activity, |
| | 556 | 347 | | correlationId, |
| | 556 | 348 | | AsyncResponseEnvelopeJson.Serialize(envelope), |
| | 556 | 349 | | typedResponse: response, |
| | 556 | 350 | | rawResponseJson: null, |
| | 556 | 351 | | cancellationToken).ConfigureAwait(false); |
| | 553 | 352 | | } |
| | 3 | 353 | | catch (Exception ex) |
| | | 354 | | { |
| | 3 | 355 | | _logger.LogError(ex, "Failed to publish {Provider} response for correlationId {CorrelationId}.", _providerNa |
| | 3 | 356 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 3 | 357 | | throw; |
| | | 358 | | } |
| | 555 | 359 | | } |
| | | 360 | | |
| | | 361 | | private async Task SetRawResponseJsonCore(string responseJson, string correlationId, CancellationToken cancellationT |
| | | 362 | | { |
| | 23 | 363 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.ingress.raw_response", ActivityKind.P |
| | 23 | 364 | | activity?.SetTag("asyncresponse.channel", _activityTag); |
| | 23 | 365 | | AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId); |
| | | 366 | | |
| | 23 | 367 | | if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the raw response", dropContractViolati |
| | 3 | 368 | | return; |
| | | 369 | | |
| | | 370 | | try |
| | | 371 | | { |
| | 20 | 372 | | await PublishResponseWithRecoveryAsync<object>( |
| | 20 | 373 | | activity, |
| | 20 | 374 | | correlationId, |
| | 20 | 375 | | SerializeRawSuccessEnvelope(responseJson), |
| | 20 | 376 | | typedResponse: null, |
| | 20 | 377 | | rawResponseJson: responseJson, |
| | 20 | 378 | | cancellationToken).ConfigureAwait(false); |
| | 17 | 379 | | } |
| | 3 | 380 | | catch (Exception ex) |
| | | 381 | | { |
| | 3 | 382 | | _logger.LogError(ex, "Failed to publish {Provider} raw response for correlationId {CorrelationId}.", _provid |
| | 3 | 383 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 3 | 384 | | throw; |
| | | 385 | | } |
| | 20 | 386 | | } |
| | | 387 | | |
| | | 388 | | /// <summary> |
| | | 389 | | /// The publish-with-recovery protocol shared by <see cref="SetResponseCore{T}"/> and |
| | | 390 | | /// <see cref="SetRawResponseJsonCore"/>, which carried lockstep copies of it (and the raw |
| | | 391 | | /// copy parsed its body twice when the RetryLive branch fell through to a failed delivery |
| | | 392 | | /// confirmation). The recovery payload is <paramref name="typedResponse"/> when |
| | | 393 | | /// <paramref name="rawResponseJson"/> is null; otherwise the raw body is deserialized |
| | | 394 | | /// lazily — once — on the cold branches that dispatch to recovery, so the delivered-live |
| | | 395 | | /// path never parses it. Generic so the typed path hands the dispatcher the publisher's |
| | | 396 | | /// DECLARED <typeparamref name="TPayload"/> — erasing to <c>object</c> made the recovery |
| | | 397 | | /// wire form the RUNTIME-type serialization, diverging from the declared-type envelope this |
| | | 398 | | /// method just wrote (and from Redis/NATS/in-memory) whenever the runtime type carries |
| | | 399 | | /// members the declared contract does not. |
| | | 400 | | /// </summary> |
| | | 401 | | private async Task PublishResponseWithRecoveryAsync<TPayload>( |
| | | 402 | | Activity? activity, |
| | | 403 | | string correlationId, |
| | | 404 | | string envelopeJson, |
| | | 405 | | TPayload? typedResponse, |
| | | 406 | | string? rawResponseJson, |
| | | 407 | | CancellationToken cancellationToken) |
| | | 408 | | { |
| | 576 | 409 | | object? rawRecoveryPayload = null; |
| | 576 | 410 | | var rawRecoveryPayloadMaterialized = false; |
| | | 411 | | |
| | | 412 | | // One dispatch shape per payload source, so generic inference binds the declared type on |
| | | 413 | | // the typed path and object on the raw path — never object for both. |
| | | 414 | | Task<LostSubscriberDispatchResult> DispatchToRecoveryAsync(Func<ValueTask<bool>>? hasLiveSubscriber) |
| | | 415 | | { |
| | 21 | 416 | | if (rawResponseJson is null) |
| | | 417 | | { |
| | 6 | 418 | | return _lostSubscriberDispatcher.DispatchLostResponses( |
| | 6 | 419 | | _recoveryStateStore, correlationId, typedResponse, ChannelName(correlationId), cancellationToken, ha |
| | | 420 | | } |
| | | 421 | | |
| | 15 | 422 | | if (!rawRecoveryPayloadMaterialized) |
| | | 423 | | { |
| | 15 | 424 | | rawRecoveryPayload = new RawJsonResponse(rawResponseJson).DeserializeUntyped(); |
| | 15 | 425 | | rawRecoveryPayloadMaterialized = true; |
| | | 426 | | } |
| | | 427 | | |
| | 15 | 428 | | return _lostSubscriberDispatcher.DispatchLostResponses( |
| | 15 | 429 | | _recoveryStateStore, correlationId, rawRecoveryPayload, ChannelName(correlationId), cancellationToken, h |
| | | 430 | | } |
| | | 431 | | |
| | 576 | 432 | | var subscribers = await _store.CountActiveSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(fals |
| | 570 | 433 | | activity?.SetTag("asyncresponse.subscribers", subscribers); |
| | 570 | 434 | | if (subscribers <= 0) |
| | | 435 | | { |
| | 19 | 436 | | var dispatchResult = await DispatchToRecoveryAsync( |
| | 19 | 437 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 19 | 438 | | .ConfigureAwait(false); |
| | 19 | 439 | | if (!dispatchResult.RetryLive) |
| | | 440 | | { |
| | 19 | 441 | | AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, dispatchResult.Action, dispatchResult.RouteMix |
| | 19 | 442 | | AsyncResponseDiagnostics.RecordLostSubscriber("response", dispatchResult.Action, dispatchResult.Callback |
| | 19 | 443 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked); |
| | 19 | 444 | | return; |
| | | 445 | | } |
| | | 446 | | |
| | | 447 | | // A waiter registered between the count and the recovery-state read — publish live |
| | | 448 | | // instead of consuming its registration. |
| | | 449 | | } |
| | | 450 | | |
| | 551 | 451 | | var messageId = Guid.NewGuid(); |
| | 551 | 452 | | using var confirmation = BeginConfirmation(messageId); |
| | 551 | 453 | | await PublishMessageAsync(messageId, correlationId, envelopeJson, cancellationToken).ConfigureAwait(false); |
| | | 454 | | |
| | 551 | 455 | | if (!await TryConfirmDeliveryAsync(confirmation, cancellationToken).ConfigureAwait(false)) |
| | | 456 | | { |
| | 2 | 457 | | var dispatchResult = await DispatchToRecoveryAsync(hasLiveSubscriber: null).ConfigureAwait(false); |
| | 2 | 458 | | AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, dispatchResult.Action, dispatchResult.RouteMixed); |
| | 2 | 459 | | AsyncResponseDiagnostics.RecordLostSubscriber("response", dispatchResult.Action, dispatchResult.CallbackInvo |
| | 2 | 460 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked); |
| | | 461 | | } |
| | 570 | 462 | | } |
| | | 463 | | |
| | | 464 | | /// <inheritdoc /> |
| | | 465 | | public async Task SetException(Exception exception, string correlationId, CancellationToken cancellationToken = defa |
| | | 466 | | { |
| | 12 | 467 | | ArgumentNullException.ThrowIfNull(exception); |
| | | 468 | | |
| | 12 | 469 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.set_exception", ActivityKind.Producer |
| | 12 | 470 | | activity?.SetTag("asyncresponse.channel", _activityTag); |
| | 12 | 471 | | activity?.SetTag("asyncresponse.exception_type", exception.GetType().FullName ?? exception.GetType().Name); |
| | 12 | 472 | | AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId); |
| | | 473 | | |
| | 12 | 474 | | if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the exception", exception)) |
| | 2 | 475 | | return; |
| | | 476 | | |
| | | 477 | | try |
| | | 478 | | { |
| | 9 | 479 | | var subscribers = await _store.CountActiveSubscribersAsync(correlationId, cancellationToken).ConfigureAwait( |
| | 6 | 480 | | activity?.SetTag("asyncresponse.subscribers", subscribers); |
| | 6 | 481 | | if (subscribers <= 0) |
| | | 482 | | { |
| | 2 | 483 | | var dispatchResult = await _lostSubscriberDispatcher |
| | 2 | 484 | | .DispatchLostExceptions( |
| | 2 | 485 | | _recoveryStateStore, |
| | 2 | 486 | | correlationId, |
| | 2 | 487 | | exception, |
| | 2 | 488 | | ChannelName(correlationId), |
| | 2 | 489 | | cancellationToken, |
| | 2 | 490 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 2 | 491 | | .ConfigureAwait(false); |
| | 2 | 492 | | if (!dispatchResult.RetryLive) |
| | | 493 | | { |
| | 2 | 494 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked); |
| | 2 | 495 | | AsyncResponseDiagnostics.RecordLostSubscriber("exception", action: RecoveryAction.Fail, dispatchResu |
| | 2 | 496 | | return; |
| | | 497 | | } |
| | | 498 | | |
| | | 499 | | // A waiter registered between the count and the recovery-state read — publish live |
| | | 500 | | // instead of consuming its registration. |
| | | 501 | | } |
| | | 502 | | |
| | 4 | 503 | | var envelope = new AsyncResponseEnvelope<object> |
| | 4 | 504 | | { |
| | 4 | 505 | | Success = false, |
| | 4 | 506 | | ExceptionMessage = exception.Message, |
| | 4 | 507 | | ExceptionStackTrace = RemoteStackTrace.ForWire(exception.StackTrace, _options.IncludeRemoteStackTrace, _ |
| | 4 | 508 | | Payload = null |
| | 4 | 509 | | }; |
| | 4 | 510 | | var json = AsyncResponseEnvelopeJson.Serialize(envelope); |
| | 4 | 511 | | var messageId = Guid.NewGuid(); |
| | 4 | 512 | | using var confirmation = BeginConfirmation(messageId); |
| | 4 | 513 | | await PublishMessageAsync(messageId, correlationId, json, cancellationToken).ConfigureAwait(false); |
| | | 514 | | |
| | 4 | 515 | | if (!await TryConfirmDeliveryAsync(confirmation, cancellationToken).ConfigureAwait(false)) |
| | | 516 | | { |
| | | 517 | | // No live re-check here: TryClaimForRecoveryAsync already won the message for the |
| | | 518 | | // recovery path, so live delivery of it is no longer possible. |
| | 1 | 519 | | var dispatchResult = await _lostSubscriberDispatcher |
| | 1 | 520 | | .DispatchLostExceptions(_recoveryStateStore, correlationId, exception, ChannelName(correlationId), c |
| | 1 | 521 | | .ConfigureAwait(false); |
| | 1 | 522 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked); |
| | 1 | 523 | | AsyncResponseDiagnostics.RecordLostSubscriber("exception", action: RecoveryAction.Fail, dispatchResult.C |
| | | 524 | | } |
| | 4 | 525 | | } |
| | 3 | 526 | | catch (Exception ex) |
| | | 527 | | { |
| | 3 | 528 | | _logger.LogError(ex, "Failed to publish {Provider} exception response for correlationId {CorrelationId}.", _ |
| | 3 | 529 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 3 | 530 | | throw; |
| | | 531 | | } |
| | 8 | 532 | | } |
| | | 533 | | |
| | | 534 | | /// <inheritdoc /> |
| | | 535 | | public async ValueTask<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken = |
| | | 536 | | { |
| | 9 | 537 | | if (string.IsNullOrWhiteSpace(correlationId)) |
| | 2 | 538 | | return 0L; |
| | | 539 | | |
| | | 540 | | try |
| | | 541 | | { |
| | 7 | 542 | | return await _store.CountActiveSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false); |
| | | 543 | | } |
| | 1 | 544 | | catch (Exception ex) when (ex is not OperationCanceledException) |
| | | 545 | | { |
| | 1 | 546 | | _logger.LogDebug(ex, "Failed to count {Provider} subscribers for correlationId {CorrelationId}.", _providerN |
| | | 547 | | // Negative = "could not be probed" (the watchdog's documented unknown-liveness |
| | | 548 | | // contract): returning 0 would assert there is definitively no live waiter, flagging |
| | | 549 | | // every over-threshold registration stale during a transient probe outage. |
| | 1 | 550 | | return -1L; |
| | | 551 | | } |
| | 9 | 552 | | } |
| | | 553 | | |
| | | 554 | | /// <summary> |
| | | 555 | | /// Drops local subscriptions while leaving recovery state intact. Used by the sample app to |
| | | 556 | | /// simulate a redeploy for lost-subscriber integration tests. |
| | | 557 | | /// </summary> |
| | | 558 | | internal async Task DropLocalSubscriptionsAsync(CancellationToken cancellationToken = default) |
| | | 559 | | { |
| | 16 | 560 | | foreach (var (correlationId, group) in _subscriptions.ToArray()) |
| | | 561 | | { |
| | 16 | 562 | | foreach (var subscription in group.Values.ToArray()) |
| | | 563 | | { |
| | 4 | 564 | | await subscription.DropLocalAsync(cancellationToken).ConfigureAwait(false); |
| | | 565 | | // Retire the registry registration too (as RemoveSubscription does): a leftover |
| | | 566 | | // refcount would defeat the tombstone set by the RemoveAsync below, letting a |
| | | 567 | | // later delivery recreate an executor nothing ever retires. |
| | 4 | 568 | | if (group.TryRemove(subscription.Id, out _)) |
| | 4 | 569 | | _executors.OnSubscriptionRetired(ChannelName(correlationId)); |
| | 4 | 570 | | } |
| | | 571 | | |
| | 4 | 572 | | UnlinkIfEmpty(correlationId, group); |
| | | 573 | | |
| | 4 | 574 | | await _executors.RemoveAsync(ChannelName(correlationId)).ConfigureAwait(false); |
| | 4 | 575 | | } |
| | 4 | 576 | | } |
| | | 577 | | |
| | | 578 | | /// <summary> |
| | | 579 | | /// Re-probes waiter liveness for the lost-subscriber dispatcher's snapshot-race re-check, |
| | | 580 | | /// using the same active-subscriber count the publish path consulted. |
| | | 581 | | /// </summary> |
| | | 582 | | private async ValueTask<bool> HasLiveSubscriberAsync(string correlationId, CancellationToken cancellationToken) |
| | 21 | 583 | | => await _store.CountActiveSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false) > 0; |
| | | 584 | | |
| | | 585 | | private protected void AddSubscription(string correlationId, IDbSubscription subscription) |
| | | 586 | | { |
| | | 587 | | // Register with the executor registry BEFORE publishing into the subscription map: every |
| | | 588 | | // dispatch path consults the map and then enqueues, so a delivery racing a visible-but- |
| | | 589 | | // unregistered subscription on a correlation id reused within the tombstone lifetime would |
| | | 590 | | // be silently dropped. In the reversed window (registered, not yet visible) the delivery |
| | | 591 | | // just waits for the next sweep or falls back to lost-subscriber recovery. |
| | 481 | 592 | | _executors.OnSubscriptionRegistered(ChannelName(correlationId)); |
| | 0 | 593 | | while (true) |
| | | 594 | | { |
| | 957 | 595 | | var group = _subscriptions.GetOrAdd(correlationId, _ => new ConcurrentDictionary<Guid, IDbSubscription>()); |
| | 481 | 596 | | group[subscription.Id] = subscription; |
| | | 597 | | |
| | | 598 | | // A concurrent RemoveSubscription may have unlinked this group between the GetOrAdd |
| | | 599 | | // and the insert above (its emptiness check cannot see the in-flight insert). If the |
| | | 600 | | // group this subscription landed in is no longer the mapped one, move it to the live |
| | | 601 | | // group so it stays reachable to every dispatch path. |
| | 481 | 602 | | if (_subscriptions.TryGetValue(correlationId, out var current) && ReferenceEquals(current, group)) |
| | 481 | 603 | | return; |
| | | 604 | | |
| | 0 | 605 | | group.TryRemove(subscription.Id, out _); |
| | | 606 | | } |
| | | 607 | | } |
| | | 608 | | |
| | | 609 | | private void RemoveSubscription(string correlationId, Guid registrationId) |
| | | 610 | | { |
| | 462 | 611 | | if (!_subscriptions.TryGetValue(correlationId, out var group)) |
| | 4 | 612 | | return; |
| | | 613 | | |
| | 458 | 614 | | if (group.TryRemove(registrationId, out _)) |
| | 458 | 615 | | _executors.OnSubscriptionRetired(ChannelName(correlationId)); |
| | 458 | 616 | | UnlinkIfEmpty(correlationId, group); |
| | 458 | 617 | | } |
| | | 618 | | |
| | | 619 | | // Unlinks an emptied subscription group from the map without orphaning a concurrent |
| | | 620 | | // registration: the emptiness read and the map removal cannot be one atomic step, so a |
| | | 621 | | // waiter registered for a reused correlation id in that window would land in an unreachable |
| | | 622 | | // group and time out despite its response being published. Unlink only our exact group, |
| | | 623 | | // then re-link (or merge) anything a racing AddSubscription slipped into it. |
| | | 624 | | private void UnlinkIfEmpty(string correlationId, ConcurrentDictionary<Guid, IDbSubscription> group) |
| | | 625 | | { |
| | 462 | 626 | | if (!group.IsEmpty) |
| | 3 | 627 | | return; |
| | | 628 | | |
| | 459 | 629 | | if (!((ICollection<KeyValuePair<string, ConcurrentDictionary<Guid, IDbSubscription>>>)_subscriptions) |
| | 459 | 630 | | .Remove(new KeyValuePair<string, ConcurrentDictionary<Guid, IDbSubscription>>(correlationId, group))) |
| | 0 | 631 | | return; |
| | | 632 | | |
| | 459 | 633 | | if (group.IsEmpty) |
| | 459 | 634 | | return; |
| | | 635 | | |
| | 0 | 636 | | var merged = _subscriptions.GetOrAdd(correlationId, group); |
| | 0 | 637 | | if (ReferenceEquals(merged, group)) |
| | 0 | 638 | | return; |
| | | 639 | | |
| | 0 | 640 | | foreach (var entry in group) |
| | 0 | 641 | | merged[entry.Key] = entry.Value; |
| | 0 | 642 | | } |
| | | 643 | | |
| | | 644 | | // Executor retirements started off the cleanup path (see CleanupCoreAsync). Keyed by the task |
| | | 645 | | // itself and self-evicting, so a long-lived channel never accumulates completed entries. |
| | 419 | 646 | | private readonly ConcurrentDictionary<Task, byte> _pendingRetirements = new(); |
| | | 647 | | |
| | | 648 | | private void TrackRetirement(Task retirement) |
| | | 649 | | { |
| | 462 | 650 | | _pendingRetirements[retirement] = 0; |
| | 462 | 651 | | _ = retirement.ContinueWith( |
| | 462 | 652 | | static (completed, state) => ((ConcurrentDictionary<Task, byte>)state!).TryRemove(completed, out _), |
| | 462 | 653 | | _pendingRetirements, |
| | 462 | 654 | | CancellationToken.None, |
| | 462 | 655 | | TaskContinuationOptions.ExecuteSynchronously, |
| | 462 | 656 | | TaskScheduler.Default); |
| | 462 | 657 | | } |
| | | 658 | | |
| | | 659 | | /// <summary> |
| | | 660 | | /// The effective minimum interval between full safety-net sweeps, <c>null</c> to sweep on |
| | | 661 | | /// every poll tick. Defaults to the configured <c>FullSweepInterval</c>; a provider overrides |
| | | 662 | | /// it when its push wake is not carrying delivery, because the throttled sweep is then the |
| | | 663 | | /// ONLY cross-process wake and a throttle equal to the delivery-confirmation timeout routed |
| | | 664 | | /// live waiters' responses into lost-subscriber recovery. |
| | | 665 | | /// </summary> |
| | 5909 | 666 | | protected virtual TimeSpan? CurrentFullSweepInterval() => _options.FullSweepInterval; |
| | | 667 | | |
| | | 668 | | private void ThrowIfDisposed() |
| | | 669 | | { |
| | 461 | 670 | | lock (_listenerGate) |
| | | 671 | | { |
| | 461 | 672 | | if (_disposed) |
| | 3 | 673 | | throw new ObjectDisposedException(_channelTypeName); |
| | 458 | 674 | | } |
| | 458 | 675 | | } |
| | | 676 | | |
| | | 677 | | private protected void EnsureListenerStarted() |
| | | 678 | | { |
| | 461 | 679 | | lock (_listenerGate) |
| | | 680 | | { |
| | | 681 | | // Checked under the same gate DisposeAsync sets it under: a racing registration must |
| | | 682 | | // never recreate the CTS and loops after disposal tore them down. |
| | 461 | 683 | | if (_disposed) |
| | 0 | 684 | | throw new ObjectDisposedException(_channelTypeName); |
| | | 685 | | |
| | 461 | 686 | | if (_listenerCts is not null) |
| | 97 | 687 | | return; |
| | | 688 | | |
| | 364 | 689 | | var listenerCts = new CancellationTokenSource(); |
| | 364 | 690 | | _listenerCts = listenerCts; |
| | 364 | 691 | | _listenTask = StartWakeListener(listenerCts.Token); |
| | 728 | 692 | | _dispatchTask = Task.Run(() => DispatchLoopAsync(listenerCts.Token)); |
| | 728 | 693 | | _heartbeatTask = Task.Run(() => HeartbeatLoopAsync(listenerCts.Token)); |
| | 364 | 694 | | } |
| | 461 | 695 | | } |
| | | 696 | | |
| | | 697 | | // The REAL clock, deliberately — here and in the dispatch loop's poll and rescan delays — |
| | | 698 | | // although waiter timeouts and the delivery-confirmation wait arm on _timeProvider. These |
| | | 699 | | // loops keep pace with state that lives in the database and moves in real time whatever clock |
| | | 700 | | // the process was handed: subscriber rows expire on the SERVER's clock, and another process's |
| | | 701 | | // response becomes visible when ITS transaction commits. A heartbeat parked on a virtual clock |
| | | 702 | | // that a test never advances lets the rows of live waiters expire server-side (their responses |
| | | 703 | | // then route to lost-subscriber recovery), and a parked poll never delivers a cross-process |
| | | 704 | | // response at all. What the injected clock owns is time the process itself defines. |
| | | 705 | | private async Task HeartbeatLoopAsync(CancellationToken cancellationToken) |
| | | 706 | | { |
| | 4646 | 707 | | while (!cancellationToken.IsCancellationRequested) |
| | | 708 | | { |
| | | 709 | | try |
| | | 710 | | { |
| | 4645 | 711 | | await Task.Delay(_options.SubscriberHeartbeatInterval, cancellationToken).ConfigureAwait(false); |
| | 4286 | 712 | | var registrations = SnapshotActiveRegistrations(); |
| | 4286 | 713 | | if (registrations.Count > 0) |
| | | 714 | | { |
| | | 715 | | try |
| | | 716 | | { |
| | 1905 | 717 | | await _store.HeartbeatSubscribersAsync( |
| | 1905 | 718 | | _instanceId, |
| | 1905 | 719 | | registrations, |
| | 1905 | 720 | | _options.SubscriberHeartbeatTimeout, |
| | 1905 | 721 | | cancellationToken).ConfigureAwait(false); |
| | 1895 | 722 | | } |
| | 1 | 723 | | catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) |
| | | 724 | | { |
| | 1 | 725 | | return; |
| | | 726 | | } |
| | 9 | 727 | | catch (Exception ex) |
| | | 728 | | { |
| | | 729 | | // The round still compensates below: SQL Server commits per-batch, |
| | | 730 | | // MongoDB bulk-writes unordered, and any provider can fail after some |
| | | 731 | | // upserts landed — a registration dropped mid-round may already be |
| | | 732 | | // resurrected even though the round as a whole threw. Skipping the |
| | | 733 | | // re-check on failure left exactly those rows phantom until TTL. |
| | 9 | 734 | | _logger.LogWarning(ex, "{Provider} subscriber heartbeat failed; retrying for all local waiters." |
| | 9 | 735 | | } |
| | | 736 | | |
| | 1904 | 737 | | await DeleteRegistrationsDroppedDuringHeartbeatAsync(registrations, cancellationToken).ConfigureAwai |
| | | 738 | | } |
| | 4282 | 739 | | } |
| | 362 | 740 | | catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) |
| | | 741 | | { |
| | 362 | 742 | | return; |
| | | 743 | | } |
| | 0 | 744 | | catch (Exception ex) |
| | | 745 | | { |
| | | 746 | | // Distinct from the inner catch's message, which reports a failed store upsert |
| | | 747 | | // that the drop compensation below it still runs. Reaching HERE means the round's |
| | | 748 | | // own bookkeeping broke — the snapshot, or the compensating deletes — so subscriber |
| | | 749 | | // rows dropped mid-round stay resurrected and suppress lost-subscriber recovery for |
| | | 750 | | // their correlation ids until the heartbeat timeout. Different cause, different |
| | | 751 | | // operator response, so it must not read identically. |
| | 0 | 752 | | _logger.LogWarning( |
| | 0 | 753 | | ex, |
| | 0 | 754 | | "{Provider} subscriber heartbeat round failed outside the store upsert (snapshot or drop compensatio |
| | 0 | 755 | | _providerName); |
| | 0 | 756 | | } |
| | | 757 | | } |
| | 364 | 758 | | } |
| | | 759 | | |
| | | 760 | | private List<(string CorrelationId, Guid RegistrationId)> SnapshotActiveRegistrations() |
| | | 761 | | { |
| | | 762 | | // Full (correlation id, registration id) pairs: the heartbeat UPSERTs the subscriber |
| | | 763 | | // records, so it needs everything required to re-create one the store's expiry pruning |
| | | 764 | | // (relational pruner / TTL reaper) has already deleted. |
| | 4286 | 765 | | var registrations = new List<(string CorrelationId, Guid RegistrationId)>(); |
| | 13774 | 766 | | foreach (var (correlationId, group) in _subscriptions) |
| | | 767 | | { |
| | 10406 | 768 | | foreach (var subscription in group.Values) |
| | | 769 | | { |
| | 2602 | 770 | | if (!subscription.Dropped) |
| | 2545 | 771 | | registrations.Add((correlationId, subscription.Id)); |
| | | 772 | | } |
| | | 773 | | } |
| | | 774 | | |
| | 4286 | 775 | | return registrations; |
| | | 776 | | } |
| | | 777 | | |
| | | 778 | | /// <summary> |
| | | 779 | | /// Closes the heartbeat/cleanup race: a subscription can be dropped (and its subscriber row |
| | | 780 | | /// deleted) AFTER the snapshot above was taken but BEFORE the round's upsert landed — the |
| | | 781 | | /// upsert then resurrects the deleted row, and until it ages out past the heartbeat timeout |
| | | 782 | | /// every publisher counts a live waiter that no longer exists, suppressing lost-subscriber |
| | | 783 | | /// recovery for the correlation id. Both cleanup paths set <c>Dropped</c> BEFORE issuing |
| | | 784 | | /// their delete, which makes this post-round re-check airtight: either the drop is visible |
| | | 785 | | /// here and the compensating delete below lands after the resurrecting upsert, or the drop |
| | | 786 | | /// happened after this check — and then the cleanup's own delete is ordered after the upsert |
| | | 787 | | /// and removes the row itself. Best-effort like the cleanup delete: a failed compensation |
| | | 788 | | /// ages out via the heartbeat timeout. |
| | | 789 | | /// </summary> |
| | | 790 | | private async Task DeleteRegistrationsDroppedDuringHeartbeatAsync( |
| | | 791 | | List<(string CorrelationId, Guid RegistrationId)> heartbeaten, |
| | | 792 | | CancellationToken cancellationToken) |
| | | 793 | | { |
| | 8853 | 794 | | foreach (var (correlationId, registrationId) in heartbeaten) |
| | | 795 | | { |
| | 2522 | 796 | | if (IsRegistrationLive(correlationId, registrationId)) |
| | | 797 | | continue; |
| | | 798 | | |
| | 37 | 799 | | _logger.LogDebug( |
| | 37 | 800 | | "Deleting {Provider} subscriber {RegistrationId} for correlationId {CorrelationId}: it was dropped while |
| | 37 | 801 | | _providerName, registrationId, correlationId); |
| | | 802 | | try |
| | | 803 | | { |
| | 37 | 804 | | await _store.DeleteSubscriberAsync(correlationId, registrationId, cancellationToken).ConfigureAwait(fals |
| | 30 | 805 | | } |
| | 3 | 806 | | catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) |
| | | 807 | | { |
| | 3 | 808 | | throw; |
| | | 809 | | } |
| | 4 | 810 | | catch (Exception ex) |
| | | 811 | | { |
| | 4 | 812 | | _logger.LogError(ex, |
| | 4 | 813 | | "Failed to delete {Provider} subscriber {SubscriberRecord} for correlationId {CorrelationId} after i |
| | 4 | 814 | | _providerName, _subscriberRecordNoun, correlationId); |
| | 4 | 815 | | } |
| | 34 | 816 | | } |
| | 1903 | 817 | | } |
| | | 818 | | |
| | | 819 | | private bool IsRegistrationLive(string correlationId, Guid registrationId) |
| | 2522 | 820 | | => _subscriptions.TryGetValue(correlationId, out var group) |
| | 2522 | 821 | | && group.TryGetValue(registrationId, out var subscription) |
| | 2522 | 822 | | && !subscription.Dropped; |
| | | 823 | | |
| | | 824 | | private async Task DispatchLoopAsync(CancellationToken cancellationToken) |
| | | 825 | | { |
| | 7110 | 826 | | while (!cancellationToken.IsCancellationRequested) |
| | | 827 | | { |
| | | 828 | | try |
| | | 829 | | { |
| | 6874 | 830 | | var scope = await CollectDispatchScopeAsync(cancellationToken).ConfigureAwait(false); |
| | 6780 | 831 | | await DispatchPendingMessagesAsync(scope, cancellationToken).ConfigureAwait(false); |
| | 6744 | 832 | | } |
| | 109 | 833 | | catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) |
| | | 834 | | { |
| | 109 | 835 | | return; |
| | | 836 | | } |
| | 21 | 837 | | catch (Exception ex) |
| | | 838 | | { |
| | 21 | 839 | | _logger.LogWarning(ex, "{Provider} response dispatch loop failed; retrying after poll delay.", _provider |
| | 21 | 840 | | await Task.Delay(CurrentPollInterval(), cancellationToken).ConfigureAwait(false); |
| | | 841 | | } |
| | | 842 | | } |
| | 345 | 843 | | } |
| | | 844 | | |
| | | 845 | | /// <summary> |
| | | 846 | | /// Waits for the next dispatch trigger and returns its scope. <c>null</c> means scan every |
| | | 847 | | /// subscribed correlation id — a full sweep requested explicitly (a null signal) or by the |
| | | 848 | | /// periodic poll that is the missed-wake / cross-process-delivery safety net. A non-null set |
| | | 849 | | /// scans only the signaled correlation ids, so a flood of wake signals never forces a scan of |
| | | 850 | | /// every waiter. |
| | | 851 | | /// <para> |
| | | 852 | | /// The poll deadline is ABSOLUTE and judged after either wake source. It used to be a fresh |
| | | 853 | | /// <c>Task.Delay</c> per pass that only counted when it won the race, so a steady stream of |
| | | 854 | | /// targeted signals cancelled every delay and the full sweep never ran: a response published |
| | | 855 | | /// from another process with no local signal — every cross-process response on SQL Server, |
| | | 856 | | /// any missed or dropped notification elsewhere (the signal channel itself drops its oldest |
| | | 857 | | /// entry when full) — sat undelivered for as long as unrelated local traffic continued. |
| | | 858 | | /// </para> |
| | | 859 | | /// </summary> |
| | | 860 | | private protected async Task<HashSet<string>?> CollectDispatchScopeAsync(CancellationToken cancellationToken) |
| | | 861 | | { |
| | | 862 | | // Armed on the first pass rather than at construction: the loop starts lazily, and a |
| | | 863 | | // deadline measured from the constructor would already be overdue by then. |
| | 7512 | 864 | | _pollArmedAt ??= Stopwatch.GetTimestamp(); |
| | | 865 | | |
| | 7512 | 866 | | var signalled = false; |
| | 7512 | 867 | | var untilPoll = CurrentPollInterval() - Stopwatch.GetElapsedTime(_pollArmedAt.Value); |
| | 7512 | 868 | | var pollDue = untilPoll <= TimeSpan.Zero; |
| | 7512 | 869 | | if (!pollDue) |
| | | 870 | | { |
| | | 871 | | // The WhenAny loser is cancelled via the per-iteration linked source: an abandoned |
| | | 872 | | // WaitToReadAsync would otherwise stay parked in the channel's blocked-reader list until |
| | | 873 | | // the next signal — one per poll interval, accumulating without bound on an idle channel. |
| | 7321 | 874 | | using var iteration = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); |
| | 7321 | 875 | | var delay = Task.Delay(untilPoll, iteration.Token); |
| | 7321 | 876 | | var signal = _signals.Reader.WaitToReadAsync(iteration.Token).AsTask(); |
| | 7321 | 877 | | var completed = await Task.WhenAny(delay, signal).ConfigureAwait(false); |
| | 7321 | 878 | | iteration.Cancel(); |
| | 7321 | 879 | | if (completed == signal) |
| | | 880 | | { |
| | 1649 | 881 | | await signal.ConfigureAwait(false); |
| | 1555 | 882 | | signalled = true; |
| | | 883 | | } |
| | | 884 | | else |
| | | 885 | | { |
| | | 886 | | // The timer is the authority for its own tick: it may fire a hair before the |
| | | 887 | | // stopwatch agrees, and that tick must not degrade into an empty pass. |
| | 5672 | 888 | | pollDue = true; |
| | | 889 | | } |
| | 7227 | 890 | | } |
| | | 891 | | |
| | | 892 | | // A signal does not excuse the poll. Re-read the interval while judging it: a signal from |
| | | 893 | | // a new waiter is what re-arms SQL Server's tight active cadence, and that waiter's first |
| | | 894 | | // poll must not wait out the idle interval. |
| | 7418 | 895 | | if (pollDue || Stopwatch.GetElapsedTime(_pollArmedAt.Value) >= CurrentPollInterval()) |
| | | 896 | | { |
| | 5909 | 897 | | _pollArmedAt = Stopwatch.GetTimestamp(); |
| | | 898 | | |
| | | 899 | | // The timer sweep costs one store query per subscribed correlation id, so with W |
| | | 900 | | // waiters an idle channel pays W queries per poll tick. FullSweepInterval bounds that: |
| | | 901 | | // a tick whose sweep is not yet due scans only what was signalled (possibly nothing). |
| | | 902 | | // Provider-resolved: a provider whose push wake is off or unavailable has no other |
| | | 903 | | // cross-process delivery path and must sweep every tick. |
| | 5909 | 904 | | if (CurrentFullSweepInterval() is not { } fullSweepInterval |
| | 5909 | 905 | | || _lastFullSweepAt is not { } lastFullSweepAt |
| | 5909 | 906 | | || Stopwatch.GetElapsedTime(lastFullSweepAt) >= fullSweepInterval) |
| | | 907 | | { |
| | | 908 | | // Queued signals stay queued: the sweep covers their correlation ids, and the |
| | | 909 | | // next pass re-scans them as a cheap targeted scope instead of this pass having |
| | | 910 | | // to reason about signals written while the sweep was running. |
| | 5883 | 911 | | _lastFullSweepAt = Stopwatch.GetTimestamp(); |
| | 5883 | 912 | | return null; |
| | | 913 | | } |
| | | 914 | | } |
| | | 915 | | |
| | 1535 | 916 | | var scope = new HashSet<string>(StringComparer.Ordinal); |
| | 1535 | 917 | | var fullSweep = false; |
| | 6352 | 918 | | for (var read = 0; read < MaxSignalsPerPass && _signals.Reader.TryRead(out var correlationId); read++) |
| | | 919 | | { |
| | 1641 | 920 | | if (string.IsNullOrEmpty(correlationId)) |
| | 2 | 921 | | fullSweep = true; |
| | | 922 | | else |
| | 1639 | 923 | | scope.Add(correlationId); |
| | | 924 | | } |
| | | 925 | | |
| | 1535 | 926 | | if (fullSweep || (signalled && scope.Count == 0)) |
| | | 927 | | { |
| | | 928 | | // A signal-driven full sweep does the timer sweep's work; stamping it defers the next |
| | | 929 | | // timer sweep by a full interval instead of re-scanning everything twice in a row. |
| | 2 | 930 | | _lastFullSweepAt = Stopwatch.GetTimestamp(); |
| | 2 | 931 | | return null; |
| | | 932 | | } |
| | | 933 | | |
| | 1533 | 934 | | return scope.Count == 0 ? EmptyDispatchScope : scope; |
| | 7418 | 935 | | } |
| | | 936 | | |
| | | 937 | | /// <summary>Returned for a poll tick whose full sweep is not yet due: scan nothing. Never mutated.</summary> |
| | 12 | 938 | | private static readonly HashSet<string> EmptyDispatchScope = []; |
| | | 939 | | |
| | | 940 | | /// <summary> |
| | | 941 | | /// Most signals one pass folds into its scope. The channel holds this many, so a pass still |
| | | 942 | | /// takes everything that was queued when it started; the bound only stops it from chasing |
| | | 943 | | /// writers that refill the channel as fast as it drains, which would keep the dispatch — and |
| | | 944 | | /// the poll deadline behind it — waiting on the drain. |
| | | 945 | | /// </summary> |
| | | 946 | | private const int MaxSignalsPerPass = 1024; |
| | | 947 | | |
| | | 948 | | // Stopwatch timestamps, not wall-clock stamps: both are interval deadlines, and a system |
| | | 949 | | // clock stepping backwards must not postpone a sweep. Touched only by the dispatch loop. |
| | | 950 | | private long? _pollArmedAt; |
| | | 951 | | private long? _lastFullSweepAt; |
| | | 952 | | |
| | | 953 | | private protected async Task DispatchPendingMessagesAsync(HashSet<string>? scope, CancellationToken cancellationToke |
| | | 954 | | { |
| | 6791 | 955 | | if (scope is not null) |
| | | 956 | | { |
| | | 957 | | // A publish signals exactly one correlation id, so a targeted scan must cost |
| | | 958 | | // O(scope), not O(live waiters): enumerating the whole registry made every publish |
| | | 959 | | // quadratic under load, and the not-yet-due poll tick (an empty scope) paid the same |
| | | 960 | | // walk to match nothing. |
| | 3918 | 961 | | foreach (var correlationId in scope) |
| | | 962 | | { |
| | 1008 | 963 | | if (_subscriptions.TryGetValue(correlationId, out var group)) |
| | 966 | 964 | | await DispatchPendingCorrelationAsync(correlationId, group, cancellationToken).ConfigureAwait(false) |
| | | 965 | | } |
| | | 966 | | |
| | 936 | 967 | | return; |
| | | 968 | | } |
| | | 969 | | |
| | 22840 | 970 | | foreach (var (correlationId, group) in _subscriptions) |
| | 5600 | 971 | | await DispatchPendingCorrelationAsync(correlationId, group, cancellationToken).ConfigureAwait(false); |
| | 6751 | 972 | | } |
| | | 973 | | |
| | | 974 | | // The group owns its scan progress: removing the last subscription also makes the cursor |
| | | 975 | | // collectible, without another per-correlation registry or a cleanup race on reused ids. |
| | 419 | 976 | | private readonly ConditionalWeakTable<ConcurrentDictionary<Guid, IDbSubscription>, DispatchScan> _dispatchScans = ne |
| | | 977 | | private const int MaxForwardPagesPerPass = 16; |
| | 12 | 978 | | private static readonly Guid LastMessageId = new("ffffffff-ffff-ffff-ffff-ffffffffffff"); |
| | | 979 | | |
| | | 980 | | private sealed class MessageCursor |
| | | 981 | | { |
| | | 982 | | public DateTimeOffset? CreatedAtUtc; |
| | | 983 | | public Guid? Id; |
| | 1905 | 984 | | public void Advance(DbChannelMessage message) { CreatedAtUtc = message.CreatedAtUtc; Id = message.Id; } |
| | | 985 | | } |
| | | 986 | | |
| | | 987 | | private sealed class DispatchScan |
| | | 988 | | { |
| | 456 | 989 | | public HashSet<Guid> Registrations = []; |
| | 456 | 990 | | public MessageCursor Forward = new(); |
| | | 991 | | public bool ForwardCaughtUp; |
| | | 992 | | public MessageCursor? Reconciliation; |
| | | 993 | | public DateTimeOffset? ReconciliationEndUtc; |
| | | 994 | | public Guid? ReconciliationEndId; |
| | | 995 | | public DateTimeOffset ReconcileAfter; |
| | | 996 | | public int RewindRequested; |
| | | 997 | | } |
| | | 998 | | |
| | | 999 | | private async Task DispatchPendingCorrelationAsync( |
| | | 1000 | | string correlationId, |
| | | 1001 | | ConcurrentDictionary<Guid, IDbSubscription> group, |
| | | 1002 | | CancellationToken cancellationToken) |
| | | 1003 | | { |
| | | 1004 | | // The oldest watermark is folded into the pass that builds the list: this runs per |
| | | 1005 | | // correlation id on every sweep tick AND on every publish's targeted scan, so a separate |
| | | 1006 | | // LINQ Min() was one enumerator allocation and one delegate call per element on the |
| | | 1007 | | // dispatch hot path, over a list this loop has in hand anyway. |
| | 6566 | 1008 | | var subscriptions = new List<IDbSubscription>(group.Count); |
| | 6566 | 1009 | | var oldestStartedAtUtc = DateTimeOffset.MaxValue; |
| | 26276 | 1010 | | foreach (var subscription in group.Values) |
| | | 1011 | | { |
| | 6572 | 1012 | | if (subscription.Dropped) |
| | | 1013 | | continue; |
| | | 1014 | | |
| | 6346 | 1015 | | subscriptions.Add(subscription); |
| | 6346 | 1016 | | if (subscription.StartedAtUtc < oldestStartedAtUtc) |
| | 6342 | 1017 | | oldestStartedAtUtc = subscription.StartedAtUtc; |
| | | 1018 | | } |
| | 6566 | 1019 | | if (subscriptions.Count == 0) |
| | 226 | 1020 | | return; |
| | | 1021 | | |
| | 6340 | 1022 | | var since = oldestStartedAtUtc.AddSeconds(-1); |
| | 6340 | 1023 | | var seenCutoff = _timeProvider.GetUtcNow() - _options.MessageRetention - TimeSpan.FromMinutes(1); |
| | 25372 | 1024 | | foreach (var subscription in subscriptions) |
| | 6346 | 1025 | | subscription.PruneSeen(seenCutoff); |
| | | 1026 | | |
| | 6340 | 1027 | | var scan = _dispatchScans.GetOrCreateValue(group); |
| | 12686 | 1028 | | var registrations = subscriptions.Select(subscription => subscription.Id).ToHashSet(); |
| | 6340 | 1029 | | var now = _timeProvider.GetUtcNow(); |
| | 6340 | 1030 | | if (!scan.Registrations.SetEquals(registrations) || Interlocked.Exchange(ref scan.RewindRequested, 0) != 0) |
| | | 1031 | | { |
| | 459 | 1032 | | scan.Registrations = registrations; |
| | 459 | 1033 | | scan.Forward = new MessageCursor(); |
| | 459 | 1034 | | scan.ForwardCaughtUp = false; |
| | 459 | 1035 | | scan.Reconciliation = null; |
| | 459 | 1036 | | scan.ReconcileAfter = now + _options.HistoryReconciliationInterval; |
| | | 1037 | | } |
| | | 1038 | | |
| | | 1039 | | // Normal polls and targeted signals continue after the last admitted page. A new waiter |
| | | 1040 | | // resets progress so its own watermark, not another waiter's seen set, decides fan-out. |
| | 6340 | 1041 | | var previousForward = scan.Forward; |
| | 6340 | 1042 | | if (scan.ForwardCaughtUp && scan.Forward.CreatedAtUtc is { } lastTick && lastTick > DateTimeOffset.MinValue) |
| | | 1043 | | { |
| | | 1044 | | // A database clock tick can contain several random ids. A newly committed message |
| | | 1045 | | // in the LAST tick must not wait for historical reconciliation merely because its |
| | | 1046 | | // id sorts before the previous message. Revisit that tick, not the entire history. |
| | | 1047 | | // The provider may truncate the sub-tick timestamp to milliseconds/microseconds; |
| | | 1048 | | // the maximum id excludes rows at that preceding, truncated timestamp. |
| | 261 | 1049 | | scan.Forward = new MessageCursor { CreatedAtUtc = lastTick.AddTicks(-1), Id = LastMessageId }; |
| | | 1050 | | } |
| | 6340 | 1051 | | scan.ForwardCaughtUp = false; |
| | 6340 | 1052 | | var forwardReadAny = false; |
| | 12680 | 1053 | | for (var page = 0; page < MaxForwardPagesPerPass; page++) |
| | | 1054 | | { |
| | 6340 | 1055 | | var (more, admitted, _) = await DispatchPageAsync(scan.Forward).ConfigureAwait(false); |
| | 6300 | 1056 | | if (!admitted) |
| | 1 | 1057 | | return; |
| | 6299 | 1058 | | if (!more) |
| | | 1059 | | { |
| | 6299 | 1060 | | scan.ForwardCaughtUp = true; |
| | 6299 | 1061 | | break; |
| | | 1062 | | } |
| | 0 | 1063 | | if (page == MaxForwardPagesPerPass - 1) |
| | 0 | 1064 | | ScheduleBackpressureRescan(correlationId, cancellationToken); |
| | | 1065 | | } |
| | 6299 | 1066 | | if (!forwardReadAny) |
| | 5665 | 1067 | | scan.Forward = previousForward; // Expired/pruned tail: do not walk backward on idle polls. |
| | | 1068 | | |
| | | 1069 | | // Creation keys are NOT commit order: a transaction can become visible behind the |
| | | 1070 | | // cursor, even with the same timestamp and a lower id, and another process may already |
| | | 1071 | | // have acknowledged it. Reconcile retained history periodically, one page per pass. |
| | | 1072 | | // Both unacked and acked rows participate; filtering acked rows would break fan-out. |
| | 6299 | 1073 | | if (scan.Reconciliation is null && now >= scan.ReconcileAfter && scan.Forward.Id is not null) |
| | | 1074 | | { |
| | 1 | 1075 | | scan.Reconciliation = new MessageCursor(); |
| | 1 | 1076 | | scan.ReconciliationEndUtc = scan.Forward.CreatedAtUtc; |
| | 1 | 1077 | | scan.ReconciliationEndId = scan.Forward.Id; |
| | | 1078 | | } |
| | 6299 | 1079 | | if (scan.Reconciliation is { } reconciliation) |
| | | 1080 | | { |
| | 1 | 1081 | | var (more, admitted, reachedEnd) = await DispatchPageAsync(reconciliation, reconcile: true).ConfigureAwait(f |
| | 1 | 1082 | | if (admitted && (!more || reachedEnd)) |
| | | 1083 | | { |
| | 1 | 1084 | | scan.Reconciliation = null; |
| | 1 | 1085 | | scan.ReconcileAfter = _timeProvider.GetUtcNow() + _options.HistoryReconciliationInterval; |
| | | 1086 | | } |
| | | 1087 | | else |
| | 0 | 1088 | | ScheduleBackpressureRescan(correlationId, cancellationToken); |
| | | 1089 | | } |
| | | 1090 | | |
| | | 1091 | | async Task<(bool More, bool Admitted, bool ReachedEnd)> DispatchPageAsync(MessageCursor cursor, bool reconcile = |
| | | 1092 | | { |
| | 6341 | 1093 | | var messages = await _store.LoadMessagesAsync( |
| | 6341 | 1094 | | correlationId, since, _options.PendingMessageBatchSize, |
| | 6341 | 1095 | | cursor.CreatedAtUtc, cursor.Id, cancellationToken).ConfigureAwait(false); |
| | | 1096 | | |
| | | 1097 | | // Acknowledged rows stay eligible for fan-out, but travel header-only. Hydrate |
| | | 1098 | | // only those a live subscription still needs, then enqueue in page order. |
| | 6302 | 1099 | | List<DbChannelMessage>? eligible = null; |
| | 6302 | 1100 | | List<Guid>? headerOnly = null; |
| | 14114 | 1101 | | foreach (var message in messages) |
| | | 1102 | | { |
| | | 1103 | | // The store was asked for ONE exact correlation id, but "exact" is the |
| | | 1104 | | // database's opinion: a case-insensitive (or accent-insensitive) column |
| | | 1105 | | // collation — the SQL Server default in most deployments — answers a query for |
| | | 1106 | | // "FOO" with the rows of "foo". Delivering those would hand one waiter another |
| | | 1107 | | // waiter's response, so the id is re-checked ordinally here, where the |
| | | 1108 | | // library's own comparison rules apply. This also covers pre-existing tables |
| | | 1109 | | // created before the collation was pinned in the DDL. |
| | 755 | 1110 | | if (!string.Equals(message.CorrelationId, correlationId, StringComparison.Ordinal)) |
| | | 1111 | | { |
| | 0 | 1112 | | _logger.LogError( |
| | 0 | 1113 | | "The {Provider} channel store returned a message for correlationId '{ReturnedCorrelationId}' whe |
| | 0 | 1114 | | "The correlation-id column is not using a case-sensitive/binary collation, so distinct correlati |
| | 0 | 1115 | | "The message was NOT delivered to the wrong waiter. Re-create the AsyncResponse tables (or ALTER |
| | 0 | 1116 | | _providerName, message.CorrelationId, correlationId); |
| | 0 | 1117 | | continue; |
| | | 1118 | | } |
| | | 1119 | | |
| | | 1120 | | // Reconciliation and last-tick overlap revisit seen headers; keep those out of |
| | | 1121 | | // the executor queue. The work item re-checks after admission as well. |
| | 755 | 1122 | | if (!WouldDeliverToAnySubscription(message, subscriptions)) |
| | | 1123 | | continue; |
| | | 1124 | | |
| | 86 | 1125 | | (eligible ??= []).Add(message); |
| | 86 | 1126 | | if (message.EnvelopeJson is null) |
| | 17 | 1127 | | (headerOnly ??= []).Add(message.Id); |
| | | 1128 | | } |
| | | 1129 | | |
| | 6302 | 1130 | | if (eligible is not null && !await EnqueueEligibleAsync(correlationId, eligible, headerOnly, subscriptions, |
| | 1 | 1131 | | return (false, false, false); // Retry this page: never advance past refused work. |
| | | 1132 | | |
| | 6300 | 1133 | | if (messages.Count > 0) |
| | | 1134 | | { |
| | 635 | 1135 | | cursor.Advance(messages[^1]); |
| | 635 | 1136 | | if (!reconcile) |
| | 634 | 1137 | | forwardReadAny = true; |
| | | 1138 | | } |
| | 6300 | 1139 | | var reachedEnd = reconcile && messages.Any(message => |
| | 6301 | 1140 | | message.Id == scan.ReconciliationEndId || message.CreatedAtUtc > scan.ReconciliationEndUtc); |
| | 6300 | 1141 | | return (messages.Count == _options.PendingMessageBatchSize, true, reachedEnd); |
| | 6301 | 1142 | | } |
| | 6526 | 1143 | | } |
| | | 1144 | | |
| | | 1145 | | /// <summary> |
| | | 1146 | | /// Second pass of one sweep page: hydrates the header-only rows among <paramref name="eligible"/> |
| | | 1147 | | /// and admits every row to the correlation id's executor in page order. Returns <c>false</c> |
| | | 1148 | | /// when the executor is full (the page's remaining rows are left in the store, in order, and |
| | | 1149 | | /// a rescan is scheduled), which ends the correlation id's scan for this sweep. |
| | | 1150 | | /// </summary> |
| | | 1151 | | private async Task<bool> EnqueueEligibleAsync( |
| | | 1152 | | string correlationId, |
| | | 1153 | | List<DbChannelMessage> eligible, |
| | | 1154 | | List<Guid>? headerOnly, |
| | | 1155 | | List<IDbSubscription> subscriptions, |
| | | 1156 | | CancellationToken cancellationToken) |
| | | 1157 | | { |
| | 86 | 1158 | | Dictionary<Guid, DbChannelMessage>? hydrated = null; |
| | 86 | 1159 | | if (headerOnly is not null) |
| | | 1160 | | { |
| | 17 | 1161 | | var loaded = await _store.LoadMessagesByIdAsync(correlationId, headerOnly, cancellationToken).ConfigureAwait |
| | 16 | 1162 | | hydrated = new Dictionary<Guid, DbChannelMessage>(loaded.Count); |
| | 64 | 1163 | | foreach (var message in loaded) |
| | | 1164 | | { |
| | | 1165 | | // The by-id read is exact on the id (unique), but a hydrated row must carry its |
| | | 1166 | | // envelope: a store that answered header-only here would hand the waiter nothing. |
| | 16 | 1167 | | if (message.EnvelopeJson is not null) |
| | 16 | 1168 | | hydrated[message.Id] = message; |
| | | 1169 | | } |
| | | 1170 | | } |
| | | 1171 | | |
| | 339 | 1172 | | foreach (var message in eligible) |
| | | 1173 | | { |
| | 85 | 1174 | | var deliverable = message; |
| | 85 | 1175 | | if (message.EnvelopeJson is null) |
| | | 1176 | | { |
| | | 1177 | | // Pruned or expired between the page read and the hydration: nothing to deliver |
| | | 1178 | | // now; a row that is still there is re-evaluated by the next sweep. |
| | 16 | 1179 | | if (hydrated is null || !hydrated.TryGetValue(message.Id, out deliverable)) |
| | | 1180 | | continue; |
| | | 1181 | | } |
| | | 1182 | | |
| | | 1183 | | // Work-item class, not a lambda: a queued closure would chain display classes |
| | | 1184 | | // pinning this paging frame (batch list, cursors, watermark) for as long as the |
| | | 1185 | | // item sits in the executor's bounded queue. |
| | | 1186 | | // |
| | | 1187 | | // NON-BLOCKING admission. This loop is the process-wide dispatch sweep and walks |
| | | 1188 | | // correlation ids sequentially, so waiting for ONE correlation id's executor |
| | | 1189 | | // capacity here (the old EnqueueAsync) parked delivery for every other waiter in |
| | | 1190 | | // the process: a waiter wedged in a slow Until predicate, fed a backlog of NEW |
| | | 1191 | | // progress messages (the pre-filter above only screens consumed history), filled |
| | | 1192 | | // its 1024-slot executor and the sweep then blocked on slot 1025 without ever |
| | | 1193 | | // querying the next correlation id. Its per-correlation backpressure became shared |
| | | 1194 | | // delivery blockage — unrelated remote/polled responses timed out behind it. At |
| | | 1195 | | // capacity the rest of this correlation id's messages are left unclaimed in the |
| | | 1196 | | // store, in order (nothing later is enqueued ahead of them), and a rescan of just |
| | | 1197 | | // this id is scheduled for when the executor has had a poll interval to drain. |
| | 85 | 1198 | | var outcome = _executors.TryEnqueue( |
| | 85 | 1199 | | ChannelName(correlationId), |
| | 85 | 1200 | | new LocalDispatchWorkItem(this, deliverable, subscriptions, cancellationToken).InvokeAsync); |
| | 85 | 1201 | | if (outcome == SerialExecutorRegistry.TryEnqueueOutcome.Full) |
| | | 1202 | | { |
| | 1 | 1203 | | ScheduleBackpressureRescan(correlationId, cancellationToken); |
| | 1 | 1204 | | return false; |
| | | 1205 | | } |
| | | 1206 | | } |
| | | 1207 | | |
| | 84 | 1208 | | return true; |
| | 85 | 1209 | | } |
| | | 1210 | | |
| | | 1211 | | /// <summary> |
| | | 1212 | | /// Would any live subscription actually take this message? Used both as the sweep's |
| | | 1213 | | /// pre-enqueue filter and as the dispatch work item's own guard, so the two can never drift. |
| | | 1214 | | /// </summary> |
| | | 1215 | | private static bool WouldDeliverToAnySubscription(DbChannelMessage message, IReadOnlyList<IDbSubscription> subscript |
| | | 1216 | | { |
| | 4972 | 1217 | | foreach (var subscription in subscriptions) |
| | | 1218 | | { |
| | 1407 | 1219 | | if (!subscription.Dropped && IsWithinWatermark(subscription, message) && !subscription.HasSeen(message.Id)) |
| | 648 | 1220 | | return true; |
| | | 1221 | | } |
| | | 1222 | | |
| | 755 | 1223 | | return false; |
| | 648 | 1224 | | } |
| | | 1225 | | |
| | | 1226 | | private async Task PublishMessageAsync( |
| | | 1227 | | Guid messageId, |
| | | 1228 | | string correlationId, |
| | | 1229 | | string envelopeJson, |
| | | 1230 | | CancellationToken cancellationToken) |
| | | 1231 | | { |
| | | 1232 | | // The insert itself carries the remote wake where the provider has one (a NOTIFY rides the |
| | | 1233 | | // PostgreSQL insert; MongoDB change streams observe it) and the SQL Server sweep polls it |
| | | 1234 | | // up. Only the local fast path and a targeted local signal are needed on top. The store |
| | | 1235 | | // returns the fast-path message with the SERVER-stamped created_at: subscription |
| | | 1236 | | // watermarks are server-clock, and an app-clock timestamp here silently disabled the fast |
| | | 1237 | | // path whenever the app clock ran more than the 1s tolerance behind the database — delivery |
| | | 1238 | | // then quietly degraded to sweep latency on every publish. On an idempotent duplicate (a |
| | | 1239 | | // publish retry) it is the ORIGINAL row, settlement columns included: fabricating |
| | | 1240 | | // AckedAtUtc = null here bypassed IsWithinWatermark's acked-history exclusion, and a retry |
| | | 1241 | | // landing after another process had claimed and acked the first attempt replayed that |
| | | 1242 | | // consumed response to a waiter registered since — the sweep path never had the problem |
| | | 1243 | | // because LoadMessagesAsync reads acked_at. |
| | 556 | 1244 | | var message = await _store.InsertMessageAsync(messageId, correlationId, envelopeJson, _options.MessageRetention, |
| | 556 | 1245 | | .ConfigureAwait(false); |
| | 556 | 1246 | | await TryDispatchLocalSubscribersAsync(message, cancellationToken).ConfigureAwait(false); |
| | 556 | 1247 | | SignalDispatcher(correlationId); |
| | 556 | 1248 | | } |
| | | 1249 | | |
| | | 1250 | | private protected async Task DispatchMessageToSubscribersAsync( |
| | | 1251 | | DbChannelMessage message, |
| | | 1252 | | IReadOnlyList<IDbSubscription> subscriptions, |
| | | 1253 | | CancellationToken cancellationToken) |
| | | 1254 | | { |
| | | 1255 | | // Only subscriptions that are still live, inside their delivery watermark, and have not |
| | | 1256 | | // already processed this message. Skipping when there is nothing to deliver also avoids a |
| | | 1257 | | // redundant claim on every re-sweep. |
| | 648 | 1258 | | if (!WouldDeliverToAnySubscription(message, subscriptions)) |
| | 86 | 1259 | | return; |
| | | 1260 | | |
| | | 1261 | | // Take the message for live delivery. The claim sets acked_at unless the publisher already |
| | | 1262 | | // routed it to recovery (recovery_claimed); losing the claim means recovery owns it, so it is |
| | | 1263 | | // not delivered to the waiter and handled a second time. |
| | | 1264 | | // |
| | | 1265 | | // Claim-then-dispatch is deliberate — keep this ordering. The in-process handoff is |
| | | 1266 | | // at-most-once by design: a crash between the claim and the waiter's continuation can only |
| | | 1267 | | // lose delivery to waiters in THIS dying process, which no ordering could save (their |
| | | 1268 | | // continuations die with it), while pre-registered fan-out waiters in other processes |
| | | 1269 | | // still receive the acked message (IsWithinWatermark admits acked_at > started_at). |
| | | 1270 | | // Dispatch-then-ack behind an expiring claim would re-open the stale-redelivery wrong-data |
| | | 1271 | | // bug the strict acked exclusion in IsWithinWatermark closes. Durability across process |
| | | 1272 | | // death belongs to the layer above: flow re-execution, publish-time recovery routing, and |
| | | 1273 | | // the step timeout. |
| | 562 | 1274 | | if (!await _store.TryClaimForDeliveryAsync(message.Id, cancellationToken).ConfigureAwait(false)) |
| | | 1275 | | { |
| | 4 | 1276 | | foreach (var subscription in subscriptions) |
| | | 1277 | | { |
| | 1 | 1278 | | if (!subscription.Dropped) |
| | 1 | 1279 | | subscription.MarkSeen(message.Id); |
| | | 1280 | | } |
| | 1 | 1281 | | return; |
| | | 1282 | | } |
| | | 1283 | | |
| | | 1284 | | // Wake the publisher immediately if it is waiting in this process — no acked_at polling needed. |
| | 555 | 1285 | | if (_pendingConfirmations.TryGetValue(message.Id, out var confirmation)) |
| | 549 | 1286 | | confirmation.TrySetResult(true); |
| | | 1287 | | |
| | | 1288 | | // Per-subscription isolation is load-bearing, not defensive tidiness. The claim above |
| | | 1289 | | // already stamped acked_at and tripped the publisher's confirmation, so this message is |
| | | 1290 | | // consumed: IsWithinWatermark excludes it from every later sweep and the lost-subscriber |
| | | 1291 | | // path will never see it. If one subscription's dispatch throws OUTSIDE ProcessAsync's own |
| | | 1292 | | // catch — a fault in the captured-context wrapper, or CleanupOnceAsync throwing from |
| | | 1293 | | // CleanupCoreAsync's uncaught finally, which replaces the swallowed exception — letting it |
| | | 1294 | | // propagate would strand every subscription after it in this fan-out until its step |
| | | 1295 | | // timeout, with the response gone. Record the first fault and rethrow only after every |
| | | 1296 | | // sibling has had the message, so the dispatch is still reported as failed. |
| | 555 | 1297 | | List<Exception>? failures = null; |
| | 2230 | 1298 | | foreach (var subscription in subscriptions) |
| | | 1299 | | { |
| | 560 | 1300 | | if (subscription.Dropped || !IsWithinWatermark(subscription, message) || !subscription.MarkSeen(message.Id)) |
| | | 1301 | | continue; |
| | | 1302 | | |
| | | 1303 | | try |
| | | 1304 | | { |
| | 558 | 1305 | | await subscription.ProcessUnderContextAsync(message).ConfigureAwait(false); |
| | 558 | 1306 | | } |
| | 0 | 1307 | | catch (Exception ex) when (ex is not OperationCanceledException || !cancellationToken.IsCancellationRequeste |
| | | 1308 | | { |
| | 0 | 1309 | | _logger.LogError( |
| | 0 | 1310 | | ex, |
| | 0 | 1311 | | "Delivering the {Provider} response for correlationId {CorrelationId} to one waiter failed; the rema |
| | 0 | 1312 | | _providerName, |
| | 0 | 1313 | | message.CorrelationId); |
| | 0 | 1314 | | (failures ??= []).Add(ex); |
| | 0 | 1315 | | } |
| | | 1316 | | } |
| | | 1317 | | |
| | | 1318 | | // Always aggregated, never a bare rethrow of failures[0]: a rethrow would reset that |
| | | 1319 | | // exception's stack trace, and AggregateException carries every inner one intact. |
| | 555 | 1320 | | if (failures is not null) |
| | 0 | 1321 | | throw new AggregateException($"Delivering the response for correlationId {message.CorrelationId} failed for |
| | 642 | 1322 | | } |
| | | 1323 | | |
| | | 1324 | | /// <summary> |
| | | 1325 | | /// Per-subscription delivery watermark. The sweep queries with the OLDEST waiter's watermark on |
| | | 1326 | | /// a shared correlation id, so without this filter a late-joining waiter would receive retained |
| | | 1327 | | /// messages created before it registered. Same 1s tolerance as the query watermark. |
| | | 1328 | | /// <para> |
| | | 1329 | | /// The creation-time tolerance alone re-admits history: a message created inside the 1s skew |
| | | 1330 | | /// window may have already been delivered and acked for a PREVIOUS waiter that reused the |
| | | 1331 | | /// correlation id, and per-subscription seen-tracking cannot dedupe what a different |
| | | 1332 | | /// subscription processed. A message acked before this subscription existed is history, not |
| | | 1333 | | /// delivery — waiters that legitimately participate in a delivery (including cross-process |
| | | 1334 | | /// fan-out) were registered before its claim stamped <c>acked_at</c>. The acked comparison is |
| | | 1335 | | /// deliberately strict, with no skew tolerance: under skew, strictness can only make a waiter |
| | | 1336 | | /// whose registration raced another process's in-flight ack keep waiting for its own response, |
| | | 1337 | | /// whereas a tolerance would re-open the stale-redelivery window this check closes. |
| | | 1338 | | /// </para> |
| | | 1339 | | /// <para> |
| | | 1340 | | /// The comparison must be STRICTLY greater, and that is load-bearing rather than stylistic: a |
| | | 1341 | | /// server clock's resolution is far coarser than its column precision, so equal timestamps are |
| | | 1342 | | /// routine, not a measure-zero tie. SQL Server stamps <c>datetime2(7)</c> from |
| | | 1343 | | /// <c>SYSUTCDATETIME()</c> — 100ns precision, but the clock behind it advances in ~5ms ticks |
| | | 1344 | | /// (measured: 30,344 samples over 300ms yielded 61 distinct values, mean gap 4.9ms), and |
| | | 1345 | | /// MongoDB's <c>$$NOW</c> is millisecond-resolution. A waiter that reuses a correlation id |
| | | 1346 | | /// within one tick of the previous waiter's ack therefore registers at exactly |
| | | 1347 | | /// <c>acked_at</c>, and a non-strict comparison hands it the response its predecessor already |
| | | 1348 | | /// consumed. Registration is ordered strictly after that ack in real time and the clock is |
| | | 1349 | | /// non-decreasing, so <c>acked_at <= started_at</c> always holds for history and the strict |
| | | 1350 | | /// form excludes it deterministically — not probabilistically. |
| | | 1351 | | /// </para> |
| | | 1352 | | /// <para> |
| | | 1353 | | /// The same-tick equality is symmetric — it can also be a genuine cross-process fan-out |
| | | 1354 | | /// delivery (this waiter registered and another process's claim stamped <c>acked_at</c> |
| | | 1355 | | /// inside one clock tick) — and no timestamp can separate the two cases. The store's |
| | | 1356 | | /// monotonic ack sequence arbitrates that tie (and only that tie): every delivery claim |
| | | 1357 | | /// stamps <c>acked_seq</c> drawn from the same monotonic source the subscription drew |
| | | 1358 | | /// <c>StartedSeq</c> from at registration. The arbitration is conservative-exact — exact |
| | | 1359 | | /// whenever the claim's draw was not stalled across ticks; the stalled-draw residual below |
| | | 1360 | | /// resolves as history. |
| | | 1361 | | /// </para> |
| | | 1362 | | /// <para> |
| | | 1363 | | /// The sequence deliberately does NOT outrank truthful (unequal) timestamps. A claim's |
| | | 1364 | | /// sequence value is drawn BEFORE the claim becomes visible — MongoDB draws from a separate |
| | | 1365 | | /// counter document, SQL Server in a <c>DECLARE</c> ahead of an <c>UPDATE</c> that may block |
| | | 1366 | | /// on a row lock, and even PostgreSQL's <c>nextval</c> evaluates before the commit — so a |
| | | 1367 | | /// claim can draw <c>41</c>, stall, and land AFTER a waiter registered at <c>42</c>. Ranking |
| | | 1368 | | /// the sequence above timestamps would exclude that delivery as history even though |
| | | 1369 | | /// <c>acked_at</c> truthfully post-dates the registration tick. With timestamps primary, the |
| | | 1370 | | /// stalled claim lands in a LATER tick and is delivered by the timestamp rule; the sequence |
| | | 1371 | | /// is consulted only when the tick is identical. |
| | | 1372 | | /// </para> |
| | | 1373 | | /// <para> |
| | | 1374 | | /// Inside the tie the sequence can never replay history: a claim visible before a same-tick |
| | | 1375 | | /// registration drew its value before that visibility, hence before the registration's own |
| | | 1376 | | /// draw — <c>acked_seq < StartedSeq</c> — and is excluded. The only residual conservatism |
| | | 1377 | | /// is a claim whose draw-to-execution stall ends exactly in the registration's tick: it |
| | | 1378 | | /// resolves as history, which is the same verdict the timestamp-only rule gave every tie — |
| | | 1379 | | /// never worse, and exact whenever draws are not stalled (the overwhelmingly common case). |
| | | 1380 | | /// Rows acked by a pre-sequence build carry no <c>acked_seq</c> and keep the old at-most-once |
| | | 1381 | | /// tie resolution (excluded fan-out recovers through its step timeout and the |
| | | 1382 | | /// idempotent-restart contract). |
| | | 1383 | | /// </para> |
| | | 1384 | | /// </summary> |
| | | 1385 | | private static bool IsWithinWatermark(IDbSubscription subscription, DbChannelMessage message) |
| | | 1386 | | { |
| | 1510 | 1387 | | if (message.CreatedAtUtc < subscription.StartedAtUtc.AddSeconds(-1)) |
| | 2 | 1388 | | return false; |
| | | 1389 | | |
| | 1508 | 1390 | | if (message.AckedAtUtc is null) |
| | 1230 | 1391 | | return true; |
| | | 1392 | | |
| | | 1393 | | // Timestamps are primary: strictly later tick = delivered, strictly earlier = history. |
| | 278 | 1394 | | if (message.AckedAtUtc > subscription.StartedAtUtc) |
| | 262 | 1395 | | return true; |
| | 16 | 1396 | | if (message.AckedAtUtc < subscription.StartedAtUtc) |
| | 9 | 1397 | | return false; |
| | | 1398 | | |
| | | 1399 | | // Same-tick tie: the monotonic ack sequence arbitrates when the claim carries one. |
| | 7 | 1400 | | if (message.AckedSeq is { } ackedSeq) |
| | 5 | 1401 | | return ackedSeq > subscription.StartedSeq; |
| | | 1402 | | |
| | | 1403 | | // Legacy tie (row acked by a pre-sequence build): the old conservative resolution. |
| | 2 | 1404 | | return false; |
| | | 1405 | | } |
| | | 1406 | | |
| | | 1407 | | private async Task TryDispatchLocalSubscribersAsync(DbChannelMessage message, CancellationToken cancellationToken) |
| | | 1408 | | { |
| | 560 | 1409 | | if (!_subscriptions.TryGetValue(message.CorrelationId, out var group)) |
| | 6 | 1410 | | return; |
| | | 1411 | | |
| | 554 | 1412 | | var subscriptions = new List<IDbSubscription>(group.Count); |
| | 2222 | 1413 | | foreach (var subscription in group.Values) |
| | | 1414 | | { |
| | 557 | 1415 | | if (!subscription.Dropped) |
| | 555 | 1416 | | subscriptions.Add(subscription); |
| | | 1417 | | } |
| | 554 | 1418 | | if (subscriptions.Count == 0) |
| | 2 | 1419 | | return; |
| | | 1420 | | |
| | | 1421 | | // Same-process fast path: skips the wake round trip / sweep latency but still runs on the |
| | | 1422 | | // per-correlation serial executor — completion predicates are guaranteed serial, in-order |
| | | 1423 | | // invocation on every channel, and a direct dispatch here could otherwise run concurrently |
| | | 1424 | | // with a sweep-enqueued dispatch of a different message for the same subscription. MarkSeen |
| | | 1425 | | // keeps the sweep from double-processing this message. |
| | 552 | 1426 | | await _executors.EnqueueAsync( |
| | 552 | 1427 | | ChannelName(message.CorrelationId), |
| | 552 | 1428 | | new LocalDispatchWorkItem(this, message, subscriptions, cancellationToken).InvokeAsync, |
| | 552 | 1429 | | cancellationToken).ConfigureAwait(false); |
| | 560 | 1430 | | } |
| | | 1431 | | |
| | | 1432 | | /// <summary> |
| | | 1433 | | /// Registers an in-process delivery completion for a message id. Disposing it removes the entry, |
| | | 1434 | | /// so a publish that throws or completes never leaks the registration. |
| | | 1435 | | /// </summary> |
| | | 1436 | | private protected PendingConfirmation BeginConfirmation(Guid messageId) |
| | | 1437 | | { |
| | 560 | 1438 | | var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously); |
| | 560 | 1439 | | _pendingConfirmations[messageId] = tcs; |
| | 560 | 1440 | | return new PendingConfirmation(this, messageId, tcs); |
| | | 1441 | | } |
| | | 1442 | | |
| | | 1443 | | /// <summary> |
| | | 1444 | | /// Confirms a published response reached a live waiter. Returns <c>true</c> once a waiter has |
| | | 1445 | | /// acknowledged it; on confirmation timeout, atomically claims the message for the lost-subscriber |
| | | 1446 | | /// path and returns <c>false</c> only if that claim wins — so the recovery callback and a |
| | | 1447 | | /// slow-but-live waiter are mutually exclusive. |
| | | 1448 | | /// </summary> |
| | | 1449 | | private protected async Task<bool> TryConfirmDeliveryAsync(PendingConfirmation confirmation, CancellationToken cance |
| | | 1450 | | { |
| | 558 | 1451 | | if (await WaitForAcknowledgementAsync(confirmation, cancellationToken).ConfigureAwait(false)) |
| | 552 | 1452 | | return true; |
| | | 1453 | | |
| | 4 | 1454 | | return !await _store.TryClaimForRecoveryAsync(confirmation.MessageId, cancellationToken).ConfigureAwait(false); |
| | 556 | 1455 | | } |
| | | 1456 | | |
| | 1646 | 1457 | | private protected void SignalDispatcher(string? correlationId = null) => _signals.Writer.TryWrite(correlationId); |
| | | 1458 | | |
| | | 1459 | | /// <summary> |
| | | 1460 | | /// Correlation ids whose executor was at capacity during a sweep and that have a rescan |
| | | 1461 | | /// pending. One pending rescan per id: a saturated id is re-signalled once per poll interval, |
| | | 1462 | | /// not once per sweep that found it full. |
| | | 1463 | | /// </summary> |
| | 419 | 1464 | | private readonly ConcurrentDictionary<string, byte> _backpressureRescans = new(StringComparer.Ordinal); |
| | | 1465 | | |
| | | 1466 | | /// <summary> |
| | | 1467 | | /// Re-signals a targeted scan of <paramref name="correlationId"/> after one poll interval — |
| | | 1468 | | /// the time the sweep would otherwise have waited for the saturated executor, spent letting |
| | | 1469 | | /// every other correlation id deliver instead. The messages themselves stay in the store |
| | | 1470 | | /// (unclaimed, unseen) until that scan enqueues them, in their original order. |
| | | 1471 | | /// </summary> |
| | | 1472 | | private void ScheduleBackpressureRescan(string correlationId, CancellationToken cancellationToken) |
| | | 1473 | | { |
| | 1 | 1474 | | if (!_backpressureRescans.TryAdd(correlationId, 0)) |
| | 0 | 1475 | | return; |
| | | 1476 | | |
| | 1 | 1477 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | | 1478 | | { |
| | 0 | 1479 | | _logger.LogDebug( |
| | 0 | 1480 | | "{Provider} dispatch for correlationId {CorrelationId} is at executor capacity; the remaining messages a |
| | 0 | 1481 | | _providerName, correlationId); |
| | | 1482 | | } |
| | | 1483 | | |
| | 1 | 1484 | | _ = RescanAfterDelayAsync(correlationId, cancellationToken); |
| | 1 | 1485 | | } |
| | | 1486 | | |
| | | 1487 | | private async Task RescanAfterDelayAsync(string correlationId, CancellationToken cancellationToken) |
| | | 1488 | | { |
| | | 1489 | | try |
| | | 1490 | | { |
| | 1 | 1491 | | await Task.Delay(CurrentPollInterval(), cancellationToken).ConfigureAwait(false); |
| | 0 | 1492 | | } |
| | 1 | 1493 | | catch (OperationCanceledException) |
| | | 1494 | | { |
| | | 1495 | | // Listener stopping: nothing to rescan for. |
| | 1 | 1496 | | } |
| | | 1497 | | finally |
| | | 1498 | | { |
| | 1 | 1499 | | _backpressureRescans.TryRemove(correlationId, out _); |
| | | 1500 | | } |
| | | 1501 | | |
| | 1 | 1502 | | if (!cancellationToken.IsCancellationRequested) |
| | 0 | 1503 | | SignalDispatcher(correlationId); |
| | 1 | 1504 | | } |
| | | 1505 | | |
| | | 1506 | | private async Task<bool> WaitForAcknowledgementAsync(PendingConfirmation confirmation, CancellationToken cancellatio |
| | | 1507 | | { |
| | | 1508 | | // MONOTONIC, on the injected clock: the confirmation budget is a pure interval, and it used |
| | | 1509 | | // to be a wall-clock deadline (GetUtcNow() + timeout). A system clock stepped forward while |
| | | 1510 | | // a publish waited here — an NTP correction, a VM resumed or migrated — made `remaining` |
| | | 1511 | | // non-positive at once, the loop body never ran, and TryConfirmDeliveryAsync went straight |
| | | 1512 | | // to TryClaimForRecoveryAsync: the message was claimed for lost-subscriber recovery under |
| | | 1513 | | // a live waiter the dispatch loop was about to deliver it to. This is the same rule the |
| | | 1514 | | // poll deadlines below already follow (_pollArmedAt); TimeProvider's timestamp keeps the |
| | | 1515 | | // wait drivable by a virtual clock, which a raw Stopwatch would not. |
| | 560 | 1516 | | var startedAt = _timeProvider.GetTimestamp(); |
| | 653 | 1517 | | TimeSpan Remaining() => _options.DeliveryConfirmationTimeout - _timeProvider.GetElapsedTime(startedAt); |
| | | 1518 | | |
| | | 1519 | | // One `remaining` computation drives both the loop condition and the poll delay: the old |
| | | 1520 | | // shape tested the deadline twice, one line apart, so the code read as if two different |
| | | 1521 | | // conditions mattered when the second could only ever agree with the first. |
| | | 1522 | | // |
| | | 1523 | | // The fast-path wait is a WaitAsync on the confirmation rather than a fresh Task.Delay |
| | | 1524 | | // raced by WhenAny. WhenAny abandoned its loser every iteration, so the overwhelmingly |
| | | 1525 | | // common same-process delivery left a live timer entry and a registration on the caller's |
| | | 1526 | | // token behind on every publish; WaitAsync tears its timer down when the confirmation |
| | | 1527 | | // wins. A lapsed poll interval surfaces as TimeoutException, which is the loop condition, |
| | | 1528 | | // not a failure. |
| | 560 | 1529 | | for (var remaining = Remaining(); |
| | 653 | 1530 | | remaining > TimeSpan.Zero; |
| | 93 | 1531 | | remaining = Remaining()) |
| | | 1532 | | { |
| | 649 | 1533 | | var pollDelay = remaining < _options.DeliveryConfirmationPollInterval |
| | 649 | 1534 | | ? remaining |
| | 649 | 1535 | | : _options.DeliveryConfirmationPollInterval; |
| | | 1536 | | |
| | | 1537 | | try |
| | | 1538 | | { |
| | | 1539 | | // Fast path: an in-process delivery trips the completion and we return without a query. |
| | 649 | 1540 | | return await confirmation.Delivered.WaitAsync(pollDelay, _timeProvider, cancellationToken).ConfigureAwai |
| | | 1541 | | } |
| | 130 | 1542 | | catch (TimeoutException) |
| | | 1543 | | { |
| | | 1544 | | // Nothing local within this poll interval; fall through to the store check. |
| | 130 | 1545 | | } |
| | | 1546 | | |
| | | 1547 | | // Slow path: a delivery in another process only set acked_at, so poll for it. |
| | 130 | 1548 | | if (await _store.IsMessageAcknowledgedAsync(confirmation.MessageId, cancellationToken).ConfigureAwait(false) |
| | 35 | 1549 | | return true; |
| | | 1550 | | } |
| | | 1551 | | |
| | 4 | 1552 | | return confirmation.Delivered.IsCompletedSuccessfully |
| | 4 | 1553 | | || await _store.IsMessageAcknowledgedAsync(confirmation.MessageId, cancellationToken).ConfigureAwait(false); |
| | 558 | 1554 | | } |
| | | 1555 | | |
| | | 1556 | | private static string SerializeRawSuccessEnvelope(string payloadJson) |
| | | 1557 | | { |
| | 20 | 1558 | | JsonSafety.ThrowIfClearlyNotJson(payloadJson); |
| | | 1559 | | |
| | 20 | 1560 | | var buffer = new ArrayBufferWriter<byte>(); |
| | 20 | 1561 | | using (var writer = new Utf8JsonWriter(buffer)) |
| | | 1562 | | { |
| | 20 | 1563 | | writer.WriteStartObject(); |
| | 20 | 1564 | | writer.WriteNumber("SchemaVersion", AsyncResponseEnvelopeSchema.Current); |
| | 20 | 1565 | | writer.WriteBoolean("Success", true); |
| | 20 | 1566 | | writer.WritePropertyName("Payload"); |
| | 20 | 1567 | | writer.WriteRawValue(payloadJson); |
| | 20 | 1568 | | writer.WriteNull("ExceptionMessage"); |
| | 20 | 1569 | | writer.WriteNull("ExceptionStackTrace"); |
| | 20 | 1570 | | writer.WriteEndObject(); |
| | 20 | 1571 | | } |
| | | 1572 | | |
| | 20 | 1573 | | return Encoding.UTF8.GetString(buffer.WrittenSpan); |
| | | 1574 | | } |
| | | 1575 | | |
| | | 1576 | | [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage] |
| | | 1577 | | private static void OnWaiterTimeout(object? state) |
| | | 1578 | | => ((IWaiterTimeoutState)state!).Schedule(); |
| | | 1579 | | |
| | | 1580 | | private async Task HandleWaiterTimeoutAsync<T>( |
| | | 1581 | | DbSubscription<T> subscription, |
| | | 1582 | | Activity? activity, |
| | | 1583 | | string correlationId) where T : IAsyncResponsePayload |
| | | 1584 | | { |
| | 5 | 1585 | | _logger.LogWarning("Timed out waiting for {Provider} response for correlationId {CorrelationId}.", _providerName |
| | 5 | 1586 | | AsyncResponseDiagnostics.SetError(activity, "timeout", $"Timed out waiting for response for correlationId {corre |
| | 5 | 1587 | | AsyncResponseDiagnostics.RecordWaiterTimeout(_activityTag); |
| | 5 | 1588 | | await subscription.DrainThenCleanupAsync( |
| | 5 | 1589 | | deleteRecoveryState: true, |
| | 5 | 1590 | | new TimeoutException($"Timed out waiting for response for correlationId {correlationId}.")).ConfigureAwait(f |
| | 5 | 1591 | | } |
| | | 1592 | | |
| | | 1593 | | private interface IWaiterTimeoutState |
| | | 1594 | | { |
| | | 1595 | | void Schedule(); |
| | | 1596 | | } |
| | | 1597 | | |
| | | 1598 | | [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage] |
| | | 1599 | | private sealed class WaiterTimeoutState<T>( |
| | | 1600 | | DbAsyncResponseChannelBase owner, |
| | | 1601 | | DbSubscription<T> subscription, |
| | | 1602 | | Activity? activity, |
| | | 1603 | | string correlationId) : IWaiterTimeoutState where T : IAsyncResponsePayload |
| | | 1604 | | { |
| | | 1605 | | public void Schedule() |
| | | 1606 | | => _ = Task.Run(async () => |
| | | 1607 | | { |
| | | 1608 | | try |
| | | 1609 | | { |
| | | 1610 | | await owner.HandleWaiterTimeoutAsync(subscription, activity, correlationId).ConfigureAwait(false); |
| | | 1611 | | } |
| | | 1612 | | catch (Exception ex) |
| | | 1613 | | { |
| | | 1614 | | // Fire-and-forget: nothing awaits this task, so an escaped fault would vanish. |
| | | 1615 | | owner._logger.LogError(ex, "Error handling {Provider} waiter timeout for correlationId {CorrelationI |
| | | 1616 | | } |
| | | 1617 | | }); |
| | | 1618 | | } |
| | | 1619 | | |
| | | 1620 | | [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage] |
| | | 1621 | | private sealed class LocalDispatchWorkItem( |
| | | 1622 | | DbAsyncResponseChannelBase owner, |
| | | 1623 | | DbChannelMessage message, |
| | | 1624 | | IReadOnlyList<IDbSubscription> subscriptions, |
| | | 1625 | | CancellationToken cancellationToken) |
| | | 1626 | | { |
| | | 1627 | | public async Task InvokeAsync() |
| | | 1628 | | { |
| | | 1629 | | try |
| | | 1630 | | { |
| | | 1631 | | await owner.DispatchMessageToSubscribersAsync(message, subscriptions, cancellationToken).ConfigureAwait( |
| | | 1632 | | } |
| | | 1633 | | catch (Exception ex) when (ex is not OperationCanceledException || !cancellationToken.IsCancellationRequeste |
| | | 1634 | | { |
| | | 1635 | | if (owner._subscriptions.TryGetValue(message.CorrelationId, out var group) |
| | | 1636 | | && owner._dispatchScans.TryGetValue(group, out var scan)) |
| | | 1637 | | { |
| | | 1638 | | Interlocked.Exchange(ref scan.RewindRequested, 1); |
| | | 1639 | | owner.ScheduleBackpressureRescan(message.CorrelationId, cancellationToken); |
| | | 1640 | | } |
| | | 1641 | | owner._logger.LogDebug( |
| | | 1642 | | ex, |
| | | 1643 | | "Local {Provider} response dispatch failed for correlationId {CorrelationId}; {RetryHint}.", |
| | | 1644 | | owner._providerName, |
| | | 1645 | | message.CorrelationId, |
| | | 1646 | | owner._localDispatchRetryHint); |
| | | 1647 | | } |
| | | 1648 | | } |
| | | 1649 | | } |
| | | 1650 | | |
| | | 1651 | | /// <inheritdoc /> |
| | | 1652 | | public async ValueTask DisposeAsync() |
| | | 1653 | | { |
| | | 1654 | | CancellationTokenSource? cts; |
| | | 1655 | | Task? listenTask; |
| | | 1656 | | Task? dispatchTask; |
| | | 1657 | | Task? heartbeatTask; |
| | 2128 | 1658 | | lock (_listenerGate) |
| | | 1659 | | { |
| | | 1660 | | // Set under the gate so EnsureListenerStarted can never observe "not disposed" and |
| | | 1661 | | // then recreate the CTS/loops this teardown is about to stop. |
| | 2128 | 1662 | | _disposed = true; |
| | 2128 | 1663 | | cts = _listenerCts; |
| | 2128 | 1664 | | listenTask = _listenTask; |
| | 2128 | 1665 | | dispatchTask = _dispatchTask; |
| | 2128 | 1666 | | heartbeatTask = _heartbeatTask; |
| | 2128 | 1667 | | _listenerCts = null; |
| | 2128 | 1668 | | _listenTask = null; |
| | 2128 | 1669 | | _dispatchTask = null; |
| | 2128 | 1670 | | _heartbeatTask = null; |
| | 2128 | 1671 | | } |
| | | 1672 | | |
| | 2128 | 1673 | | if (cts is not null) |
| | | 1674 | | { |
| | 364 | 1675 | | await cts.CancelAsync().ConfigureAwait(false); |
| | | 1676 | | try |
| | | 1677 | | { |
| | 364 | 1678 | | await Task.WhenAll(new[] { listenTask, dispatchTask, heartbeatTask }.OfType<Task>()).ConfigureAwait(fals |
| | 345 | 1679 | | } |
| | 19 | 1680 | | catch (OperationCanceledException) |
| | | 1681 | | { |
| | 19 | 1682 | | } |
| | 364 | 1683 | | cts.Dispose(); |
| | | 1684 | | } |
| | | 1685 | | |
| | 4300 | 1686 | | foreach (var (correlationId, group) in _subscriptions.ToArray()) |
| | | 1687 | | { |
| | 88 | 1688 | | foreach (var subscription in group.Values.ToArray()) |
| | 22 | 1689 | | await subscription.DrainThenCleanupAsync(deleteRecoveryState: false).ConfigureAwait(false); |
| | 22 | 1690 | | await _executors.RemoveAsync(ChannelName(correlationId)).ConfigureAwait(false); |
| | 22 | 1691 | | } |
| | | 1692 | | |
| | | 1693 | | // Retirements for subscriptions that cleaned themselves up (a response landing during |
| | | 1694 | | // shutdown) unlink their correlation id before scheduling, so the loop above never sees |
| | | 1695 | | // them. Awaiting them here is what makes disposal mean "every executor is retired" rather |
| | | 1696 | | // than "every executor still in the map is retired". Their bodies swallow, so this cannot |
| | | 1697 | | // throw; the drain budgets inside RemoveAsync bound how long it can take. |
| | 2128 | 1698 | | var retirements = _pendingRetirements.Keys.ToArray(); |
| | 2128 | 1699 | | if (retirements.Length > 0) |
| | 7 | 1700 | | await Task.WhenAll(retirements).ConfigureAwait(false); |
| | 2128 | 1701 | | } |
| | | 1702 | | |
| | | 1703 | | /// <summary>Scopes an in-process delivery completion; <see cref="Dispose"/> unregisters it.</summary> |
| | | 1704 | | private protected readonly struct PendingConfirmation( |
| | | 1705 | | DbAsyncResponseChannelBase owner, |
| | | 1706 | | Guid messageId, |
| | | 1707 | | TaskCompletionSource<bool> tcs) : IDisposable |
| | | 1708 | | { |
| | 138 | 1709 | | public Guid MessageId => messageId; |
| | 653 | 1710 | | public Task<bool> Delivered => tcs.Task; |
| | 557 | 1711 | | public void Dispose() => owner._pendingConfirmations.TryRemove(messageId, out _); |
| | | 1712 | | } |
| | | 1713 | | |
| | | 1714 | | private protected interface IDbSubscription |
| | | 1715 | | { |
| | | 1716 | | Guid Id { get; } |
| | | 1717 | | DateTimeOffset StartedAtUtc { get; } |
| | | 1718 | | long StartedSeq { get; } |
| | | 1719 | | bool Dropped { get; } |
| | | 1720 | | Func<DbChannelMessage, Task> ProcessUnderContextAsync { get; set; } |
| | | 1721 | | bool HasSeen(Guid messageId); |
| | | 1722 | | bool MarkSeen(Guid messageId); |
| | | 1723 | | void PruneSeen(DateTimeOffset cutoffUtc); |
| | | 1724 | | Task ProcessAsync(DbChannelMessage message); |
| | | 1725 | | ValueTask CleanupOnceAsync(bool deleteRecoveryState); |
| | | 1726 | | ValueTask DrainThenCleanupAsync(bool deleteRecoveryState, Exception? terminalIfUndelivered = null); |
| | | 1727 | | ValueTask DropLocalAsync(CancellationToken cancellationToken); |
| | | 1728 | | } |
| | | 1729 | | |
| | | 1730 | | private sealed class DbSubscription<T> : IDbSubscription where T : IAsyncResponsePayload |
| | | 1731 | | { |
| | | 1732 | | private readonly DbAsyncResponseChannelBase _owner; |
| | | 1733 | | private readonly string _correlationId; |
| | | 1734 | | private readonly Func<T, ValueTask<bool>> _completionPredicate; |
| | | 1735 | | private readonly TaskCompletionSource<T> _tcs; |
| | | 1736 | | private readonly Activity? _activity; |
| | 498 | 1737 | | private readonly HashSet<Guid> _seen = []; |
| | 498 | 1738 | | private readonly Queue<(Guid Id, DateTimeOffset SeenAtUtc)> _seenOrder = []; |
| | 498 | 1739 | | private readonly object _seenGate = new(); |
| | | 1740 | | private int _cleanupStarted; |
| | | 1741 | | private volatile bool _dropped; |
| | | 1742 | | |
| | | 1743 | | /// <summary>True once cleanup began — the arm-last waiter-timeout guard reads this.</summary> |
| | 456 | 1744 | | internal bool CleanupStarted => Volatile.Read(ref _cleanupStarted) != 0; |
| | 498 | 1745 | | private readonly object _cleanupGate = new(); |
| | | 1746 | | private Task? _cleanupTask; |
| | | 1747 | | |
| | 498 | 1748 | | public DbSubscription( |
| | 498 | 1749 | | DbAsyncResponseChannelBase owner, |
| | 498 | 1750 | | string correlationId, |
| | 498 | 1751 | | Guid registrationId, |
| | 498 | 1752 | | DateTimeOffset startedAtUtc, |
| | 498 | 1753 | | long startedSeq, |
| | 498 | 1754 | | Func<T, ValueTask<bool>> completionPredicate, |
| | 498 | 1755 | | TaskCompletionSource<T> tcs, |
| | 498 | 1756 | | Activity? activity) |
| | | 1757 | | { |
| | 498 | 1758 | | _owner = owner; |
| | 498 | 1759 | | _correlationId = correlationId; |
| | 498 | 1760 | | Id = registrationId; |
| | 498 | 1761 | | StartedAtUtc = startedAtUtc; |
| | 498 | 1762 | | StartedSeq = startedSeq; |
| | 498 | 1763 | | _completionPredicate = completionPredicate; |
| | 498 | 1764 | | _tcs = tcs; |
| | 498 | 1765 | | _activity = activity; |
| | 498 | 1766 | | ProcessUnderContextAsync = ProcessAsync; |
| | 498 | 1767 | | } |
| | | 1768 | | |
| | 10768 | 1769 | | public Guid Id { get; } |
| | 14492 | 1770 | | public DateTimeOffset StartedAtUtc { get; } |
| | 5 | 1771 | | public long StartedSeq { get; } |
| | 14205 | 1772 | | public bool Dropped => _dropped; |
| | 1374 | 1773 | | public Func<ValueTask>? TimeoutRegistration { get; set; } |
| | 918 | 1774 | | public CancellationTokenSource? TimeoutCancellation { get; set; } |
| | 1512 | 1775 | | public Func<DbChannelMessage, Task> ProcessUnderContextAsync { get; set; } |
| | | 1776 | | |
| | | 1777 | | public bool HasSeen(Guid messageId) |
| | | 1778 | | { |
| | 939 | 1779 | | lock (_seenGate) |
| | | 1780 | | { |
| | 939 | 1781 | | return _seen.Contains(messageId); |
| | | 1782 | | } |
| | 939 | 1783 | | } |
| | | 1784 | | |
| | | 1785 | | public bool MarkSeen(Guid messageId) |
| | | 1786 | | { |
| | 565 | 1787 | | lock (_seenGate) |
| | | 1788 | | { |
| | 565 | 1789 | | if (!_seen.Add(messageId)) |
| | 3 | 1790 | | return false; |
| | | 1791 | | |
| | | 1792 | | // Use the local observation time, not the database creation time. This keeps the |
| | | 1793 | | // pruning queue monotonic and avoids immediate eviction when app and DB clocks differ. |
| | | 1794 | | // It MUST come from the same clock PruneSeen's cutoff is computed on: mixing a |
| | | 1795 | | // wall-clock stamp with a TimeProvider cutoff makes every entry look either |
| | | 1796 | | // permanently fresh or permanently expired under a virtual clock. |
| | 562 | 1797 | | _seenOrder.Enqueue((messageId, _owner._timeProvider.GetUtcNow())); |
| | 562 | 1798 | | return true; |
| | | 1799 | | } |
| | 565 | 1800 | | } |
| | | 1801 | | |
| | | 1802 | | public void PruneSeen(DateTimeOffset cutoffUtc) |
| | | 1803 | | { |
| | 6348 | 1804 | | lock (_seenGate) |
| | | 1805 | | { |
| | 6350 | 1806 | | while (_seenOrder.TryPeek(out var entry) && entry.SeenAtUtc < cutoffUtc) |
| | | 1807 | | { |
| | 2 | 1808 | | _seenOrder.Dequeue(); |
| | 2 | 1809 | | _seen.Remove(entry.Id); |
| | 2 | 1810 | | } |
| | 6348 | 1811 | | } |
| | 6348 | 1812 | | } |
| | | 1813 | | |
| | | 1814 | | public async Task ProcessAsync(DbChannelMessage message) |
| | | 1815 | | { |
| | 574 | 1816 | | if (_dropped) |
| | 2 | 1817 | | return; |
| | | 1818 | | |
| | 572 | 1819 | | var finished = false; |
| | | 1820 | | try |
| | | 1821 | | { |
| | | 1822 | | // JsonSafety, not the raw reader: a parse failure is logged below and handed to the |
| | | 1823 | | // waiter, and the reader's own message quotes inbound property names and dictionary |
| | | 1824 | | // keys (docs/security.md, "never logs a message body"). Size and position only. |
| | | 1825 | | // A header-only sweep row never reaches delivery: the sweep hydrates it first. |
| | 572 | 1826 | | var envelopeJson = message.EnvelopeJson |
| | 572 | 1827 | | ?? throw new InvalidOperationException($"The {_owner._providerName} channel message {message.Id} rea |
| | 572 | 1828 | | var envelope = JsonSafety.SafeDeserialize(envelopeJson, AsyncResponseEnvelopeJson.TypeInfo<T>()); |
| | 567 | 1829 | | if (envelope is null) |
| | | 1830 | | { |
| | 3 | 1831 | | finished = true; |
| | 3 | 1832 | | var error = new JsonException($"Failed to deserialize envelope for correlationId {_correlationId}.") |
| | 3 | 1833 | | AsyncResponseDiagnostics.SetError(_activity, "deserialize_failure", error.Message); |
| | 3 | 1834 | | _tcs.TrySetException(error); |
| | | 1835 | | } |
| | 564 | 1836 | | else if (!AsyncResponseEnvelopeSchema.IsReadable(envelope.SchemaVersion)) |
| | | 1837 | | { |
| | 3 | 1838 | | finished = true; |
| | 3 | 1839 | | var error = new InvalidOperationException( |
| | 3 | 1840 | | $"Response envelope for correlationId {_correlationId} has schema version {envelope.SchemaVersio |
| | 3 | 1841 | | $"which this build does not support (current: {AsyncResponseEnvelopeSchema.Current})."); |
| | 3 | 1842 | | AsyncResponseDiagnostics.SetError(_activity, "schema_mismatch", error.Message); |
| | 3 | 1843 | | _tcs.TrySetException(error); |
| | | 1844 | | } |
| | 561 | 1845 | | else if (!envelope.Success) |
| | | 1846 | | { |
| | 5 | 1847 | | finished = true; |
| | 5 | 1848 | | var remoteFailure = new Exception(envelope.ExceptionMessage ?? "Unknown error during asynchronous pr |
| | 5 | 1849 | | if (!string.IsNullOrEmpty(envelope.ExceptionStackTrace)) |
| | 3 | 1850 | | remoteFailure.Data["RemoteStackTrace"] = RemoteStackTrace.Cap(envelope.ExceptionStackTrace, _own |
| | 5 | 1851 | | AsyncResponseDiagnostics.SetError(_activity, "remote_failure", remoteFailure.Message); |
| | 5 | 1852 | | _tcs.TrySetException(remoteFailure); |
| | | 1853 | | } |
| | | 1854 | | else |
| | | 1855 | | { |
| | 556 | 1856 | | finished = await _completionPredicate(envelope.Payload!).ConfigureAwait(false); |
| | 555 | 1857 | | if (finished) |
| | 437 | 1858 | | _tcs.TrySetResult(envelope.Payload!); |
| | | 1859 | | } |
| | 566 | 1860 | | } |
| | 6 | 1861 | | catch (Exception ex) |
| | | 1862 | | { |
| | 6 | 1863 | | finished = true; |
| | 6 | 1864 | | _owner._logger.LogError(ex, "Error processing {Provider} response for correlationId {CorrelationId}.", _ |
| | 6 | 1865 | | AsyncResponseDiagnostics.SetError(_activity, ex); |
| | 6 | 1866 | | _tcs.TrySetException(ex); |
| | 6 | 1867 | | } |
| | | 1868 | | finally |
| | | 1869 | | { |
| | 572 | 1870 | | if (finished) |
| | 454 | 1871 | | await CleanupOnceAsync(deleteRecoveryState: true).ConfigureAwait(false); |
| | | 1872 | | } |
| | 574 | 1873 | | } |
| | | 1874 | | |
| | | 1875 | | /// <summary> |
| | | 1876 | | /// Task-latched so EVERY caller completes only when the one real cleanup has finished — |
| | | 1877 | | /// a fire-once flag alone would let a disposing waiter racing the timeout return before |
| | | 1878 | | /// the response task was settled. |
| | | 1879 | | /// </summary> |
| | | 1880 | | public ValueTask CleanupOnceAsync(bool deleteRecoveryState) |
| | | 1881 | | { |
| | | 1882 | | Task task; |
| | 946 | 1883 | | lock (_cleanupGate) |
| | | 1884 | | { |
| | 946 | 1885 | | task = _cleanupTask ??= CleanupCoreAsync(deleteRecoveryState); |
| | 946 | 1886 | | } |
| | | 1887 | | |
| | 946 | 1888 | | return task.IsCompletedSuccessfully ? ValueTask.CompletedTask : new ValueTask(task); |
| | | 1889 | | } |
| | | 1890 | | |
| | | 1891 | | /// <summary> |
| | | 1892 | | /// Dispose-path cleanup: DRAINS the per-correlation serial executor before settling. A |
| | | 1893 | | /// delivery may be mid <c>Until</c>-predicate holding a message the claim already acked; |
| | | 1894 | | /// the marker work item completes only after that in-flight item finished, so by the time |
| | | 1895 | | /// cleanup cancels, the task is either settled by the delivery or genuinely undelivered — |
| | | 1896 | | /// never a cancellation stealing a consumed response. Must NOT be called from dispatch |
| | | 1897 | | /// code (which runs ON the executor): the dispatch-triggered cleanup calls |
| | | 1898 | | /// <see cref="CleanupOnceAsync"/> directly, its task already settled. |
| | | 1899 | | /// <para> |
| | | 1900 | | /// The drain is bounded by <c>DisposalDrainTimeout</c> — a single budget covering marker |
| | | 1901 | | /// ADMISSION too, since a full bounded queue behind a wedged item blocks the enqueue |
| | | 1902 | | /// itself. A lapsed budget must not fall back to the cleanup's cancel: the wedged delivery |
| | | 1903 | | /// holds a message the claim already consumed, and "canceled" would tell a re-attaching |
| | | 1904 | | /// caller nothing was delivered. It faults the task with the explicit indeterminate |
| | | 1905 | | /// contract instead, routing durable flows to a fresh idempotent restart. An enqueue |
| | | 1906 | | /// suppressed by the registry's tombstone is the opposite case — the retired executor |
| | | 1907 | | /// finished everything it ever admitted, so nothing is in flight and the plain cancel |
| | | 1908 | | /// below is truthful. |
| | | 1909 | | /// </para> |
| | | 1910 | | /// </summary> |
| | | 1911 | | public async ValueTask DrainThenCleanupAsync(bool deleteRecoveryState, Exception? terminalIfUndelivered = null) |
| | | 1912 | | { |
| | 490 | 1913 | | if (Volatile.Read(ref _cleanupStarted) == 0) |
| | | 1914 | | { |
| | 24 | 1915 | | var drainTimeout = _owner._options.DisposalDrainTimeout; |
| | 24 | 1916 | | var drained = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); |
| | | 1917 | | try |
| | | 1918 | | { |
| | 24 | 1919 | | using var budget = new CancellationTokenSource(drainTimeout); |
| | 24 | 1920 | | var accepted = await _owner._executors.EnqueueAsync(_owner.ChannelName(_correlationId), () => |
| | 24 | 1921 | | { |
| | 24 | 1922 | | drained.TrySetResult(); |
| | 24 | 1923 | | return Task.CompletedTask; |
| | 24 | 1924 | | }, budget.Token).ConfigureAwait(false); |
| | 24 | 1925 | | if (accepted) |
| | 24 | 1926 | | await drained.Task.WaitAsync(budget.Token).ConfigureAwait(false); |
| | 23 | 1927 | | } |
| | 1 | 1928 | | catch (Exception drainEx) |
| | | 1929 | | { |
| | | 1930 | | // Budget lapse — or an unforeseen drain failure: either way the marker never |
| | | 1931 | | // ran, so an in-flight delivery cannot be ruled out (only accepted=false |
| | | 1932 | | // proves the executor finished everything). Settlement unproven means the |
| | | 1933 | | // cleanup's cancel below would be a false "nothing was delivered" — fault |
| | | 1934 | | // with the explicit indeterminate contract instead. A TrySetResult from the |
| | | 1935 | | // late-finishing dispatch loses against this and is dropped; its cleanup |
| | | 1936 | | // call is a no-op behind the latch. |
| | 1 | 1937 | | _owner._logger.LogWarning( |
| | 1 | 1938 | | "Disposal drain for {Provider} correlationId {CorrelationId} did not prove settlement within {Dr |
| | 1 | 1939 | | _owner._providerName, _correlationId, drainTimeout); |
| | 1 | 1940 | | AsyncResponseDiagnostics.SetError(_activity, "indeterminate_delivery", "Disposal drain did not prove |
| | 1 | 1941 | | if (drainEx is not OperationCanceledException) |
| | 0 | 1942 | | _owner._logger.LogDebug(drainEx, "Dispatch drain failed for correlationId {CorrelationId}.", _co |
| | 1 | 1943 | | _tcs.TrySetException(new AsyncResponseIndeterminateDeliveryException(_correlationId, drainTimeout)); |
| | 1 | 1944 | | } |
| | 24 | 1945 | | } |
| | | 1946 | | |
| | | 1947 | | // Settle AFTER the drain, never before it. A delivery already inside the per-correlation |
| | | 1948 | | // executor may hold a message the claim acked — the publisher was told "delivered" and |
| | | 1949 | | // the watermark excludes it from every later sweep, so it exists nowhere else. Faulting |
| | | 1950 | | // first let a timeout beat that in-flight delivery and report a consumed response as a |
| | | 1951 | | // timeout; TrySet loses here if the delivery won, which is the whole point. (A lapsed |
| | | 1952 | | // drain budget has already faulted the task as indeterminate above, and TrySet is a |
| | | 1953 | | // no-op behind it.) |
| | 490 | 1954 | | if (terminalIfUndelivered is not null) |
| | 9 | 1955 | | _tcs.TrySetException(terminalIfUndelivered); |
| | | 1956 | | |
| | 490 | 1957 | | await CleanupOnceAsync(deleteRecoveryState).ConfigureAwait(false); |
| | 490 | 1958 | | } |
| | | 1959 | | |
| | | 1960 | | private async Task CleanupCoreAsync(bool deleteRecoveryState) |
| | | 1961 | | { |
| | | 1962 | | // The flag is kept alongside the task latch: dispatch cores and white-box tests gate |
| | | 1963 | | // on it, and a pre-set flag (test isolation) must keep skipping the network cleanup. |
| | 491 | 1964 | | if (Interlocked.Exchange(ref _cleanupStarted, 1) != 0) |
| | 29 | 1965 | | return; |
| | | 1966 | | |
| | | 1967 | | // A waiter disposed before any terminal signal must not leave ResponseTask pending |
| | | 1968 | | // forever for callers that hold it directly — the timeout dies with this cleanup, so |
| | | 1969 | | // nothing else could ever complete the task. This also covers channel DisposeAsync at |
| | | 1970 | | // host shutdown, which runs this cleanup over every in-flight subscription and would |
| | | 1971 | | // otherwise hang still-awaiting WaitAsync callers. Cancellation is a no-op after a |
| | | 1972 | | // normal completion, timeout, fault, or a delivery drained by DrainThenCleanupAsync. |
| | 462 | 1973 | | _tcs.TrySetCanceled(); |
| | | 1974 | | |
| | | 1975 | | try |
| | | 1976 | | { |
| | 462 | 1977 | | _dropped = true; |
| | | 1978 | | |
| | | 1979 | | try |
| | | 1980 | | { |
| | | 1981 | | // Delete the recovery state BEFORE removing the subscription (locally and in the |
| | | 1982 | | // subscriber store). In the reverse order a publish landing in the window sees |
| | | 1983 | | // "no subscriber, state present" and fires a spurious recovery callback for a wait |
| | | 1984 | | // that already reached a terminal state. In this order the window shows a |
| | | 1985 | | // subscriber that drops the message — a late or duplicate terminal message is |
| | | 1986 | | // droppable; a resurrected recovery callback is not. (Shutdown/redeploy paths pass |
| | | 1987 | | // deleteRecoveryState: false and keep the state for lost-subscriber recovery.) |
| | 462 | 1988 | | if (deleteRecoveryState) |
| | 460 | 1989 | | await _owner._recoveryStateStore.TryDeleteAsync(_correlationId, Id).ConfigureAwait(false); |
| | 460 | 1990 | | } |
| | 2 | 1991 | | catch (Exception ex) |
| | | 1992 | | { |
| | | 1993 | | // Best-effort: the state expires on its own, and a transient store failure must |
| | | 1994 | | // not skip the local teardown below. |
| | 2 | 1995 | | _owner._logger.LogError(ex, "Failed to delete {Provider} recovery state for correlationId {Correlati |
| | 2 | 1996 | | } |
| | | 1997 | | |
| | | 1998 | | try |
| | | 1999 | | { |
| | 462 | 2000 | | await _owner._store.DeleteSubscriberAsync(_correlationId, Id, CancellationToken.None).ConfigureAwait |
| | 456 | 2001 | | } |
| | 6 | 2002 | | catch (Exception ex) |
| | | 2003 | | { |
| | | 2004 | | // Best-effort: an orphaned subscriber record ages out via the heartbeat timeout. |
| | 6 | 2005 | | _owner._logger.LogError(ex, "Failed to delete {Provider} subscriber {SubscriberRecord} for correlati |
| | 6 | 2006 | | } |
| | | 2007 | | } |
| | | 2008 | | finally |
| | | 2009 | | { |
| | | 2010 | | // Purely local teardown runs no matter which network call above failed — the |
| | | 2011 | | // cleanup latch is already set, so a skipped removal would leak the subscription |
| | | 2012 | | // map entry and the executor until process exit. |
| | 462 | 2013 | | _owner.RemoveSubscription(_correlationId, Id); |
| | | 2014 | | |
| | | 2015 | | // Schedule the executor retirement on the thread pool; do not await directly — |
| | | 2016 | | // dispatch-loop deliveries run this cleanup ON the executor, and RemoveAsync waits |
| | | 2017 | | // for the executor's drain loop to finish, which would be a circular await. |
| | | 2018 | | // TRACKED, though: RemoveSubscription above already unlinked this correlation id, |
| | | 2019 | | // so DisposeAsync's own retirement loop will not see it, and an untracked |
| | | 2020 | | // retirement could still be inside its 30-second drain budget when the host tears |
| | | 2021 | | // down the logger and exits — logging into a disposed logger, or being killed |
| | | 2022 | | // mid-drain. DisposeAsync awaits whatever is still outstanding here. |
| | 462 | 2023 | | var channelName = _owner.ChannelName(_correlationId); |
| | 462 | 2024 | | _owner.TrackRetirement(Task.Run(async () => |
| | 462 | 2025 | | { |
| | 462 | 2026 | | try |
| | 462 | 2027 | | { |
| | 462 | 2028 | | await _owner._executors.RemoveAsync(channelName).ConfigureAwait(false); |
| | 462 | 2029 | | } |
| | 0 | 2030 | | catch (Exception ex) |
| | 462 | 2031 | | { |
| | 0 | 2032 | | _owner._logger.LogError(ex, "Failed to retire the executor for channel {Channel}.", channelName) |
| | 0 | 2033 | | } |
| | 924 | 2034 | | })); |
| | | 2035 | | |
| | 462 | 2036 | | if (TimeoutRegistration is not null) |
| | 456 | 2037 | | await TimeoutRegistration().ConfigureAwait(false); |
| | 462 | 2038 | | TimeoutCancellation?.Dispose(); |
| | 462 | 2039 | | _activity?.Dispose(); |
| | | 2040 | | } |
| | 491 | 2041 | | } |
| | | 2042 | | |
| | | 2043 | | public async ValueTask DropLocalAsync(CancellationToken cancellationToken) |
| | | 2044 | | { |
| | 4 | 2045 | | _dropped = true; |
| | 4 | 2046 | | await _owner._store.DeleteSubscriberAsync(_correlationId, Id, cancellationToken).ConfigureAwait(false); |
| | 4 | 2047 | | } |
| | | 2048 | | } |
| | | 2049 | | } |