[guilhermeappel]

Description

maxOutStandingMessages is never enforced when the producer's outbox implements only IAmAnOutboxAsync. OutboxProducerMediator.OutstandingMessagesCheck counted outstanding messages only through the sync outbox, so with an async-only outbox the count stayed at 0 and PostAsync never threw OutboxLimitReachedException.

It now counts through the async outbox when there is no sync one. OutstandingMessagesCheck is only called from the Task.Run in CheckOutstandingMessages, so it blocks there with GetAwaiter().GetResult() on a thread-pool thread with no SynchronizationContext.

There are two commits, because CONTRIBUTING asks to keep structural and behavioural changes apart. The first merges the two sync calls, which differed only in maxCount, into one call (no behaviour change). The second adds the async branch and the tests.

Every outbox in this repo implements both interfaces, so none of them is affected. This matters for custom or third-party outboxes that are async-only. For those, a configured limit starts being enforced, so PostAsync can now throw OutboxLimitReachedException where it never did before.

I found this while looking at #4489. It is independent of it.

Related Issues

Related to #4489 (found while investigating; does not fix it).

Type of Change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Documentation update

Checklist

• I have read the Contributing Guide
• I have checked the documentation for relevant guidance
• I have added/updated XML documentation for any public API changes (no public API change)
• I have added/updated tests as appropriate
• My changes follow the existing code style and conventions

Additional Notes

• Tests: OutstandingLimitAnyOutboxTests posts until the limit is hit, once with an async-only outbox and once with an outbox that has both interfaces. I ran them on a clean checkout of master with only the new test files added: the async-only test fails on Assert.True(reachedLimit), the other passes. With this change both pass. The Core suite on net10.0 passes (1627 passed, 7 skipped).
• Question: blocking on the async count is the smallest change. If you would rather make the check itself async, I can do that.
• CLA: the link in CONTRIBUTING (clahub.com/agreements/iancooper/Paramore) returns "page not found" for me. How would you like me to sign it?
• Please merge with a merge commit rather than squash, as CONTRIBUTING asks.

[iancooper]

@claude Can we check this one, in particular this comment from the author: "Question: blocking on the async count is the smallest change. If you would rather make the check itself async, I can do that." This might be dangerous given that we have a single-threaded pump with our own SynchronizationContext and Scheduler. Blocking on the count might hurt us. Can you trace that?

[claude]

Claude finished @iancooper's task in 1m 2s —— View job

---

Tracing the blocking async count against the pump's SynchronizationContext

• Read PR diff
• Trace CheckOutstandingMessages / OutstandingMessagesCheck call sites
• Trace pump BrighterAsyncContext / BrighterSynchronizationContext / BrighterTaskScheduler interaction with Task.Run
• Post findings

Short answer

This can't deadlock the pump. The .GetAwaiter().GetResult() never runs on the pump thread, and nothing on the pump thread waits for it. The author's comment on line 1360 is correct. There are still two smaller costs (see below), and an async version would be a reasonable follow-up, but neither should block this PR.

The trace

1. Call sites. OutstandingMessagesCheck is private and has one caller, CheckOutstandingMessages (OutboxProducerMediator.cs:797). That is reached from ClearOutbox, ClearOutboxAsync, the background clear and both resilience-pipeline catch blocks (lines 413, 482, 772, 1409, 1441). It always queues the work and never waits for it:

   Task.Run(() => OutstandingMessagesCheck(requestContext));   // the task is never awaited or waited on

   So the pump thread just queues the work and moves on. It never waits on the count.

2. Task.Run skips the pump's scheduler. The Proactor runs its loop inside BrighterAsyncContext.Run(() => EventLoop()) (Proactor.cs:99). Task.Run always uses TaskScheduler.Default, whatever TaskScheduler.Current is. On top of that, BrighterAsyncContext builds its factory with HideScheduler (BrighterAsyncContext.cs:80-84), so TaskScheduler.Current is already Default inside pump work. The delegate runs on a thread-pool thread, not in BrighterTaskQueue.

3. The SynchronizationContext doesn't follow it. ExecutionContext flows into Task.Run, but SynchronizationContext does not. This holds on .NET Core and on netfx (Task captures with IgnoreSyncCtx). On the pool thread, SynchronizationContext.Current is null, so any await inside a custom GetOutstandingMessageCountAsync resumes on the pool. It never posts back to BrighterSynchronizationContext, whatever ConfigureAwait value the outbox uses.

4. No circular wait. Sync-over-async deadlocks when the blocked thread is the one that has to run the continuation. Here the blocked thread is a pool thread, and the continuations also run on the pool. The pump thread waits on neither. AsyncLocals (e.g. Activity.Current) do flow, but that only affects tracing.

