[iancooper]

Symptom

OutboxCompactionDispatchedOnlyTests.When_compacting_only_dispatched_messages_in_outbox fails intermittently in CI with "Oldest dispatched message should be removed by compaction". The failed test takes exactly 2s, which is the length of its polling window, so compaction did not happen within that window.

Root causes

There are two independent causes, and each can produce this symptom on its own.

1. Product defect: InMemoryBox silently drops cleanup work

• Expiry and compaction ran as fire-and-forget tasks. Each called Monitor.TryEnter on one shared lock and did nothing if the lock was held.
• EnforceCapacityLimit records the compaction attempt time before the task runs. So a dropped compaction still counts as an attempt, and the cooldown blocks any retry for a full ExpirationScanInterval. A dropped expiry scan is lost the same way, because _lastScanAt has already been updated.
• In production both use the same interval, so they tend to fire on the same Add. With EntryLimit or more entries, the O(n) expiry scan holds the lock long enough for compaction to lose the race repeatedly. EntryLimit then goes unenforced.
• This hits the outbox hardest, because expiry never removes undispatched messages. InMemoryInbox shares the same defect through the common base class.
• [Feat] InMemoryBox Expiry and Outbox limits #4116 added the cooldown. Before that, a dropped compaction was retried on the next Add.

2. Test-suite defect: two tests starve the thread pool

• ConcurrentStartGuardTests puts 20 Task.Run work items on a Barrier. The scheduler's same-id concurrency test puts 100.
• None of those work items can proceed until the pool has grown that large. On a CI runner, that starved every other pool work item in the process for 19–27s, so background cleanup ran long after the tests' waits had expired.
• It showed up on this PR's first CI run: the failure fell entirely inside a 19s run of ConcurrentStartGuardTests.
• ConcurrentStartGuardTests was added in June (af0501b35), after [Feat] InMemoryBox Expiry and Outbox limits #4116. That fits the issue's remark that the test "may have become more unreliable due to a change".

Fix

• InMemoryBox.cs: expiry and compaction are now recorded as requests and handled by at most one background worker per box.
  • The worker loops until no requests remain. It only ever stops under the same lock that requests are recorded under, so no request can be missed or stranded.
  • Each operation in a pass runs in its own try/catch, so a failed expiry does not skip a compaction taken in the same pass. The first failure is rethrown once the worker has stopped, and still surfaces as an unobserved task exception, as before.
  • Compaction re-checks the limit when it runs. It skips if expiry has already brought the box under EntryLimit, and it never trims below the target size.
  • The worker runs on TaskScheduler.Default and contains no await or blocking wait. It never runs on a Reactor/Proactor pump's single-threaded SynchronizationContext, and at most one pool thread per box is used however heavy the load.
  • Nothing else changed: no default values, cooldown logic or interval logic.
• How the design evolved:
  • 85d3325 used a blocking lock. Under load that holds a pool thread per waiting request, and those can pile up.
  • 10c28b1 introduced the single worker.
  • cd80f65 hardened the failure path, following the review below.
• Tests (e72f0fa): the two Barrier-based test classes now run in a non-parallel ThreadPoolSaturating collection, so they no longer overlap other tests.

Tests

The new tests hold the cleanup lock or worker deterministically through Blocking* and FailingSecondExpiry* test doubles. Each one was measured against every version of InMemoryBox.cs on this branch (net9.0). The figures are EntryCount, except for the coalescing test, which counts expiry scans.

Test	master 4b25517	blocking lock 85d3325	single worker 10c28b1	final
Outbox/When_compaction_contends_with_expiry_…_should_still_compact	❌ 6	✅	✅	✅ 3
Inbox/When_inbox_compaction_contends_with_expiry_…_should_still_compact	❌ 6	✅	✅	✅ 3
Outbox/When_expiry_contends_with_compaction_…_should_still_expire	❌ 3	✅	✅	✅ 1
Outbox/When_expiry_is_requested_repeatedly_during_a_cleanup_should_coalesce_into_one_scan	❌ 6 †	❌ 6	✅	✅ 2
Outbox/When_expiry_brings_the_outbox_under_its_limit_before_a_waiting_compaction_should_not_compact	✅ ‡	❌ 1	✅	✅ 3
Outbox/When_expiry_throws_in_the_same_cleanup_pass_as_a_compaction_should_still_compact	❌ 6	✅	❌ 6	✅ 3
Outbox/When_expiry_leaves_the_outbox_over_its_limit_before_a_waiting_compaction_should_compact_only_to_target §	❌ 8	❌ 1	✅	✅ 2

† On master this value depends on timing. Scans that reach the lock while it is held are dropped, but this test releases the lock straight after its reads, so they usually find it free and run. The blocking lock gives 6 every time.

