Repository navigation
[Mocha] Cancel message handlers and release interrupted messages on shutdown - #10504
Merged
Merged
Conversation
tobias-tengler
marked this pull request as draft
October 7, 2026 20:00
tobias-tengler
force-pushed
the
mocha/shutdown-handler-cancellation
branch
2 times, most recently
from
October 7, 2026 20:06
171c463 to
2fa5296
Compare
tobias-tengler
force-pushed
the
mocha/shutdown-minimal
branch
from
October 7, 2026 20:06
750e258 to
6167eff
Compare
Contributor
Patch coverage85.1% of changed lines covered (240/282) Uncovered changed lines (JSON){
"sha": "6063f29b49b7fcc28de6e120be3387cce4dfab91",
"files": [
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Middlewares/Receive/RabbitMQAcknowledgementMiddleware.cs", "ranges": [[38, 38]] },
{ "path": "src/Mocha/src/Mocha/Runtime/MessagingRuntime.cs", "ranges": [[202, 208], [228, 231]] },
{ "path": "src/Mocha/src/Mocha/Transport/MessagingTransport.cs", "ranges": [[251, 254], [272, 274]] },
{ "path": "src/Mocha/src/Mocha.Transport.Postgres/PostgresReceiveEndpoint.cs", "ranges": [[209, 212], [240, 243], [261, 263]] },
{ "path": "src/Mocha/src/Mocha/Consumers/Implementations/BatchConsumer.cs", "ranges": [[91, 93], [240, 241], [244, 248], [250, 250]] },
{ "path": "src/Mocha/src/Mocha/Consumers/Batching/BatchCollector.cs", "ranges": [[85, 85]] }
]
}Project coverage: 58.7% (315704/538143 lines) |
tobias-tengler
force-pushed
the
mocha/shutdown-handler-cancellation
branch
from
October 8, 2026 07:05
2fa5296 to
b84e810
Compare
tobias-tengler
force-pushed
the
mocha/shutdown-minimal
branch
2 times, most recently
from
October 8, 2026 07:08
8bc09cf to
7bb10f2
Compare
tobias-tengler
force-pushed
the
mocha/shutdown-handler-cancellation
branch
3 times, most recently
from
October 8, 2026 08:39
f616eee to
57f2418
Compare
tobias-tengler
changed the title
tobias-tengler
force-pushed
the
mocha/shutdown-handler-cancellation
branch
2 times, most recently
from
October 8, 2026 13:30
7a16145 to
85ffdaf
Compare
tobias-tengler
force-pushed
the
mocha/shutdown-minimal
branch
from
October 8, 2026 13:30
aae6435 to
13d6f71
Compare
tobias-tengler
force-pushed
the
mocha/shutdown-handler-cancellation
branch
2 times, most recently
from
October 12, 2026 14:48
29d485e to
7a28857
Compare
tobias-tengler
marked this pull request as ready for review
October 12, 2026 16:49
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
PostgreSQL lease release retains last_delivered, preventing the promised immediate redelivery.
1 open finding
What changed in this PR
This PR improves Mocha shutdown behavior by canceling in-flight handlers, safely settling interrupted messages, and making lifecycle cleanup resilient to individual failures.
Changes:
- Propagates shutdown cancellation through individual and batch handlers.
- Releases or requeues interrupted transport messages.
- Continues stopping and disposing remaining resources after failures.
| File | Description |
|---|---|
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ShutdownTests.cs |
Tests cancellation and requeue on shutdown. |
src/Mocha/test/Mocha.Transport.Postgres.Tests/Behaviors/ShutdownTests.cs |
Tests PostgreSQL shutdown settlement. |
src/Mocha/test/Mocha.Transport.AzureServiceBus.Tests/ReceiveEndpointLifecycleUnitTests.cs |
Tests endpoint failures and cancellation. |
src/Mocha/test/Mocha.Transport.AzureServiceBus.Tests/Helpers/FakeServiceBusClient.cs |
Adds message-processing test hooks. |
src/Mocha/test/Mocha.Transport.AzureServiceBus.Tests/AzureServiceBusAcknowledgementMiddlewareTests.cs |
Verifies uncancelled settlement tokens. |
src/Mocha/test/Mocha.Tests/Runtime/MessagingRuntimeTests.cs |
Tests resilient runtime disposal. |
src/Mocha/test/Mocha.Tests/Runtime/MessagingRuntimeShutdownTests.cs |
Tests handler and batch cancellation. |
src/Mocha/test/Mocha.Tests/Middlewares/Consume/Retry/RetryExecutorTests.cs |
Tests cancellation during retries. |
src/Mocha/test/Mocha.Tests/Consumers/Batching/BatchCollectorTests.cs |
Tests buffered-entry removal. |
src/Mocha/src/Mocha/Transport/MessagingTransport.cs |
Continues stopping endpoints after failures. |
src/Mocha/src/Mocha/Runtime/MessagingRuntime.cs |
Makes stop and disposal failure-resilient. |
src/Mocha/src/Mocha/Middlewares/ReceiveContext.cs |
Retains the receive cancellation token. |
src/Mocha/src/Mocha/Middlewares/Receive/Redelivery/ReceiveRedeliveryMiddleware.cs |
Skips redelivery handling during cancellation. |
src/Mocha/src/Mocha/Middlewares/Receive/ReceiveFaultMiddleware.cs |
Skips fault generation during cancellation. |
src/Mocha/src/Mocha/Middlewares/Receive/ReceiveDeadLetterMiddleware.cs |
Suppresses cancellation error logging. |
src/Mocha/src/Mocha/Middlewares/Consume/Retry/RetryExecutor.cs |
Stops retry processing when cancelled. |
src/Mocha/src/Mocha/Endpoints/ReceiveEndpoint.cs |
Propagates cancellation without critical logging. |
src/Mocha/src/Mocha/Consumers/Implementations/BatchConsumer.cs |
Isolates batching per receive endpoint. |
src/Mocha/src/Mocha/Consumers/Batching/BatchOptions.cs |
Documents per-endpoint concurrency. |
src/Mocha/src/Mocha/Consumers/Batching/BatchCollector.cs |
Supports removing buffered entries. |
src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQReceiveEndpoint.cs |
Handles shutdown cancellation. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Middlewares/Receive/RabbitMQAcknowledgementMiddleware.cs |
Settles messages despite cancellation. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQConsumerManager.cs |
Cancels handlers before disconnecting. |
src/Mocha/src/Mocha.Transport.Postgres/PostgresReceiveEndpoint.cs |
Deletes or releases interrupted messages. |
src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresMessageStore.cs |
Adds bulk lease release support. |
src/Mocha/src/Mocha.Transport.InMemory/InMemoryReceiveEndpoint.cs |
Handles expected shutdown cancellation. |
src/Mocha/src/Mocha.Transport.AzureServiceBus/Middlewares/Receive/AzureServiceBusAcknowledgementMiddleware.cs |
Settles messages independently of cancellation. |
src/Mocha/src/Mocha.Transport.AzureServiceBus/AzureServiceBusReceiveEndpoint.cs |
Handles cancelled processor callbacks. |
🧠 Review effort: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
PascalSenn
reviewed
Oct 12, 2026
… failures Delete or release handled Postgres messages with a token that stop does not cancel, and release the rest of a batch when polling is cancelled, so in-flight messages are not redelivered after the lock timeout. Keep stopping the remaining transports and endpoints when one fails, then rethrow the first failure.
The shutdown tests asserted that stopping produced no warning or error logs through a LogRecorder test helper. They now assert only the handler cancellation and the resulting database state, and the helper is removed.
The acknowledgement middlewares complete or ack with CancellationToken.None and the Postgres endpoint leaves an interrupted message to the batch release. The reasons were only in the review threads, so each site now states the invariant: once the handler returns its work is committed and settlement runs to completion, while a cancelled handler throws and is abandoned, nacked or released instead.
The receive endpoint swallowed the exception of a handler cancelled by the stopping processor. A message cancelled before the acknowledgement middleware ran was never settled, so it stayed locked until the lock expired and gained a delivery count. The SDK abandons an unsettled message only when the handler throws. The handler now lets the exception bubble. The processor error handler returns without logging when the reported exception is a cancellation of the processor's own token. Messages the pipeline already abandoned are unaffected because the SDK skips its abandon for settled messages.
…very ReleaseLeasedMessagesAsync reset the delivery count of an interrupted message but kept last_delivered, so the read query's backoff still skipped the row for up to 2^delivery_count seconds after a stop. The release now clears last_delivered as well, so the row looks like one that was never delivered.
PascalSenn
approved these changes
Oct 13, 2026
This was referenced Oct 25, 2026
Merged
This was referenced Oct 26, 2026
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.
When a receive endpoint stopped, in-flight handlers were never cancelled because
ReceiveContextdiscarded the transport's cancellation token, so they kept running after shutdown began. Stopping also left PostgreSQL messages behind. A handler that finished after the stop began could not delete its message, because the delete used the already cancelled token, and messages that were leased but not finished stayed locked until the 5 minute lease timeout. Both were then redelivered and processed a second time. Finally, a transport or endpoint that failed to stop kept the remaining ones from stopping.This PR passes the endpoint's cancellation token to handlers, including batch handlers, and keeps shutdown cancellation from triggering retries, fault handling or error logs. RabbitMQ and Azure Service Bus settle messages with
CancellationToken.None, so an interrupted message is still nacked or abandoned. Batches are collected and processed per receive endpoint, so stopping one endpoint only cancels its own batches, andMaxConcurrentBatchesnow applies per endpoint.The PostgreSQL endpoint deletes or releases a handled message with
CancellationToken.None. A delete that fails after the handler completed is logged and the message is released with the error recorded. When polling is cancelled, the endpoint releases the messages of the current batch that it still holds and undoes their delivery count bump, so they are redelivered right away instead of after the lease timeout, and the interrupted attempt does not count toward max delivery.MessagingRuntime.StopAsyncandMessagingTransport.StopAsynckeep stopping the remaining transports and endpoints when one fails, then rethrow the first failure. A transport that failed to stop stays started, so disposing the runtime tries it again. Disposal runs every remaining step when one fails and rethrows the first failure once everything is released, except a stop failure that an earlier stop already reported.