This matches the existing design: the sync-outbox branch already does blocking I/O on the same pool thread for the same reason (//This is expensive, so use a background thread).

Remaining costs (minor)

• Sync-over-async ties up pool threads. While the count runs, one pool thread sits blocked, and the I/O continuation needs a second pool thread to finish. Under pool starvation this is the classic amplifier. It is bounded, because s_checkOutstandingSemaphoreToken lets only one count run at a time, process-wide. The sync branch already blocks one pool thread for the same work, so the extra cost is about one more thread for the continuation.
• Already true before this PR, but the async branch makes it more visible:
  • s_checkOutstandingSemaphoreToken is static, so all mediators in the process share it.
  • _lastOutStandingMessageCheckAt is only updated after a queued task gets the semaphore (line 1338). A burst of posts before the first check runs can therefore queue several Task.Runs, and each one blocks a pool thread on s_checkOutstandingSemaphoreToken.Wait().
  • No cancellation token or timeout is passed to the count. A hung outbox call holds the static semaphore indefinitely, which stops outstanding-count checks for every mediator. The sync branch has the same problem.

Should the check be async?

Not needed for correctness. If you want it, it's a small change and fixes the pool-thread costs above at the same time:

// CheckOutstandingMessages
_ = Task.Run(() => OutstandingMessagesCheckAsync(requestContext));

private async Task OutstandingMessagesCheckAsync(RequestContext? requestContext)
{
    await s_checkOutstandingSemaphoreToken.WaitAsync().ConfigureAwait(false);
    try
    {
        ...
        if (_outBox != null) { /* sync call as now */ return; }
        if (_asyncOutbox != null)
        {
            _outStandingCount = await _asyncOutbox
                .GetOutstandingMessageCountAsync(_maxOutStandingCheckInterval, requestContext, maxCount, args: _outBoxBag)
                .ConfigureAwait(false);
            return;
        }
        ...
    }
    finally { s_checkOutstandingSemaphoreToken.Release(); }
}

This still runs off the pump: Task.Run puts it on the pool, and ConfigureAwait(false) doesn't matter there because there's no SynchronizationContext. It also lets semaphore waiters wait without holding threads. It's a structural change, so under Tidy First it would be its own commit.

Recommendation

• The blocking approach in this PR is safe for both the Proactor and the Reactor. There's no path back onto the pump's BrighterSynchronizationContext or BrighterTaskScheduler.
• Taking the async variant above is fine, either in this PR as a separate structural commit or as a follow-up. It's a small improvement in thread efficiency, not a fix for a deadlock.
• One suggestion: extend the comment on line 1360 to say why there's no SynchronizationContext. Task.Run uses TaskScheduler.Default, and SyncContext doesn't flow through ExecutionContext. Without that, a later refactor that inlines the call or awaits the Task.Run (which would put it back on the pump) could break this without anyone noticing.

I only read the code for this. I did not run the tests.

[CLAassistant]

All committers have signed the CLA.

[iancooper]

@guilhermeappel

This is worth fixing:

One suggestion: extend the comment on line 1360 to say why there's no SynchronizationContext. Task.Run uses TaskScheduler.Default, and SyncContext doesn't flow through ExecutionContext. Without that, a later refactor that inlines the call or awaits the Task.Run (which would put it back on the pump) could break this without anyone noticing.

This one might be worth raising as its own issue, it may bite at scale

s_checkOutstandingSemaphoreToken is static, so all mediators in the process share it.
_lastOutStandingMessageCheckAt is only updated after a queued task gets the semaphore (line 1338). A burst of posts before the first check runs can therefore queue several Task.Runs, and each one blocks a pool thread on s_checkOutstandingSemaphoreToken.Wait().
No cancellation token or timeout is passed to the count. A hung outbox call holds the static semaphore indefinitely, which stops outstanding-count checks for every mediator. The sync branch has the same problem.

[iancooper]

A heads-up from the investigation in #4560 and #4554. OutstandingMessagesCheck already runs inside Task.Run and blocks there on a static semaphore (s_checkOutstandingSemaphoreToken.Wait()). With the DI default interval of zero, it is queued after every Post.

Adding GetAwaiter().GetResult() on the async outbox path means each check also holds a pool thread for the whole async count. For a database-backed async-only outbox, that is a network round trip. In #4560 we measured that blocked pool threads delay TimedOutboxSweeper ticks 3× or more, and that the per-Post check cuts Post throughput by about 30×.

Two possible ways to avoid adding to that:

• Make the check itself async: Task.Run(async () => await OutstandingMessagesCheckAsync(...)) with WaitAsync on the semaphore.
• Or land it after the Each Post sorts the whole in-memory outbox to count outstanding messages, even with no limit set #4554 changes (skip at -1, record the time before queuing, at most one check in flight), so the blocking happens at most once per interval.

Not a blocker for the async-only bug itself. It only affects async-only outboxes, and none in the repo are.

[guilhermeappel]

Thanks, Ian. I'll push the extended comment on why that thread has no SynchronizationContext, with a small test that posts from inside the pump's context and checks the count doesn't deadlock it.

On the blocking call, I'll go with your second option: I'll bring this PR up to date once the skip at -1 from #4554 has merged, so the async count only runs when a limit is set.

I'll open an issue for the static semaphore and the missing timeout and cancellation on the count once the #4554 PRs are up, so it only covers what's left.

[iancooper]

Thanks, Ian. I'll push the extended comment on why that thread has no SynchronizationContext, with a small test that posts from inside the pump's context and checks the count doesn't deadlock it.

On the blocking call, I'll go with your second option: I'll bring this PR up to date once the skip at -1 from #4554 has merged, so the async count only runs when a limit is set.

@guilhermeappel tag me when it's done and I'll re-review

[iancooper]

Thanks @guilhermeappel