‡ This test does not fail against master, because master drops the waiting compaction altogether. It pins a fault that the blocking lock introduced (over-trimming from an out-of-date count), not the original bug.

§ A characterisation test, added after review, for the Math.Min(entriesRequested, count - newSize) cap in CompactIfStillOverLimit. The cap has been in place since 10c28b1, so the test passed as soon as it was written. It was observed failing (Expected: 2, Actual: 1) under the mutation Compact(entriesRequested), and the mutation was then reverted. On master the compaction is dropped (8); on the blocking lock, compaction over-trims (1).

These tests can still pass by mistake if the thread pool is extremely slow, but a slow pool can no longer make them fail.

Suite results:

• Paramore.Brighter.InMemory.Tests: 160/160 on net9.0 and net10.0.
• Paramore.Brighter.Core.Tests (net9.0, after merging master): 1498 passed, 7 skipped, 0 failed.
• All 18 compaction, expiry and cache-size tests, run 10 times with DOTNET_PROCESSOR_COUNT=2: green every run.

The full diagnosis is in bugfixes/0038-inmemory-outbox-compaction-flaky/bugfix.md.

Fixes #4475

🤖 Generated with Claude Code

[iancooper]

@claude please review this PR, focusing only on the change it makes (not the wider codebase):

1. Single-worker cleanup in src/Paramore.Brighter/InMemoryBox.cs:
   • Check the handoff between EnsureCleanupWorker, RunCleanupWorker and TryTakeCleanupRequests. Is there any interleaving where a request made while the worker is stopping is missed, or where two workers run at once?
   • Check the exception path. If RemoveExpiredMessages or Compact throws, is the running flag reset correctly? Are any pending requests left in a sensible state?
   • Check CompactIfStillOverLimit. Is re-checking against EntryLimit, and capping at count - newSize, correct for both InMemoryOutbox (only dispatched entries are eligible) and InMemoryInbox?
   • Could any of this run on, or block, a Reactor/Proactor pump's single-threaded SynchronizationContext? The worker is started on TaskScheduler.Default and has no awaits.
   • _lastScanAt and _lastCompactionAttemptAt are read and written without synchronisation. That predates this PR, but does the new design make it worse?
2. Tests in tests/Paramore.Brighter.InMemory.Tests:
   • Review the five new *LockContention*, *Coalescing* and *CurrentCount* tests and their Blocking* test doubles. Are they deterministic in the failing direction? Do they test behaviour rather than implementation details?
   • Review ThreadPoolSaturatingCollection. Is moving the two Barrier-based test classes into a non-parallel collection a sound fix for thread-pool starvation? In particular, does taking the scheduler test out of the implicit CommandProcess collection lose any isolation it needed?

Background and the full diagnosis are in the PR description and in bugfixes/0038-inmemory-outbox-compaction-flaky/bugfix.md.

[claude]

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

---

Review of PR #4482: InMemoryBox single-worker cleanup and thread-pool starvation

• Read the diff and bugfix.md
• Review the single-worker cleanup: handoff, exception path, compaction re-check, SynchronizationContext, timestamps
• Review the new tests and the Blocking* doubles
• Review ThreadPoolSaturatingCollection
• Post review

Verdict: the design is sound. The handoff has no lost-wakeup race and cannot run two workers at once. I found no correctness bugs in the normal path. There are two low-severity points on the exception path, one gap in test coverage, and one inaccuracy in the PR description's "before fix" column. I did not build the code or run the tests; this review comes from reading the code.

---

1. src/Paramore.Brighter/InMemoryBox.cs

Handoff between EnsureCleanupWorker, RunCleanupWorker and TryTakeCleanupRequests: ✅ correct

The invariant holds. _cleanupWorkerRunning only goes from true to false under _cleanupStateLock, and only when both request slots are empty. A caller only starts a worker after moving the flag from false to true under the same lock. I worked through every order of events between a caller (which records the request under the lock, then calls EnsureCleanupWorker, which takes the lock again) and a worker calling TryTakeCleanupRequests:

Interleaving	Outcome
Request recorded → worker takes it → caller sees running == true and returns	Request is handled by the current pass ✅
Worker finds nothing and clears the flag → request recorded → caller sees false and starts a worker	New worker handles it ✅
Request recorded → worker takes it, finishes and clears the flag → caller then sees false and starts a worker	That worker finds nothing and exits. This costs one wasted pool dispatch but loses no work ✅

