< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.InMemoryWorkerTransportOptions
Assembly: AsyncResponse.Core
File(s): /_/src/AsyncResponse.Core/InMemoryWorkerTransport.cs
Line coverage
100%
Covered lines: 25
Uncovered lines: 0
Coverable lines: 25
Total lines: 888
Line coverage: 100%
Branch coverage
100%
Covered branches: 16
Total branches: 16
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_QueueCapacity()100%11100%
get_WorkerCount()100%11100%
get_InJobOverflowCapacity()100%11100%
get_DelayedJobCapacity()100%11100%
get_MaxDeliveryAttempts()100%11100%
get_RetryBaseDelay()100%11100%
get_RetryMaxDelay()100%11100%
Validate()100%1616100%

File(s)

/_/src/AsyncResponse.Core/InMemoryWorkerTransport.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using System.Diagnostics;
 5using System.Threading.Channels;
 6using System.Collections.Concurrent;
 7
 8namespace AsyncResponse;
 9
 10/// <summary>
 11/// An in-memory <see cref="IWorkerTransport"/> backed by a bounded
 12/// <see cref="Channel{T}"/>, registered by <c>AddAsyncResponse().WithInMemoryTransport()</c>.
 13/// Jobs run in the current process and survive only as long as it does — use a broker-backed
 14/// transport for durability. Intended for development, tests, and single-node deployments.
 15/// <para>
 16/// Envelopes have broker wire parity: each publish serializes the job to its wire JSON and the
 17/// worker receives an instance materialized from it — <c>[JsonIgnore]</c> argument state is
 18/// excluded, post-publish mutations are invisible, and a non-serializable argument throws at
 19/// <see cref="PublishAsync(WorkerJobEnvelope, CancellationToken)"/> — so behavior observed here
 20/// carries over unchanged to every broker-backed transport.
 21/// </para>
 22/// <para>
 23/// Because the job stays in-process, the enqueuer's <see cref="ExecutionContext"/> is captured and
 24/// the job runs under it (see <see cref="InMemoryWorkerHost"/>), so ambient <see cref="AsyncLocal{T}"/>
 25/// state — trace id, principal, logging scope — flows automatically without any serializable
 26/// context propagator.
 27/// </para>
 28/// </summary>
 29public sealed class InMemoryWorkerTransport : IWorkerTransport, IDelayedWorkerTransport
 30{
 31    private readonly Channel<QueuedJob> _queue;
 32    private readonly TimeProvider _timeProvider;
 33    private readonly object _delayedGate = new();
 34    private readonly Dictionary<DelayedJob, ITimer> _delayedJobs = [];
 35    private int _outstanding;
 36    private volatile bool _draining;
 37
 38    /// <summary>
 39    /// One slot per delayed job the transport may hold, from acceptance until the fired job has
 40    /// entered the queue (or was dropped). Neither queue bound covered delayed jobs: every
 41    /// scheduled publish retained its materialized envelope and captured execution context
 42    /// against no limit at all, and when a burst's timers fired, each started an asynchronous
 43    /// channel write that pended outside the bounded queue — so a flood of scheduled jobs grew
 44    /// the process without either configured capacity giving a signal. Bounded by
 45    /// <see cref="InMemoryWorkerTransportOptions.DelayedJobCapacity"/>: an external publisher
 46    /// waits for a slot (honoring its cancellation token); a publish from inside a running job
 47    /// is rejected instead, as its immediate follow-ups are at the overflow bound — a worker
 48    /// waiting for a slot that only a fired timer entering the queue (through that worker) frees
 49    /// would be waiting on itself.
 50    /// </summary>
 51    private readonly SemaphoreSlim _delayedSlots;
 52
 53    /// <summary>Creates a transport with default bounded-queue options.</summary>
 54    public InMemoryWorkerTransport()
 55        : this(Microsoft.Extensions.Options.Options.Create(new InMemoryWorkerTransportOptions()))
 56    {
 57    }
 58
 59    /// <summary>Creates a transport with configured capacity and worker concurrency.</summary>
 60    public InMemoryWorkerTransport(IOptions<InMemoryWorkerTransportOptions> options, TimeProvider? timeProvider = null)
 61    {
 62        Options = options.Value;
 63        Options.Validate();
 64        _timeProvider = timeProvider ?? TimeProvider.System;
 65        _queue = Channel.CreateBounded<QueuedJob>(new BoundedChannelOptions(Options.QueueCapacity)
 66        {
 67            SingleReader = Options.WorkerCount == 1,
 68            SingleWriter = false,
 69            FullMode = BoundedChannelFullMode.Wait,
 70            AllowSynchronousContinuations = false
 71        });
 72        _delayedSlots = new SemaphoreSlim(Options.DelayedJobCapacity, Options.DelayedJobCapacity);
 73        AsyncResponseDiagnostics.TrackInMemoryOverflow(this);
 74    }
 75
 76    internal ChannelReader<QueuedJob> Reader => _queue.Reader;
 77    internal InMemoryWorkerTransportOptions Options { get; }
 78
 79    /// <summary>
 80    /// Follow-up jobs published from inside a running job that did not fit the bounded queue. They
 81    /// are already counted in <c>_outstanding</c>, so the drain cannot complete the writer while
 82    /// any remain; a worker that finishes a job moves them into the queue while it has room and
 83    /// runs the rest itself before it reads the queue again (<see cref="TryTakeOverflow"/>). Bounded by
 84    /// <see cref="InMemoryWorkerTransportOptions.InJobOverflowCapacity"/> (tracked in
 85    /// <see cref="_overflowDepth"/>): unbounded, a fan-out handler could retain every follow-up
 86    /// envelope and its captured ExecutionContext until the process ran out of memory, with the
 87    /// configured queue capacity giving no signal at all.
 88    /// </summary>
 89    private readonly ConcurrentQueue<QueuedJob> _overflow = new();
 90    private int _overflowDepth;
 91
 92    /// <summary>
 93    /// Serializes <see cref="PumpOverflow"/>: with multiple workers, an unguarded
 94    /// peek/write/dequeue interleaving lets two pumpers write the same job twice and then dequeue
 95    /// a job that was never written — a duplicated execution plus a silently lost job.
 96    /// </summary>
 97    private readonly object _overflowPumpGate = new();
 98
 99    /// <summary>Follow-up jobs currently held past the queue's capacity (the overflow-depth gauge and test inspection).
 100    internal int OverflowDepth => Volatile.Read(ref _overflowDepth);
 101
 102    /// <summary>
 103    /// Moves overflow jobs into the queue while it has room, so idle workers can share them.
 104    /// Called by a worker before it reports a job finished. On its own this does NOT drain the
 105    /// overflow — see <see cref="TryTakeOverflow"/>.
 106    /// </summary>
 107    internal void PumpOverflow()
 108    {
 109        lock (_overflowPumpGate)
 110        {
 111            while (_overflow.TryPeek(out var queued))
 112            {
 113                if (!_queue.Writer.TryWrite(queued))
 114                    return;
 115
 116                _overflow.TryDequeue(out _);
 117                Interlocked.Decrement(ref _overflowDepth);
 118            }
 119        }
 120    }
 121
 122    /// <summary>
 123    /// Hands the oldest overflow job straight to the calling worker, bypassing the queue. A
 124    /// bounded channel gives a slot freed by a read directly to a producer already parked in
 125    /// <c>WriteAsync</c>, so while ANY external producer (or fired delayed job) is waiting for
 126    /// room, <see cref="PumpOverflow"/>'s <c>TryWrite</c> finds the queue full every single time:
 127    /// under sustained external load the overflow never drained at all. Follow-up work — a child
 128    /// flow's start, a parent's wake-up — starved behind an endless supply of NEW external work,
 129    /// the overflow filled, and every job that published a follow-up then failed at the bound and
 130    /// was eventually dropped. Follow-ups continue work the queue already admitted, so the worker
 131    /// that just finished a job runs them before it takes anything new; external producers stay
 132    /// parked meanwhile, which is the backpressure the capacity exists to apply.
 133    /// <para>
 134    /// Under the pump gate: an unguarded dequeue between a pumper's peek and its dequeue would run
 135    /// one job twice and drop the next.
 136    /// </para>
 137    /// </summary>
 138    internal bool TryTakeOverflow(out QueuedJob queued)
 139    {
 140        lock (_overflowPumpGate)
 141        {
 142            if (!_overflow.TryDequeue(out queued))
 143                return false;
 144
 145            Interlocked.Decrement(ref _overflowDepth);
 146            return true;
 147        }
 148    }
 149
 150    /// <summary>
 151    /// Admits a follow-up job to the overflow if it is under its capacity. The depth is reserved
 152    /// with a compare-and-swap BEFORE the enqueue, so concurrent in-job publishers (several
 153    /// workers) cannot overshoot the bound between a check and an add.
 154    /// </summary>
 155    private bool TryEnqueueOverflow(QueuedJob queued)
 156    {
 157        while (true)
 158        {
 159            var depth = Volatile.Read(ref _overflowDepth);
 160            if (depth >= Options.InJobOverflowCapacity)
 161                return false;
 162
 163            if (Interlocked.CompareExchange(ref _overflowDepth, depth + 1, depth) == depth)
 164            {
 165                _overflow.Enqueue(queued);
 166                return true;
 167            }
 168        }
 169    }
 170
 171    /// <summary>
 172    /// Marks the ambient flow as "executing a worker job", so a publish made beneath it is
 173    /// recognised as follow-up work rather than an external producer.
 174    /// </summary>
 175    internal static class InJobScope
 176    {
 177        private static readonly AsyncLocal<bool> _active = new();
 178
 179        public static bool IsActive => _active.Value;
 180
 181        /// <summary>
 182        /// Must be called from INSIDE the <see cref="ExecutionContext.Run"/> that restores the
 183        /// enqueue-time context (or, with no captured context, immediately around the execute
 184        /// call). Run replaces ambient AsyncLocal state wholesale with the captured snapshot, so a
 185        /// flag raised outside it never reaches the handler at all. Nothing clears it: Run restores
 186        /// the caller's context when it returns, and the no-context path raises it on a flow that
 187        /// ends with the job.
 188        /// </summary>
 189        public static void MarkActive() => _active.Value = true;
 190    }
 191    internal ILogger? DrainLogger { get; set; }
 192
 193    /// <summary>Jobs accepted but not yet finished (queued + executing). Test-harness idle probe.</summary>
 194    internal int OutstandingJobs => Volatile.Read(ref _outstanding);
 195
 196    /// <summary>
 197    /// Delayed jobs currently held: waiting on their due-time timer, or fired and waiting for
 198    /// queue room (the <c>asyncresponse.worker.inmemory_delayed_jobs</c> gauge and test inspection).
 199    /// </summary>
 200    internal int DelayedJobsHeld => Options.DelayedJobCapacity - _delayedSlots.CurrentCount;
 201
 202    /// <summary>The delayed jobs currently waiting on their due-time timers (test inspection).</summary>
 203    internal IReadOnlyList<WorkerJobEnvelope> SnapshotDelayedJobs()
 204    {
 205        lock (_delayedGate)
 206        {
 207            if (_delayedJobs.Count == 0)
 208                return [];
 209
 210            var envelopes = new WorkerJobEnvelope[_delayedJobs.Count];
 211            var index = 0;
 212            foreach (var delayed in _delayedJobs.Keys)
 213                envelopes[index++] = delayed.Envelope;
 214            return envelopes;
 215        }
 216    }
 217
 218    /// <summary>
 219    /// AsyncResponse.Testing only. From this call on, the shutdown drain RETAINS delayed jobs in
 220    /// the returned list instead of dropping them, and a delayed publish that arrives while the
 221    /// transport is draining (a flow suspending mid-drain) is retained instead of rejected —
 222    /// modeling the broker that keeps scheduled messages across a redeploy. A snapshot taken
 223    /// before the stop cannot do this: it misses both the drain-time publishes and any job armed
 224    /// between the snapshot and the drain. Read the list only after the stop has completed.
 225    /// </summary>
 226    internal List<WorkerJobEnvelope> BeginRetainingDelayedJobs()
 227    {
 228        lock (_delayedGate)
 229            return _drainRetention ??= [];
 230    }
 231
 232    private List<WorkerJobEnvelope>? _drainRetention;
 233
 234    /// <summary>
 235    /// Begins the shutdown drain. Called by <see cref="InMemoryWorkerHost"/> when the host starts
 236    /// stopping. The writer is deliberately NOT completed while anything is queued or running:
 237    /// accepted jobs were promised in-process execution, and a draining job may legitimately
 238    /// enqueue follow-up work (a durable-flow parent wake-up, a recovery re-enqueue) that must not
 239    /// hit a closed channel — losing it would strand the dependent flow with no redelivery to
 240    /// recover it. The last finishing job completes the writer instead, once the transport is idle.
 241    /// <para>
 242    /// Pending DELAYED jobs are different: their due time may be days away, and holding shutdown
 243    /// for them would hang the host. They are dropped with a warning — the in-memory transport is
 244    /// process-local by contract, so delayed jobs share the process's lifetime. A durable flow
 245    /// sleeping on such a wake-up must be resumed explicitly after restart (or use a broker
 246    /// transport, whose delayed messages survive). The test harness opts out of the drop via
 247    /// <see cref="BeginRetainingDelayedJobs"/> and re-publishes the retained jobs into the next
 248    /// incarnation.
 249    /// </para>
 250    /// </summary>
 251    internal void BeginShutdownDrain()
 252    {
 253        _draining = true;
 254
 255        KeyValuePair<DelayedJob, ITimer>[] pending;
 256        List<WorkerJobEnvelope>? retention;
 257        lock (_delayedGate)
 258        {
 259            pending = [.. _delayedJobs];
 260            _delayedJobs.Clear();
 261            retention = _drainRetention;
 262
 263            // Retention Adds run under the same lock as the delayed-publish Add: a flow
 264            // suspending mid-drain appends to this same List concurrently, and two
 265            // unsynchronized List<T>.Add calls can silently lose a wake-up or throw mid-grow.
 266            if (retention is not null)
 267            {
 268                foreach (var (job, _) in pending)
 269                    retention.Add(job.Envelope);
 270            }
 271        }
 272
 273        foreach (var (job, timer) in pending)
 274        {
 275            timer.Dispose();
 276            // The slot is freed whether the job is retained (it is re-published into the next
 277            // incarnation's transport, which has its own slots) or dropped.
 278            _delayedSlots.Release();
 279            if (retention is not null)
 280                continue;
 281
 282            DrainLogger?.LogWarning(
 283                "Dropping delayed in-memory worker job {Target}.{Method} due at {NotBeforeUtc} at shutdown; in-memory de
 284                job.Envelope.Call.ServiceInterfaceFullName, job.Envelope.Call.MethodName, job.Envelope.NotBeforeUtc);
 285        }
 286
 287        // Interlocked read pairs with the increment in PublishAsync: either this sees the
 288        // publisher's count (the finishing job completes the writer) or the publisher's write
 289        // lands before completion. Only a publish initiated after the transport is already idle
 290        // and draining can observe a completed channel.
 291        if (Interlocked.CompareExchange(ref _outstanding, 0, 0) == 0)
 292            _queue.Writer.TryComplete();
 293    }
 294
 295    /// <summary>
 296    /// Called by the worker host after a dequeued job finished (successfully or not). A job counts
 297    /// as outstanding from publish until here, so follow-up publishes made while it runs always
 298    /// find the writer open during the drain.
 299    /// </summary>
 300    internal void OnJobFinished()
 301    {
 302        if (Interlocked.Decrement(ref _outstanding) == 0 && _draining)
 303            _queue.Writer.TryComplete();
 304    }
 305
 306    /// <inheritdoc/>
 307    public async Task PublishAsync(WorkerJobEnvelope job, CancellationToken cancellationToken = default)
 308    {
 309        ArgumentNullException.ThrowIfNull(job);
 310        job = MaterializeFromWire(job);
 311
 312        using var activity = AsyncResponseDiagnostics.StartActivity(
 313            "asyncresponse.worker.publish",
 314            ActivityKind.Producer,
 315            job.CorrelationId);
 316        activity?.SetTag("asyncresponse.transport", "inmemory");
 317        AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget);
 318        AsyncResponseDiagnostics.SetWorker(activity, job.Call);
 319
 320        Interlocked.Increment(ref _outstanding);
 321        try
 322        {
 323            var queued = new QueuedJob(job, ExecutionContext.Capture());
 324
 325            // A publish made from INSIDE a running job must never wait for queue capacity. The
 326            // workers are the only consumers, so a worker that parks in WriteAsync against a full
 327            // queue is waiting on itself: with the default WorkerCount = 1 one in-job publish past
 328            // capacity wedges the transport permanently — no job ever completes, _outstanding never
 329            // reaches zero, the shutdown drain never completes the writer, and the whole backlog is
 330            // lost at process exit. Durable flows publish from inside jobs routinely (child start,
 331            // parent wake-up), so this is reachable in ordinary use, not an exotic case.
 332            //
 333            // Bypassing the bound for these is the safe side of the trade: the capacity exists as
 334            // backpressure on EXTERNAL producers, and follow-up work is a continuation of work the
 335            // queue already admitted. The bypass is itself bounded (InJobOverflowCapacity): past
 336            // it the publish is REJECTED, never parked — the publishing job fails and rides the
 337            // in-process redelivery ladder, which is the only backpressure a worker can be given
 338            // without waiting on itself. The catch below undoes this publish's outstanding count.
 339            if (InJobScope.IsActive)
 340            {
 341                if (_queue.Writer.TryWrite(queued) || TryEnqueueOverflow(queued))
 342                    return;
 343
 344                AsyncResponseDiagnostics.RecordInMemoryOverflowRejection();
 345                throw new InvalidOperationException(
 346                    $"The in-memory worker transport rejected a follow-up job ({job.Call.ServiceInterfaceFullName}.{job.
 347                    $"the queue is full ({nameof(InMemoryWorkerTransportOptions)}.{nameof(InMemoryWorkerTransportOptions
 348                    $"({nameof(InMemoryWorkerTransportOptions)}.{nameof(InMemoryWorkerTransportOptions.InJobOverflowCapa
 349                    "(a worker waiting on itself would deadlock), so the publishing job fails and is redelivered — make 
 350            }
 351
 352            await _queue.Writer.WriteAsync(queued, cancellationToken).ConfigureAwait(false);
 353        }
 354        catch (Exception ex)
 355        {
 356            // OnJobFinished, not a bare decrement: if this failed publish is the last thing the
 357            // drain was waiting on (the drain observed the incremented count and declined to
 358            // complete the writer), a bare decrement leaves an empty, never-completed channel —
 359            // the workers park in ReadAllAsync forever and shutdown stalls to the host's budget.
 360            OnJobFinished();
 361            AsyncResponseDiagnostics.SetError(activity, ex);
 362            throw;
 363        }
 364    }
 365
 366    // -----------------------------------------------------------------------------------------
 367    // IDelayedWorkerTransport
 368
 369    /// <inheritdoc/>
 370    /// <remarks>The in-process timer wheel has no per-hop cap; delays are bounded only by the BCL timer ceiling.</remar
 371    public TimeSpan MaxPublishDelay => TimeSpan.FromMilliseconds(uint.MaxValue - 1);
 372
 373    /// <inheritdoc/>
 374    public Task PublishAsync(WorkerJobEnvelope job, TimeSpan delay, CancellationToken cancellationToken = default)
 375    {
 376        ArgumentNullException.ThrowIfNull(job);
 377        if (delay <= TimeSpan.Zero)
 378            return PublishAsync(job, cancellationToken);
 379        if (delay > MaxPublishDelay)
 380            throw new ArgumentOutOfRangeException(nameof(delay), delay, $"Delay must be at most {MaxPublishDelay.TotalDa
 381
 382        cancellationToken.ThrowIfCancellationRequested();
 383        job = MaterializeFromWire(job);
 384
 385        // Capacity is reserved BEFORE the job is accepted, and held until the fired job has
 386        // entered the queue (WriteFiredAsync) or the drain dropped it — the delayed set and the
 387        // fired-but-pending writes together never exceed DelayedJobCapacity. A publish from
 388        // inside a running job never waits (see _delayedSlots): rejected, like an immediate
 389        // follow-up at the overflow bound, so the publishing job fails and is redelivered.
 390        if (InJobScope.IsActive)
 391        {
 392            if (!_delayedSlots.Wait(0))
 393            {
 394                AsyncResponseDiagnostics.RecordInMemoryDelayedRejection();
 395                throw new InvalidOperationException(
 396                    $"The in-memory worker transport rejected a delayed job ({job.Call.ServiceInterfaceFullName}.{job.Ca
 397                    $"{nameof(InMemoryWorkerTransportOptions)}.{nameof(InMemoryWorkerTransportOptions.DelayedJobCapacity
 398                    "Follow-up publishes never wait for room (a worker waiting on itself would deadlock), so the publish
 399                    "make its publishes idempotent, or raise the capacity.");
 400            }
 401
 402            ScheduleDelayed(job, delay);
 403            return Task.CompletedTask;
 404        }
 405
 406        return PublishDelayedFromOutsideAsync(job, delay, cancellationToken);
 407    }
 408
 409    private async Task PublishDelayedFromOutsideAsync(WorkerJobEnvelope job, TimeSpan delay, CancellationToken cancellat
 410    {
 411        // Backpressure on an external producer, exactly as the bounded queue is for its immediate
 412        // publishes: the wait ends when a scheduled job fires and enters the queue, a drain drops
 413        // the scheduled set, or the caller's token cancels.
 414        await _delayedSlots.WaitAsync(cancellationToken).ConfigureAwait(false);
 415        ScheduleDelayed(job, delay);
 416    }
 417
 418    /// <summary>Arms the timer for a job whose slot is already reserved; releases the slot when the job cannot be armed
 419    private void ScheduleDelayed(WorkerJobEnvelope job, TimeSpan delay)
 420    {
 421        using var activity = AsyncResponseDiagnostics.StartActivity(
 422            "asyncresponse.worker.publish",
 423            ActivityKind.Producer,
 424            job.CorrelationId);
 425        activity?.SetTag("asyncresponse.transport", "inmemory");
 426        activity?.SetTag("asyncresponse.worker.delay_seconds", delay.TotalSeconds);
 427        AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget);
 428        AsyncResponseDiagnostics.SetWorker(activity, job.Call);
 429
 430        var delayed = new DelayedJob(this, new QueuedJob(job, ExecutionContext.Capture()));
 431        lock (_delayedGate)
 432        {
 433            if (_draining)
 434            {
 435                // Nothing is armed here either way, so the reserved slot goes back.
 436                _delayedSlots.Release();
 437
 438                // Harness restart: a flow suspending mid-drain parks its wake-up with "the
 439                // broker" instead of faulting the draining job (and stalling the stop on the
 440                // redelivery backoff).
 441                if (_drainRetention is { } retained)
 442                {
 443                    retained.Add(job);
 444                    return;
 445                }
 446
 447                // Same contract as the shutdown drain below: delayed in-memory jobs share the
 448                // process lifetime, and a publish racing shutdown is dropped loudly, not queued
 449                // onto a channel that will complete underneath it.
 450                DrainLogger?.LogWarning(
 451                    "Rejecting delayed in-memory worker job {Target}.{Method} published during shutdown; in-memory delay
 452                    job.Call.ServiceInterfaceFullName, job.Call.MethodName);
 453                throw new InvalidOperationException("The in-memory worker transport is shutting down and no longer accep
 454            }
 455
 456            // The timer is created inside the gate so a concurrent drain either sees it in the map
 457            // (and disposes it) or the publish observed _draining above. One-shot; Fire removes it.
 458            ITimer timer;
 459            try
 460            {
 461                timer = _timeProvider.CreateTimer(static state => ((DelayedJob)state!).Fire(), delayed, delay, Timeout.I
 462            }
 463            catch
 464            {
 465                // Not armed, so nothing will ever fire — or drain — this job: hand the reserved
 466                // slot back before the publish fails. Without this every failed arming (a time
 467                // provider already disposed by a finished test fixture, a provider that rejects
 468                // the delay) burned one of DelayedJobCapacity for the life of the transport, and
 469                // once they were gone every delayed publish was rejected or blocked forever.
 470                _delayedSlots.Release();
 471                throw;
 472            }
 473
 474            _delayedJobs.Add(delayed, timer);
 475        }
 476    }
 477
 478    private void FireDelayed(DelayedJob delayed)
 479    {
 480        ITimer? timer;
 481        lock (_delayedGate)
 482        {
 483            if (!_delayedJobs.Remove(delayed, out timer))
 484                return; // The shutdown drain already claimed (and dropped) it.
 485
 486            // Count as outstanding INSIDE the gate, atomically with the removal: incremented
 487            // after the lock released, a drain snapshotting in that window saw neither the
 488            // timer-map entry nor the count — it neither retained nor waited for this job and
 489            // completed the writer underneath the write below.
 490            Interlocked.Increment(ref _outstanding);
 491        }
 492
 493        timer.Dispose();
 494        _ = WriteFiredAsync(delayed.Queued);
 495    }
 496
 497    private async Task WriteFiredAsync(QueuedJob queued)
 498    {
 499        try
 500        {
 501            await _queue.Writer.WriteAsync(queued).ConfigureAwait(false);
 502        }
 503        catch (ChannelClosedException)
 504        {
 505            // OnJobFinished, not a bare decrement, on both failure paths: if this count is the
 506            // last one a drain is waiting on, only the drain-aware decrement completes the writer
 507            // (TryComplete on an already-completed channel is a no-op here).
 508            OnJobFinished();
 509            DrainLogger?.LogWarning(
 510                "Dropping delayed in-memory worker job {Target}.{Method}: its due time fired after the transport complet
 511                queued.Job.Call.ServiceInterfaceFullName, queued.Job.Call.MethodName);
 512        }
 513        catch (Exception ex)
 514        {
 515            OnJobFinished();
 516            DrainLogger?.LogError(ex,
 517                "Failed to enqueue fired delayed in-memory worker job {Target}.{Method}.",
 518                queued.Job.Call.ServiceInterfaceFullName, queued.Job.Call.MethodName);
 519        }
 520        finally
 521        {
 522            // Held from acceptance through the pending write: the job is now either in the
 523            // bounded queue (counted there) or dropped.
 524            _delayedSlots.Release();
 525        }
 526    }
 527
 528    // Wire parity for EVERY job, in-process included: the envelope the worker receives is
 529    // re-materialized from the publisher's wire JSON — the same representation a broker delivery
 530    // carries, [JsonIgnore] argument state excluded, post-publish mutations invisible, and a
 531    // non-serializable argument failing HERE, at the publish, exactly where every broker transport
 532    // fails it. Handing the caller's live envelope through (the old path) let tests and single-node
 533    // deployments run on state no broker-backed transport can deliver. The publish serializes with
 534    // the transports' wire options and re-binds with the broker ingress's case-insensitive options.
 535    // The enqueuer's captured ExecutionContext still flows: ambient AsyncLocal state is this
 536    // transport's documented in-process feature, not envelope state.
 537    private static WorkerJobEnvelope MaterializeFromWire(WorkerJobEnvelope job)
 538        => AsyncResponseJson.DeserializeCaseInsensitive<WorkerJobEnvelope>(AsyncResponseJson.SerializeToUtf8Bytes(job))!
 539
 540    /// <summary>Identity handle for one scheduled delayed job (reference equality keys the timer map).</summary>
 541    private sealed class DelayedJob(InMemoryWorkerTransport owner, QueuedJob queued)
 542    {
 543        public QueuedJob Queued { get; } = queued;
 544        public WorkerJobEnvelope Envelope => Queued.Job;
 545
 546        public void Fire() => owner.FireDelayed(this);
 547    }
 548
 549    /// <summary>A queued job paired with the ambient execution context captured when it was enqueued.</summary>
 550    internal readonly record struct QueuedJob(WorkerJobEnvelope Job, ExecutionContext? Context);
 551}
 552
 553/// <summary>Capacity and concurrency options for the process-local worker transport.</summary>
 554public sealed class InMemoryWorkerTransportOptions
 555{
 556    /// <summary>
 557    /// Maximum queued jobs before publishers asynchronously wait. Default: 1024. Delayed jobs
 558    /// are bounded separately by <see cref="DelayedJobCapacity"/>.
 559    /// </summary>
 3896560    public int QueueCapacity { get; set; } = 1024;
 561
 562    /// <summary>Number of jobs that may execute concurrently. Default: 1.</summary>
 3962563    public int WorkerCount { get; set; } = 1;
 564
 565    /// <summary>
 566    /// Maximum number of follow-up jobs — publishes made from <em>inside</em> a running job, such
 567    /// as a durable flow starting a child or a child waking its parent — held beyond
 568    /// <see cref="QueueCapacity"/>. Follow-up publishes never wait for queue room (the workers are
 569    /// the only consumers, so a worker waiting for capacity would be waiting on itself; with the
 570    /// default <see cref="WorkerCount"/> of 1, forever) and spill into this overflow instead. Past
 571    /// it a follow-up publish throws <see cref="InvalidOperationException"/>: the publishing job
 572    /// fails and is redelivered by the in-process retry ladder, so make in-job publishes
 573    /// idempotent. Sized so an ordinary fan-out never hits it while a runaway one is bounded —
 574    /// every held job retains its materialized envelope and captured execution context. The
 575    /// current depth is the <c>asyncresponse.worker.inmemory_overflow_depth</c> gauge; rejections
 576    /// count on <c>asyncresponse.worker.inmemory_overflow_rejections</c>. Default: 4096.
 577    /// </summary>
 10562578    public int InJobOverflowCapacity { get; set; } = 4096;
 579
 580    /// <summary>
 581    /// Maximum number of delayed jobs — <c>EnqueueWorkerAsync(..., delay)</c> and the wake-ups
 582    /// behind suspended durable-flow timers — the transport holds at once: waiting on their due
 583    /// time, or fired and waiting for queue room. Neither <see cref="QueueCapacity"/> nor
 584    /// <see cref="InJobOverflowCapacity"/> covers them, and every held job retains its
 585    /// materialized envelope and captured execution context. At the bound a delayed publish from
 586    /// outside a job waits (honoring its cancellation token) until a scheduled job enters the
 587    /// queue; one made from <em>inside</em> a running job — a flow parking on a timer — never
 588    /// waits (a worker waiting for room only a fired timer draining through that worker can free
 589    /// would be waiting on itself) and throws <see cref="InvalidOperationException"/> instead:
 590    /// the publishing job fails and is redelivered by the in-process retry ladder, so make in-job
 591    /// publishes idempotent. The current count is the
 592    /// <c>asyncresponse.worker.inmemory_delayed_jobs</c> gauge; in-job rejections count on
 593    /// <c>asyncresponse.worker.inmemory_delayed_rejections</c>. Size it above the number of
 594    /// flows you expect to be sleeping at once on this transport. Default: 4096.
 595    /// </summary>
 4751596    public int DelayedJobCapacity { get; set; } = 4096;
 597
 598    /// <summary>
 599    /// Maximum number of delivery attempts before a failing job is dropped, with an error log and
 600    /// a <c>dropped</c> outcome on the worker-jobs counter. The process-local queue has no broker
 601    /// to redeliver, so retries run in-process with backoff and occupy the worker slot while they
 602    /// run — the same head-of-line trade the Kafka transport documents. This is what honors the
 603    /// durable-flow redelivery contract on this transport: a transiently failing wake-up (a lease
 604    /// held a beat too long, a revision conflict's designed "abandon and let the delivery retry")
 605    /// gets its retry instead of silently stranding the flow. <c>0</c> means unlimited retries —
 606    /// the job retries until it succeeds or the process exits, which also means a permanently
 607    /// failing job holds its worker slot (and the shutdown drain) indefinitely. Default: <c>5</c>.
 608    /// </summary>
 2570609    public int MaxDeliveryAttempts { get; set; } = 5;
 610
 611    /// <summary>Initial delay between in-process retry attempts (doubles per attempt). Default: <c>100ms</c>.</summary>
 3632612    public TimeSpan RetryBaseDelay { get; set; } = TimeSpan.FromMilliseconds(100);
 613
 614    /// <summary>Maximum delay between in-process retry attempts. Default: <c>5s</c>.</summary>
 3570615    public TimeSpan RetryMaxDelay { get; set; } = TimeSpan.FromSeconds(5);
 616
 617    internal void Validate()
 618    {
 1160619        if (QueueCapacity <= 0)
 2620            throw new InvalidOperationException($"{nameof(QueueCapacity)} must be positive.");
 1158621        if (WorkerCount <= 0)
 2622            throw new InvalidOperationException($"{nameof(WorkerCount)} must be positive.");
 1156623        if (InJobOverflowCapacity < 0)
 2624            throw new InvalidOperationException($"{nameof(InJobOverflowCapacity)} must be zero (no overflow: a follow-up
 1154625        if (DelayedJobCapacity <= 0)
 2626            throw new InvalidOperationException($"{nameof(DelayedJobCapacity)} must be positive: durable-flow timers on 
 1152627        if (MaxDeliveryAttempts < 0)
 2628            throw new InvalidOperationException($"{nameof(MaxDeliveryAttempts)} must be zero (unlimited) or positive.");
 1150629        if (RetryBaseDelay <= TimeSpan.Zero)
 2630            throw new InvalidOperationException($"{nameof(RetryBaseDelay)} must be positive.");
 1148631        if (RetryMaxDelay < RetryBaseDelay)
 2632            throw new InvalidOperationException($"{nameof(RetryMaxDelay)} must be at least {nameof(RetryBaseDelay)}.");
 633        // Bounded by the BCL timer ceiling: a value Task.Delay rejects would fail at the FIRST
 634        // retry, get swallowed by the worker loop's backstop, and drop the job without its
 635        // configured attempts or the terminal `dropped` outcome — the exact silent loss the
 636        // redelivery loop exists to prevent.
 1146637        if (RetryMaxDelay > AsyncResponseChannelOptions.MaxTimerBackedTimeout)
 4638            throw new InvalidOperationException(
 4639                $"{nameof(RetryMaxDelay)} must be at most {AsyncResponseChannelOptions.MaxTimerBackedTimeout.TotalDays:0
 1142640    }
 641}
 642
 643/// <summary>
 644/// Background consumer for <see cref="InMemoryWorkerTransport"/>: drains the queue and executes
 645/// each job via <see cref="WorkerJobExecutor"/>, under the enqueuer's captured
 646/// <see cref="ExecutionContext"/> so ambient context flows in-process. Failures are logged and
 647/// never break the loop.
 648/// <para>
 649/// Deliberately a plain <see cref="IHostedService"/>, not a <see cref="BackgroundService"/>:
 650/// since Microsoft.Extensions.Hosting.Abstractions 10.0.10, <c>BackgroundService.StartAsync</c>
 651/// queues <c>ExecuteAsync</c> to the thread pool and DISCARDS the queued work when the stopping
 652/// token fires before the pool runs it. The shutdown-drain hook is installed inside the
 653/// execution loop, so a fast start→stop under thread-pool pressure never installed it — pending
 654/// delayed jobs were neither dropped loudly nor retained (the test harness's simulated restart
 655/// lost retained wake-ups exactly this way), and accepted queued jobs sat unread forever. The
 656/// worker loops and the drain hook are part of this host's STARTED contract, so they come up
 657/// synchronously inside <see cref="StartAsync"/>, before it returns.
 658/// </para>
 659/// </summary>
 660internal sealed class InMemoryWorkerHost(
 661    InMemoryWorkerTransport _transport,
 662    WorkerJobExecutor _executor,
 663    ILogger<InMemoryWorkerHost> _logger,
 664    TimeProvider? _timeProvider = null) : IHostedService, IDisposable
 665{
 666    private CancellationTokenSource? _stopping;
 667    private Task? _execution;
 668
 669    /// <summary>Starts the worker loops and installs the shutdown-drain hook, synchronously.</summary>
 670    public Task StartAsync(CancellationToken cancellationToken)
 671    {
 672        _stopping = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 673        _execution = RunAsync(_stopping.Token);
 674        return _execution.IsCompleted ? _execution : Task.CompletedTask;
 675    }
 676
 677    /// <summary>
 678    /// Signals the drain (synchronously, via the stop registration) and waits for the workers to
 679    /// finish what was accepted, bounded by <paramref name="cancellationToken"/> — the same
 680    /// contract <c>BackgroundService.StopAsync</c> has.
 681    /// </summary>
 682    public async Task StopAsync(CancellationToken cancellationToken)
 683    {
 684        if (_execution is null)
 685            return;
 686
 687        try
 688        {
 689            _stopping!.Cancel();
 690        }
 691        finally
 692        {
 693            var cutoff = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
 694            await using var registration = cancellationToken.Register(
 695                static state => ((TaskCompletionSource)state!).TrySetResult(), cutoff);
 696            await Task.WhenAny(_execution, cutoff.Task).ConfigureAwait(false);
 697        }
 698    }
 699
 700    /// <summary>Parity with <c>BackgroundService.Dispose</c>: cancel, never dispose the source — a
 701    /// still-draining worker may hold its token.</summary>
 702    public void Dispose() => _stopping?.Cancel();
 703
 704    private async Task RunAsync(CancellationToken stoppingToken)
 705    {
 706        _transport.DrainLogger = _logger;
 707
 708        // Shutdown quiesces instead of cancelling the readers: accepted jobs were promised
 709        // in-process execution, so the workers drain the queue — including follow-up work those
 710        // jobs enqueue while draining — and the writer completes only once the transport is idle.
 711        // The drain is bounded because the queue is bounded and each job's follow-ups are finite.
 712        using var stopRegistration = stoppingToken.Register(static state =>
 713            ((InMemoryWorkerTransport)state!).BeginShutdownDrain(), _transport);
 714
 715        try
 716        {
 717            var workers = new Task[_transport.Options.WorkerCount];
 718            for (var index = 0; index < workers.Length; index++)
 719                workers[index] = RunWorkerAsync(stoppingToken);
 720            await Task.WhenAll(workers).ConfigureAwait(false);
 721        }
 722        catch (OperationCanceledException)
 723        {
 724            // Host shutdown.
 725        }
 726    }
 727
 728    private async Task RunWorkerAsync(CancellationToken stoppingToken)
 729    {
 730        // Deliberately no cancellation token on the read: the loop ends when the completed queue
 731        // is empty, never by abandoning accepted jobs mid-queue.
 732        await foreach (var queued in _transport.Reader.ReadAllAsync().ConfigureAwait(false))
 733        {
 734            await RunJobAsync(queued, stoppingToken).ConfigureAwait(false);
 735
 736            // Follow-up work first: whatever the pump could not place in the queue is run here,
 737            // BEFORE the next read — the read is what hands the freed slot to a parked external
 738            // producer, and it would keep doing so for as long as one is waiting (see
 739            // TryTakeOverflow). Each follow-up is an outstanding job in its own right and is
 740            // finished through the same path as a queued one, pump included — so whenever the
 741            // queue does have room, idle workers share a large fan-out instead of this worker
 742            // running all of it serially.
 743            while (_transport.TryTakeOverflow(out var followUp))
 744                await RunJobAsync(followUp, stoppingToken).ConfigureAwait(false);
 745        }
 746    }
 747
 748    private async Task RunJobAsync(InMemoryWorkerTransport.QueuedJob queued, CancellationToken stoppingToken)
 749    {
 750        if (stoppingToken.IsCancellationRequested && _logger.IsEnabled(LogLevel.Debug))
 751            _logger.LogDebug("Draining in-memory worker job {Target}.{Method} during shutdown.", queued.Job.Call.Service
 752
 753        try
 754        {
 755            await ExecuteWithRedeliveryAsync(queued, stoppingToken).ConfigureAwait(false);
 756        }
 757        catch (Exception ex)
 758        {
 759            // Backstop only — the redelivery loop already contains job failures. Nothing may
 760            // break this loop: it is the transport's delivery guarantee for everything still
 761            // queued behind the current job.
 762            _logger.LogError(ex, "In-memory worker job {Target}.{Method} failed.", queued.Job.Call.ServiceInterfaceFullN
 763        }
 764        finally
 765        {
 766            // Before OnJobFinished: this job just freed a slot, and the overflow entries are
 767            // still counted in the outstanding total that gates the drain's writer completion.
 768            _transport.PumpOverflow();
 769            _transport.OnJobFinished();
 770        }
 771    }
 772
 773    /// <summary>
 774    /// The transport's stand-in for broker redelivery: a failing job is retried in place with
 775    /// exponential backoff up to <see cref="InMemoryWorkerTransportOptions.MaxDeliveryAttempts"/>
 776    /// (0 = unlimited) — durable-flow wake-ups ride this queue and rely on redelivery for crash
 777    /// and contention recovery, so dropping a job on its first failure (the old behavior) could
 778    /// strand a flow that a broker-backed transport would have recovered. Retries deliberately
 779    /// run during the shutdown drain too: accepted jobs were promised in-process execution, and
 780    /// the retry budget is bounded when attempts are. The backoff SLEEP is not: a stop request
 781    /// during it drops the failing job (loudly) so the jobs queued behind it still drain.
 782    /// </summary>
 783    private async Task ExecuteWithRedeliveryAsync(InMemoryWorkerTransport.QueuedJob queued, CancellationToken stoppingTo
 784    {
 785        var options = _transport.Options;
 786        for (var attempt = 1; ; attempt++)
 787        {
 788            try
 789            {
 790                await RunAsync(queued).ConfigureAwait(false);
 791                return;
 792            }
 793            catch (Exception ex)
 794            {
 795                if (options.MaxDeliveryAttempts > 0 && attempt >= options.MaxDeliveryAttempts)
 796                {
 797                    // No broker, no dead-letter queue: dropping is the terminal outcome, so it is
 798                    // loud — an error log plus a distinct `dropped` outcome on the worker-jobs
 799                    // counter (broker transports dead-letter here instead).
 800                    _logger.LogError(ex,
 801                        "In-memory worker job {Target}.{Method} failed after {Attempts} attempts; dropping it. A durable
 802                        queued.Job.Call.ServiceInterfaceFullName, queued.Job.Call.MethodName, attempt);
 803                    AsyncResponseDiagnostics.RecordWorkerOutcome("dropped");
 804                    return;
 805                }
 806
 807                var delay = RetryDelay(options, attempt);
 808                if (stoppingToken.IsCancellationRequested && options.MaxDeliveryAttempts > 0)
 809                {
 810                    // The shutdown drain IS this token's cancellation (BeginShutdownDrain is its
 811                    // registration), so every drain-time failure lands here with the token already
 812                    // set — and binding the sleep below to it made MaxDeliveryAttempts effectively
 813                    // 1 for the whole drain: the delay threw before it began, and one failure
 814                    // dropped the job with its remaining attempts unspent, stranding the durable
 815                    // flow behind it exactly when lease contention peaks. Accepted jobs were
 816                    // promised in-process execution, so the bounded ladder still runs; each backoff
 817                    // is capped at the base delay so a failing job cannot park the worker behind
 818                    // the jobs queued after it.
 819                    var drainDelay = delay < options.RetryBaseDelay ? delay : options.RetryBaseDelay;
 820                    _logger.LogWarning(ex,
 821                        "In-memory worker job {Target}.{Method} failed on attempt {Attempt} during the shutdown drain; r
 822                        queued.Job.Call.ServiceInterfaceFullName, queued.Job.Call.MethodName, attempt, drainDelay);
 823                    await Task.Delay(drainDelay, _timeProvider ?? TimeProvider.System).ConfigureAwait(false);
 824                    continue;
 825                }
 826
 827                _logger.LogWarning(ex,
 828                    "In-memory worker job {Target}.{Method} failed on attempt {Attempt}; retrying in {Delay}.",
 829                    queued.Job.Call.ServiceInterfaceFullName, queued.Job.Call.MethodName, attempt, delay);
 830                try
 831                {
 832                    await Task.Delay(delay, _timeProvider ?? TimeProvider.System, stoppingToken).ConfigureAwait(false);
 833                }
 834                catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 835                {
 836                    // Without the token this sleep parked the (single, by default) worker for up to
 837                    // RetryMaxDelay per attempt through the whole shutdown drain, and every job
 838                    // queued behind the failing one was lost when the bounded stop returned.
 839                    _logger.LogError(ex,
 840                        "In-memory worker job {Target}.{Method} failed on attempt {Attempt} and host shutdown interrupte
 841                        queued.Job.Call.ServiceInterfaceFullName, queued.Job.Call.MethodName, attempt);
 842                    AsyncResponseDiagnostics.RecordWorkerOutcome("dropped");
 843                    return;
 844                }
 845            }
 846        }
 847    }
 848
 849    private static TimeSpan RetryDelay(InMemoryWorkerTransportOptions options, int attempt)
 850    {
 851        // Exponential backoff from the base delay, saturating at the max. Computed in ticks with
 852        // a pre-shift comparison so neither pathological attempt counts (unlimited retries) nor a
 853        // pathological base delay can overflow before the saturation check.
 854        var exponent = Math.Min(attempt - 1, 20);
 855        var baseTicks = options.RetryBaseDelay.Ticks;
 856        var maxTicks = options.RetryMaxDelay.Ticks;
 857        return baseTicks > maxTicks >> exponent
 858            ? options.RetryMaxDelay
 859            : TimeSpan.FromTicks(baseTicks << exponent);
 860    }
 861
 862    private Task RunAsync(InMemoryWorkerTransport.QueuedJob queued)
 863    {
 864        // No captured context (flow suppressed): execute directly.
 865        if (queued.Context is null)
 866        {
 867            InMemoryWorkerTransport.InJobScope.MarkActive();
 868            return _executor.ExecuteAsync(queued.Job);
 869        }
 870
 871        // Run under the enqueue-time ExecutionContext so the job inherits its ambient AsyncLocals.
 872        // The in-job marker is raised INSIDE that Run: the restore replaces ambient AsyncLocal
 873        // state with the enqueue-time snapshot, so a marker raised around this call would be wiped
 874        // before the handler ever saw it — and a publish the handler makes would take the
 875        // capacity-waiting path the marker exists to avoid.
 876        Task? task = null;
 877        ExecutionContext.Run(
 878            queued.Context,
 879            _ =>
 880            {
 881                InMemoryWorkerTransport.InJobScope.MarkActive();
 882                task = _executor.ExecuteAsync(queued.Job);
 883            },
 884            null);
 885        return task!;
 886    }
 887
 888}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

get_QueueCapacity()
get_WorkerCount()
get_InJobOverflowCapacity()
get_DelayedJobCapacity()
get_MaxDeliveryAttempts()
get_RetryBaseDelay()
get_RetryMaxDelay()
Validate()