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

Information
Class: AsyncResponse.SerialExecutorRegistry
Assembly: AsyncResponse.Core
File(s): /_/src/AsyncResponse.Core/SerialExecutorRegistry.cs
Line coverage
93%
Covered lines: 140
Uncovered lines: 9
Coverable lines: 149
Total lines: 401
Line coverage: 93.9%
Branch coverage
94%
Covered branches: 66
Total branches: 70
Branch coverage: 94.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%66100%
.cctor()100%11100%
get_UtcNow()100%11100%
OnSubscriptionRegistered(...)100%22100%
OnSubscriptionRetired(...)100%44100%
EnqueueAsync()94.44%181894.28%
TryEnqueue(...)83.33%1212100%
RemoveAsync()100%141486.36%
IsTombstonedUnderLock(...)100%44100%
PruneTombstonesUnderLock()80%101088.88%
.ctor(...)100%11100%
get_Executor()100%11100%
get_Retired()100%11100%
get_EnqueuesDrained()100%11100%
get_InFlightEnqueues()100%11100%
get_Retiring()100%11100%

File(s)

/_/src/AsyncResponse.Core/SerialExecutorRegistry.cs

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2
 3namespace AsyncResponse;
 4
 5/// <summary>
 6/// Owns the per-channel <see cref="ChannelSerialExecutor"/> instances for a response channel and
 7/// coordinates their lifecycle so that, for any one channel key, work is never enqueued onto an
 8/// executor that is concurrently being retired — which would silently drop the message — and at
 9/// most one <em>live</em> executor exists per channel at a time.
 10/// <para>
 11/// The coordination is a single lock guarding the map. <see cref="EnqueueAsync"/> reserves the live
 12/// executor under that lock, then waits for bounded queue capacity outside it. Retirement closes
 13/// admission, waits for those reservations, drains the executor, and only then lets a new executor
 14/// be created. Disposal runs outside the lock.
 15/// </para>
 16/// <para>
 17/// This replaces an earlier <c>ConcurrentDictionary</c> + fire-and-forget <c>Task.Run(remove)</c>
 18/// scheme in which a new waiter reusing a correlation id mid-drain could observe a second executor
 19/// for the same channel, briefly violating the per-channel ordering guarantee.
 20/// </para>
 21/// </summary>
 193422internal sealed class SerialExecutorRegistry(
 193423    ILogger _logger,
 193424    TimeSpan? disposeDrainLimit = null,
 193425    TimeSpan? enqueueDrainLimit = null,
 193426    TimeProvider? timeProvider = null)
 27{
 28    // How long a retired channel's tombstone blocks executor re-creation. Long enough to outlive
 29    // any enqueue that was already in flight when cleanup retired the executor, short enough that
 30    // an unpruned tombstone only ever delays a reused correlation id briefly.
 1431    internal static readonly TimeSpan TombstoneLifetime = TimeSpan.FromSeconds(30);
 32
 33    // Upper bound on how long retirement waits for in-flight enqueues to drain. A producer can be
 34    // parked indefinitely awaiting queue capacity with a token that never fires, and teardown
 35    // paths await RemoveAsync directly — they must not inherit that hang.
 1436    internal static readonly TimeSpan EnqueueDrainLimit = TimeSpan.FromSeconds(30);
 37
 38    // Upper bound on how long retirement waits for the executor's dispatched work to finish. A
 39    // dispatched item runs arbitrary user code (a completion predicate that never finishes), and
 40    // an unbounded wait here would wedge the channel key permanently — every later enqueue for
 41    // the correlation id parks on the never-completed retirement, including a NEW waiter that
 42    // legitimately re-registered the id.
 1443    internal static readonly TimeSpan DisposeDrainLimit = TimeSpan.FromSeconds(30);
 44
 193445    private readonly TimeSpan _disposeDrainLimit = disposeDrainLimit ?? DisposeDrainLimit;
 46
 47    // Overridable alongside _disposeDrainLimit: RemoveAsync waits on BOTH budgets in sequence, so
 48    // a caller that could shorten only one still paid the other's full 30 seconds — and the
 49    // worst-case retirement became "the value I passed, plus 30s" rather than the value passed.
 193450    private readonly TimeSpan _enqueueDrainLimit = enqueueDrainLimit ?? EnqueueDrainLimit;
 51
 52    // Tombstone expiry runs on the injected clock so a virtual-clock harness can advance past
 53    // TombstoneLifetime instead of sleeping 30 real seconds. Without it, the drop-a-delivery
 54    // branch in EnqueueAsync (and tombstone pruning) could not be covered deterministically.
 193455    private readonly TimeProvider _timeProvider = timeProvider ?? TimeProvider.System;
 56
 355357    private DateTimeOffset UtcNow => _timeProvider.GetUtcNow();
 58
 193459    private readonly Dictionary<string, ExecutorEntry> _executors = new(StringComparer.Ordinal);
 193460    private readonly Dictionary<string, DateTimeOffset> _tombstones = new(StringComparer.Ordinal);
 61
 62    /// <summary>
 63    /// Expiry-ordered index over <see cref="_tombstones"/> (constant lifetime, so insertion order
 64    /// is expiry order). Pruning pops expired heads instead of scanning the whole dictionary —
 65    /// the scan ran per retired waiter under <see cref="_gate"/>, the same lock every dispatched
 66    /// message's enqueue takes, so its cost grew quadratically with throughput. Entries can be
 67    /// stale (the channel was re-tombstoned later, or the tombstone was cleared); the dictionary
 68    /// stays the source of truth and each popped head is validated against it.
 69    /// </summary>
 193470    private readonly Queue<(string Channel, DateTimeOffset ExpiresAtUtc)> _tombstoneOrder = new();
 193471    private readonly Dictionary<string, int> _registrations = new(StringComparer.Ordinal);
 193472    private readonly object _gate = new();
 73
 74    /// <summary>
 75    /// Records a live subscription for <paramref name="channel"/>. While any subscription is
 76    /// registered, retirement tombstones do not drop work — a retired executor is legitimately
 77    /// recreated, and the remaining subscription's own cleanup retires it again (no leak).
 78    /// </summary>
 79    public void OnSubscriptionRegistered(string channel)
 80    {
 181781        ArgumentException.ThrowIfNullOrWhiteSpace(channel);
 82
 181783        lock (_gate)
 84        {
 181785            _registrations[channel] = _registrations.TryGetValue(channel, out var count) ? count + 1 : 1;
 181786            _tombstones.Remove(channel);
 181787        }
 181788    }
 89
 90    /// <summary>Records that a subscription for <paramref name="channel"/> is gone.</summary>
 91    public void OnSubscriptionRetired(string channel)
 92    {
 173593        ArgumentException.ThrowIfNullOrWhiteSpace(channel);
 94
 173595        lock (_gate)
 96        {
 173597            if (!_registrations.TryGetValue(channel, out var count))
 498                return;
 99
 1731100            if (count <= 1)
 1717101                _registrations.Remove(channel);
 102            else
 14103                _registrations[channel] = count - 1;
 14104        }
 1735105    }
 106
 107    /// <summary>
 108    /// Asynchronously enqueues work, applying bounded per-channel backpressure. Returns <c>true</c>
 109    /// once the work is accepted by a live executor; <c>false</c> when it was suppressed by a
 110    /// tombstone (the executor was retired with no registration left — every dispatch it ever
 111    /// admitted has fully completed, so a caller draining before disposal knows nothing is in
 112    /// flight).
 113    /// </summary>
 114    public async ValueTask<bool> EnqueueAsync(string channel, Func<Task> work, CancellationToken cancellationToken = def
 115    {
 5144116        ArgumentException.ThrowIfNullOrWhiteSpace(channel);
 5144117        ArgumentNullException.ThrowIfNull(work);
 118
 0119        while (true)
 120        {
 5146121            ExecutorEntry? entry = null;
 5146122            Task? retirement = null;
 5146123            lock (_gate)
 124            {
 5146125                if (!_executors.TryGetValue(channel, out var current))
 126                {
 127                    // A tombstoned channel was retired and no subscription is registered anymore:
 128                    // recreating an executor here (typically for an enqueue that was already in
 129                    // flight when cleanup ran) would leak it — nothing retires it again — and the
 130                    // work item would no-op anyway because the subscriptions are gone. Drop it.
 131                    // With a subscription still registered the recreate is legitimate (its own
 132                    // cleanup retires the new executor), so the tombstone does not apply.
 1322133                    if (!_registrations.ContainsKey(channel) && IsTombstonedUnderLock(channel))
 134                    {
 135                        // Deliberate, but never silent: if this fires for a live waiter, its channel
 136                        // registered the subscription only after the transport began delivering.
 6137                        _logger.LogWarning(
 6138                            "Suppressed a delivery for channel {Channel}: the channel is tombstoned and has no registere
 6139                            channel);
 6140                        return false;
 141                    }
 142
 1316143                    current = new ExecutorEntry(new ChannelSerialExecutor(_logger, channel));
 1316144                    _executors[channel] = current;
 145                }
 146
 5140147                if (current.Retiring)
 2148                    retirement = current.Retired.Task;
 149                else
 150                {
 5138151                    current.InFlightEnqueues++;
 5138152                    entry = current;
 153                }
 5140154            }
 155
 5140156            if (entry is null)
 157            {
 2158                await retirement!.WaitAsync(cancellationToken).ConfigureAwait(false);
 2159                continue;
 160            }
 161
 162            bool accepted;
 163            try
 164            {
 5138165                accepted = await entry.Executor.Enqueue(work, cancellationToken).ConfigureAwait(false);
 5138166            }
 167            finally
 168            {
 5138169                TaskCompletionSource? enqueuesDrained = null;
 5138170                lock (_gate)
 171                {
 5138172                    entry.InFlightEnqueues--;
 5138173                    if (entry.Retiring && entry.InFlightEnqueues == 0)
 2174                        enqueuesDrained = entry.EnqueuesDrained;
 5138175                }
 176
 5138177                enqueuesDrained?.TrySetResult();
 178            }
 179
 5138180            if (accepted)
 5138181                return true;
 0182        }
 5144183    }
 184
 185    /// <summary>The outcome of a non-blocking <see cref="TryEnqueue"/>.</summary>
 186    public enum TryEnqueueOutcome
 187    {
 188        /// <summary>Accepted by the channel's live executor.</summary>
 189        Accepted,
 190
 191        /// <summary>
 192        /// Not accepted right now — the executor's bounded queue is full, or (for a caller that did
 193        /// not ask to tell the two apart) the channel's executor is mid-retirement. The work was not
 194        /// queued; the producer should come back later.
 195        /// </summary>
 196        Full,
 197
 198        /// <summary>Suppressed by a tombstone (retired executor, no registration left): nothing will ever run it.</summ
 199        Suppressed,
 200
 201        /// <summary>
 202        /// Not accepted right now because the channel's executor is mid-retirement — nothing is
 203        /// overloaded, and <see cref="EnqueueAsync"/> would wait the retirement out and admit the
 204        /// work onto a fresh executor. Reported only to a caller that passes
 205        /// <c>distinguishRetiring</c>; everyone else keeps reading it as <see cref="Full"/>.
 206        /// </summary>
 207        Retiring
 208    }
 209
 210    /// <summary>
 211    /// Non-blocking counterpart of <see cref="EnqueueAsync"/>: never waits for queue capacity or
 212    /// for a retirement to finish. Built for the DB channels' process-wide dispatch sweep, which
 213    /// walks every subscribed correlation id in turn: awaiting one correlation id's capacity there
 214    /// parked the whole loop — a single waiter wedged in a slow completion predicate, fed a
 215    /// backlog of NEW progress messages, stopped every other correlation id's delivery until its
 216    /// executor drained. A <see cref="TryEnqueueOutcome.Full"/> result leaves the message
 217    /// unclaimed in the store for a later rescan of that one correlation id.
 218    /// <para>
 219    /// <paramref name="distinguishRetiring"/> is for a producer whose <see cref="TryEnqueueOutcome.Full"/>
 220    /// is TERMINAL rather than "come back later". The sweep above answers a full queue and a
 221    /// retiring executor the same way — the message stays in the store — so it keeps one case. The
 222    /// Redis channel has no store to come back to: a full queue faults the wait as overloaded, and
 223    /// reading a mid-retirement executor (a previous waiter on the same correlation id still
 224    /// tearing down) as that overload faulted a fan-out sibling or a re-attached waiter as
 225    /// indeterminate with nothing overloaded at all.
 226    /// </para>
 227    /// </summary>
 228    public TryEnqueueOutcome TryEnqueue(string channel, Func<Task> work, bool distinguishRetiring = false)
 229    {
 10423230        ArgumentException.ThrowIfNullOrWhiteSpace(channel);
 10423231        ArgumentNullException.ThrowIfNull(work);
 232
 10423233        lock (_gate)
 234        {
 10423235            if (!_executors.TryGetValue(channel, out var current))
 236            {
 237                // Same tombstone rule as EnqueueAsync: no registration and a live tombstone means
 238                // the work would run against no subscription; recreating an executor would leak it.
 464239                if (!_registrations.ContainsKey(channel) && IsTombstonedUnderLock(channel))
 240                {
 13241                    _logger.LogWarning(
 13242                        "Suppressed a delivery for channel {Channel}: the channel is tombstoned and has no registered su
 13243                        channel);
 13244                    return TryEnqueueOutcome.Suppressed;
 245                }
 246
 451247                current = new ExecutorEntry(new ChannelSerialExecutor(_logger, channel));
 451248                _executors[channel] = current;
 249            }
 250
 251            // Mid-retirement: EnqueueAsync would wait for the drain and then recreate; a
 252            // non-blocking caller simply comes back after it.
 10410253            if (current.Retiring)
 1254                return distinguishRetiring ? TryEnqueueOutcome.Retiring : TryEnqueueOutcome.Full;
 255
 256            // TryWrite is synchronous and never blocks, so it can run under the gate; no in-flight
 257            // enqueue bookkeeping is needed because nothing is left waiting for capacity.
 10409258            return current.Executor.TryEnqueue(work, logIfFull: false)
 10409259                ? TryEnqueueOutcome.Accepted
 10409260                : TryEnqueueOutcome.Full;
 261        }
 10423262    }
 263
 264    /// <summary>
 265    /// Retires the channel's serial executor (if present), draining its queued work. Safe to call
 266    /// concurrently with <see cref="EnqueueAsync"/>: admitted enqueues finish against the retiring
 267    /// executor, while later enqueues wait until it is fully drained before creating a replacement.
 268    /// </summary>
 269    public async ValueTask RemoveAsync(string channel)
 270    {
 2056271        ArgumentException.ThrowIfNullOrWhiteSpace(channel);
 272
 273        ExecutorEntry? entry;
 274        Task waitForEnqueues;
 2056275        var ownsRetirement = false;
 2056276        lock (_gate)
 277        {
 2056278            if (!_executors.TryGetValue(channel, out entry))
 275279                return;
 280
 1781281            if (entry.Retiring)
 282            {
 16283                waitForEnqueues = entry.Retired.Task;
 284            }
 285            else
 286            {
 1765287                entry.Retiring = true;
 1765288                ownsRetirement = true;
 1765289                if (entry.InFlightEnqueues == 0)
 290                {
 1763291                    waitForEnqueues = Task.CompletedTask;
 292                }
 293                else
 294                {
 2295                    entry.EnqueuesDrained = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)
 2296                    waitForEnqueues = entry.EnqueuesDrained.Task;
 297                }
 298            }
 1781299        }
 300
 1781301        if (!ownsRetirement)
 302        {
 16303            await waitForEnqueues.ConfigureAwait(false);
 16304            return;
 305        }
 306
 307        try
 308        {
 309            try
 310            {
 311                // Bounded wait: an admitted enqueue can be parked indefinitely on a full queue
 312                // with a token that never fires. Proceeding after the limit is safe — disposal
 313                // completes the executor's writer, which unparks the wedged producer, and its
 314                // retry then lands on the tombstone/recreate machinery built for exactly the
 315                // enqueue-races-retirement case.
 1765316                await waitForEnqueues.WaitAsync(_enqueueDrainLimit, _timeProvider).ConfigureAwait(false);
 1765317            }
 0318            catch (TimeoutException)
 319            {
 0320                _logger.LogWarning(
 0321                    "Timed out after {DrainLimit} waiting for in-flight enqueues on channel {Channel} to drain; disposin
 0322                    _enqueueDrainLimit,
 0323                    channel);
 0324            }
 325
 326            try
 327            {
 328                // Bounded for the same reason: disposal waits for the reader loop, which can be
 329                // parked in a dispatched item's arbitrary user code. The writer is completed
 330                // before disposal first awaits, so the abandoned loop drains and exits on its own
 331                // if the wedged item ever finishes; retiring the entry regardless (the finally
 332                // below) keeps the channel key usable for future waiters.
 1765333                await entry.Executor.DisposeAsync().AsTask().WaitAsync(_disposeDrainLimit, _timeProvider).ConfigureAwait
 1757334            }
 8335            catch (TimeoutException)
 336            {
 8337                _logger.LogWarning(
 8338                    "Timed out after {DrainLimit} waiting for in-flight work on channel {Channel} to finish; abandoning 
 8339                    _disposeDrainLimit,
 8340                    channel);
 8341            }
 1765342        }
 343        finally
 344        {
 1765345            lock (_gate)
 346            {
 1765347                if (_executors.TryGetValue(channel, out var current) && ReferenceEquals(current, entry))
 1765348                    _executors.Remove(channel);
 349
 350                // Tombstone the retired channel so an enqueue that raced this retirement cannot
 351                // recreate a leaked executor; ClearTombstone lifts it the moment a new
 352                // subscription legitimately reuses the channel.
 1765353                var tombstoneExpiresAtUtc = UtcNow + TombstoneLifetime;
 1765354                _tombstones[channel] = tombstoneExpiresAtUtc;
 1765355                _tombstoneOrder.Enqueue((channel, tombstoneExpiresAtUtc));
 1765356                PruneTombstonesUnderLock();
 1765357            }
 358
 1765359            entry.Retired.TrySetResult();
 360        }
 2056361    }
 362
 363    private bool IsTombstonedUnderLock(string channel)
 364    {
 63365        if (!_tombstones.TryGetValue(channel, out var expiresAtUtc))
 40366            return false;
 367
 23368        if (expiresAtUtc > UtcNow)
 19369            return true;
 370
 4371        _tombstones.Remove(channel);
 4372        return false;
 373    }
 374
 375    private void PruneTombstonesUnderLock()
 376    {
 1765377        if (_tombstoneOrder.Count == 0)
 0378            return;
 379
 1765380        var now = UtcNow;
 1772381        while (_tombstoneOrder.TryPeek(out var head) && head.ExpiresAtUtc <= now)
 382        {
 7383            _tombstoneOrder.Dequeue();
 384
 385            // Only drop the tombstone the popped entry still describes: a re-tombstoned channel
 386            // has a later expiry in the dictionary (its own queue entry follows), and a cleared
 387            // one is already gone.
 7388            if (_tombstones.TryGetValue(head.Channel, out var current) && current <= now)
 5389                _tombstones.Remove(head.Channel);
 5390        }
 1765391    }
 392
 1767393    private sealed class ExecutorEntry(ChannelSerialExecutor executor)
 394    {
 19079395        public ChannelSerialExecutor Executor { get; } = executor;
 3550396        public TaskCompletionSource Retired { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
 6397        public TaskCompletionSource? EnqueuesDrained { get; set; }
 22319398        public int InFlightEnqueues { get; set; }
 24234399        public bool Retiring { get; set; }
 400    }
 401}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Logging.ILogger,System.Nullable`1<System.TimeSpan>,System.Nullable`1<System.TimeSpan>,System.TimeProvider)
.cctor()
get_UtcNow()
OnSubscriptionRegistered(System.String)
OnSubscriptionRetired(System.String)
EnqueueAsync()
TryEnqueue(System.String,System.Func`1<System.Threading.Tasks.Task>,System.Boolean)
RemoveAsync()
IsTombstonedUnderLock(System.String)
PruneTombstonesUnderLock()
.ctor(AsyncResponse.ChannelSerialExecutor)
get_Executor()
get_Retired()
get_EnqueuesDrained()
get_InFlightEnqueues()
get_Retiring()