• Two workers at once: not possible. After TryTakeCleanupRequests returns false, the old worker only leaves RunCleanupWorker and touches nothing else. So a new worker started at that point never overlaps real work.
• Callers never wait on a scan. The lock is never held during RemoveExpiredMessages or Compact; it only covers a few field writes.
• Pass order is right. Expiry runs before compaction in each pass. InMemoryOutbox.Add calls ClearExpiredMessages() and then EnforceCapacityLimit() (InMemoryOutbox.cs:142-143), so when both requests arrive together, expiry gets the first chance to make compaction unnecessary.

Exception path: ✅ the flag is reset, with two small gaps (low severity)

The catch resets _cleanupWorkerRunning and rethrows. The task is fire-and-forget, so the exception only surfaces through UnobservedTaskException, which is the same as before this PR.

1. A request can be stranded. Suppose a request is recorded after the worker threw but before the catch takes the lock. That caller sees running == true and returns. The catch then clears the flag, and nothing picks the request up until the next ClearExpiredMessages or EnforceCapacityLimit call gets past its interval check. So the request is delayed by up to one ExpirationScanInterval, not lost for good. That is no worse than master, but it breaks the "no request is lost" claim in the comment at InMemoryBox.cs:157-160.
2. Compaction from the same pass is dropped. If RemoveExpiredMessages throws, any compactionRequested taken in the same pass has already been cleared from the slot and never runs. The cooldown then blocks a retry for another interval.

In practice both built-in implementations only call ConcurrentDictionary.TryRemove inside LINQ, so they are close to never throwing. That is why I rate this low. A small hardening option:

catch
{
    bool pending;
    lock (_cleanupStateLock)
    {
        _cleanupWorkerRunning = false;
        pending = _expiryRequestedAt.HasValue || _compactionRequested.HasValue;
    }
    if (pending) EnsureCleanupWorker();   // bounded by the request rate, not a hot loop
    throw;
}

Another option is to wrap each operation in its own try/catch so that a failed expiry does not skip the compaction in the same pass. Either change is optional. If you skip both, consider softening the comment to "no request is lost unless a pass throws".

Fix this →

CompactIfStillOverLimit: ✅ correct for both boxes

• Re-check against EntryLimit. It checks count < upperSize against the total EntryCount, the same measure and threshold (>=) that EnforceCapacityLimit uses to trigger. So compaction is skipped exactly when a new trigger would not fire.
• Outbox: only dispatched entries are eligible. count includes undispatched entries, and Compact takes at most N dispatched entries, so the cap is an upper bound and never a promise. If the box is full of undispatched messages it stays over the limit and compaction removes whatever dispatched entries exist. That is the same as before the PR, and the right outcome: undispatched messages must not be dropped.
• Inbox: every entry is eligible, so count - newSize is the exact number needed to reach the target. ✅
• Math.Min(entriesRequested, count - newSize). If the box grew after the request was made, the smaller, older request wins. The box then ends slightly above the target, which is conservative and matches the original sizing. If the box shrank, the current value wins and the box never goes below the target. Both are fine. You could arguably drop entriesRequested and always use the current count - newSize, but that is a matter of taste.
• Old edge case, unchanged: if CompactionPercentage > 1, count - newSize can be negative, and Take(negative) simply removes nothing. This PR does not make it worse.

Reactor/Proactor single-threaded SynchronizationContext: ✅ no risk

• The worker runs on TaskScheduler.Default and never on TaskScheduler.Current or the context. With DenyChildAttach, no awaits and no blocking waits, it cannot capture or post back to the pump's context.
• On the caller's thread, the only extra work is two very short lock (_cleanupStateLock) sections around field writes. The lock is never held during a scan, so a pump thread can only wait a few instructions on another caller or on the worker's take-and-clear. This is an improvement on 85d3325's blocking lock and has the same cost as master's TryEnter.
• One point worth knowing: under sustained load with a short ExpirationScanInterval and a large box, the worker can loop and keep one pool thread busy for as long as requests keep arriving. That limit of one thread per box is what the design intends, and much better than a pile-up of threads.

_lastScanAt and _lastCompactionAttemptAt without synchronisation: ✅ not made worse, arguably better

• The worker never reads or writes either field; only callers do, as on master.
• The two races that already existed are now less harmful:
  • Two callers can both pass the interval check. Before, that queued two scans. Now the two requests merge into one pass.
  • DateTimeOffset is 16 bytes, so a torn read is possible. At worst it causes one extra request, which is merged, or one skipped request, which is retried on the next call after the interval.
• Moving both fields under _cleanupStateLock would be cheap. It is outside this PR's scope, though, so I would not do it here.

---

2. Tests (tests/Paramore.Brighter.InMemory.Tests)

The five new tests and the Blocking* doubles

The doubles are well built:

