Scale out error ingestion for EF (1/3): ingestion owns its dispatcher - #5800
Merged
Conversation
johnsimons
force-pushed
the
john/scale_error
branch
2 times, most recently
from
August 20, 2026 02:57
2e12056 to
9357098
Compare
Refactors ErrorIngestor to receive the dispatcher as a parameter. This allows ErrorIngestion to maintain a reference to the dispatcher during teardown, ensuring that in-flight batches can still be forwarded without encountering null references if the infrastructure is shut down. Includes a new acceptance test for error forwarding.
Refactors AuditIngestor and AuditPersister to receive the dispatcher as a parameter. Similar to recent changes in ErrorIngestor, this allows AuditIngestion to maintain a reference to the dispatcher during teardown, ensuring that in-flight batches can still be processed and forwarded before the transport infrastructure is fully shut down.
johnsimons
force-pushed
the
john/scale_error
branch
from
August 20, 2026 04:15
9357098 to
3a88398
Compare
rbev
approved these changes
Aug 20, 2026
johnsimons
added a commit
that referenced
this pull request
Aug 24, 2026
ErrorIngestion and AuditIngestion had drifted into the same class. Same bounded channel sized to MaxConcurrency, same single reader drain loop, same signalling of every context's TaskCompletionSource on the two failure paths, same comments. Only the metrics differed. Hosting audit ingestion in the primary is about to add a third copy of it. IngestionPipeline in ServiceControl.Infrastructure, alongside Watchdog, now owns that machinery: receivers Enqueue, a single assembler builds batches, and a configurable number of writers write them. Each ingestion keeps everything that is genuinely its own, which is all of the transport setup, the watchdog, the fault policy and the shutdown ordering PR #5800 established. What each hands the pipeline is a delegate, so their metrics stay exactly where they were rather than becoming an abstraction in a shared project. Configured here to what the loop does today: one writer, batch size MaxConcurrency, and no batch timeout, meaning whatever has arrived is written immediately. The settings that make the other values reachable come next. The pipeline is built for more than one writer from the start, because that is what the machinery is for and what the tests cover. What it is not yet given is a way to be configured with more. Three fixes that fall out of having one implementation to fix: - startStopSemaphore was never disposed, in either ingestion - Task.Run with an already cancelled token throws before the loop body runs, so the loops take CancellationToken.None and handle cancellation themselves - cancellation dropped in-flight and pending messages without answering them. The assembler abandons what it never dispatched and whatever is left in the message channel, Run abandons batches no writer picked up, and a writer fails its own batch. Those receives are then redelivered rather than left waiting. MessageContextExtensions moves into the pipeline's project, since it is how the pipeline answers a receive. The two copies were identical apart from the ContextBag key, which is per message, so one key serves the whole process.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Part 1 of 3 introducing scale-out of error ingestion for EF persistence, where several
processes drain the error queue into one shared database.
This part contains no scale-out feature. It is groundwork plus a bug fix, and stands on
its own.
Ingestion takes its dispatcher as an argument
ErrorIngestorandAuditIngestorheld aLazy<IMessageDispatcher>. That is aconstructor dependency which cannot be satisfied at construction time: the dispatcher is
registered by
AddNServiceBusEndpointbehind a factory that dereferences the transportseam, so resolving it before the endpoint's hosted service starts throws a bare
NullReferenceException. TheLazywrapper, and the comment apologising for registrationorder next to it, were the tell.
It is not a constructor dependency, it is a call-time collaborator. The ingestors do not
own a transport: they take message contexts, persist them, announce them and forward them.
The dispatcher belongs to whoever received the batch.
So
IngestandVerifyCanReachForwardingAddressnow take anIMessageDispatcher, andErrorIngestion/AuditIngestionpass the one from theTransportInfrastructuretheyalready create.
ImportFailedErrorsandImportFailedAuditstake theLazythat theingestors used to hold, so their behaviour is unchanged.
Three things fall out:
ForwardpassinganyContext.TransportTransactionnow visiblybelongs with the dispatcher it is handed to rather than one from a different
infrastructure, the start-order hazard leaves the ingestion path, and the ingestors become
testable with a fake dispatcher and no container.
AuditPersisterthreads the dispatcher through as well. It is hand constructed insideAuditIngestor's constructor, so leaving it holding aLazywould have the file arguingwith itself. Its
IMessageSessionis untouched: saga audit commands genuinely need theendpoint.
Shutdown ordering, which was duplicating forwarded messages
StopAsynctore the transport infrastructure down before draining the channel. Worse,EnsureStoppedstops the receiver withnew CancellationToken(canceled: true)("stopreceivers ASAP"), which fires the registration in
OnMessagethat cancels every in-flightTaskCompletionSource. Those messages were abandoned and redelivered on the next start,after the drain had already forwarded them. With
ForwardErrorMessageson, everygraceful shutdown with a non-empty channel put duplicates in the error log queue.
The order is now: stop receiving under the shutdown token, complete the writer, drain, then
tear down. Stopping under the real token means
StopReceivewaits for in-flight messageswhile the loop is still running and the infrastructure is still up, so they ingest, forward
once, and their receives actually commit.
This is what the (previously dead)
finallyblock inStopAsyncwas always shaped for.Watchdogis untouched; the split is local to each ingestion class.