• volatile _armed means the fill-up phase never blocks.
• The ManualResetEventSlim gate stays set once released, so later passes run straight through.
• The 10s timeout on Release*.Wait means a broken test cannot hang the run.
• They override the protected abstract hooks, which is the natural seam, and do not touch private state.

Do they fail reliably when the bug is present?

Test	Against master (TryEnter + drop)	After fix
Compaction contends with expiry (outbox and inbox)	Fails, provided the compaction task reaches TryEnter within the 500ms delay. If the pool is slower than that, the test can pass by mistake. The PR description already says this.	Deterministic. The blocked worker takes the queued compaction on the thread it already has, so no second pool thread is needed once the gate is released ✅
Expiry contends with compaction	Same as above	Deterministic ✅
Coalesce into one scan	On master the dropped scans never reach the override, so the count is 1, not 6. The test still fails, because it expects 2.	Deterministic: exactly 2 ✅
Expiry brings the box under the limit	On master, compaction is dropped and the count is 3, so this test passes against master	Deterministic ✅

⚠️ The "Before fix" column in the PR description mixes up two baselines. The figures "6 scans" and "1 (over-trimmed)" come from the intermediate commit 85d3325 (blocking lock), not from master. That matters for two tests:

• ...should_coalesce_into_one_scan and ...before_a_waiting_compaction_should_not_compact guard against the intermediate design's failure modes: threads piling up and over-trimming.
• They do not guard against the original bug. ...should_not_compact is not RED against master at all.

Both are worth keeping as specifications of how the box should behave, but please relabel the column, for example "before fix (vs 85d3325)", or add a note. The same applies to the table in bugfix.md if it repeats these figures.

Behaviour or implementation?

• Four of the five tests assert on EntryCount and on which entries survive, which is observable behaviour ✅
• ...coalesce_into_one_scan asserts on ExpiryScans == 2, a count of calls to a protected hook. It is the most tightly coupled to the implementation: it would reject any future design that runs one cheap scan per request. It is still the only guard against the pool pile-up seen in 85d3325, so I would keep it. A comment on why the bound matters (one pool thread, bounded work) would help a future reader.

Coverage gap (minor). No test covers the count - newSize cap, meaning the case where expiry removes some entries but the box is still at or over the limit. Example: limit 5, request made at 10 entries (asks for 8 removals), expiry removes 3, leaving 7. The cap should remove 5 rather than 8, so the box ends at the target of 2 and not at 0. As things stand, deleting Math.Min would not turn any test red.

Fix this →

Nit: the new test files and Blocking* doubles have no MIT licence header, while ThreadPoolSaturatingCollection.cs does. .agent_instructions/documentation.md asks for headers on new files.

ThreadPoolSaturatingCollection: ✅ a sound fix, and no isolation is lost

• How it works: a collection with DisableParallelization = true runs after all parallel collections have finished, one test at a time. So the two Barrier tests can no longer starve anything that asserts on timing. Putting both classes in the same collection also stops them starving each other.
• What it does not do: it fixes the symptom (they stop overlapping other tests) rather than the cause (tests whose progress depends on the pool growing). A stronger, optional alternative is to start the Barrier participants with TaskCreationOptions.LongRunning, which gives each a dedicated thread, or with new Thread(...). Neither then depends on pool growth, and they could run in parallel again. The collection approach is simpler and clearly correct, so I would accept it as it is.
• Leaving the implicit CommandProcess collection: no isolation is lost.
  • "CommandProcess" has no CollectionDefinition in this assembly. It is only an implicit name that makes the scheduler and sweeper classes run one after another.
  • When_scheduling_message_with_existing_id_should_atomically_replace_timer builds all of its state per instance: its own InMemorySchedulerFactory, FakeTimeProvider, InternalBus, InMemoryOutbox and a Guid-suffixed RoutingKey.
  • InMemoryScheduler has no shared static state apart from its logger, and the only statics in Scheduler/ are read-only type and reflection caches.
  • It also now runs with nothing else in parallel at all, which is stricter isolation than before.

---

Summary

Area	Result
Handoff / no lost wakeup / at most one worker	✅ Correct
Exception path	✅ Flag is reset. ⚠️ Low severity: a request can be stranded for up to one interval, and compaction in the same pass is dropped when expiry throws
CompactIfStillOverLimit	✅ Correct for outbox and inbox. ➖ The count - newSize cap has no test
SynchronizationContext / pump safety	✅ No risk
Unsynchronised timestamps	✅ Not worse; the races that already existed are less harmful
New tests	✅ Deterministic once fixed. ⚠️ The PR's "before fix" column mixes master with 85d3325, and one test is not RED against master
ThreadPoolSaturatingCollection	✅ Sound, and no isolation lost

None of these points blocks the merge. I made no code changes.