Add poison message handling to the dispatchers#1366
Conversation
…ase of poison message handling, except for entity unlock requests
Co-authored-by: Chris Gillum <cgillum@microsoft.com>
Co-authored-by: Chris Gillum <cgillum@gmail.com>
…ad for trace activities
…vent for json deserialization, etc.
| this.Reason = reason; | ||
| } | ||
|
|
||
| // Private ctor for JSON deserialization (required by some storage providers and out-of-proc executors) |
There was a problem hiding this comment.
Unrelated to this PR but I bug I found when testing (JSON was not able to deserialize this event because it lacked a 0-arg constructor and the other constructors all had multiple parameters)
There was a problem hiding this comment.
Also unrelated to this PR, but I realized while working on it that this code I wrote a while back had some incorrect assumptions so I took the opportunity to fix it
There was a problem hiding this comment.
Pull request overview
This PR adds an extensibility hook (IPoisonMessageHandler) and integrates poison/invalid message detection into the core dispatchers so that corrupted or “poisoned” inputs can be handled deterministically (e.g., fail orchestration/activity/entity work) instead of always throwing.
Changes:
- Introduces
IPoisonMessageHandlerand wires it into orchestration/activity/entity dispatchers for invalid work items and poison message detection. - Adds structured logging support for poison-message detection (new event ID + event source + log event).
- Adds dispatch-count tracking on history events and propagates poison metadata through entity request processing.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| src/DurableTask.Core/Tracing/TraceHelper.cs | Adjusts entity invocation activity ending to better handle partial result sets. |
| src/DurableTask.Core/TaskOrchestrationDispatcher.cs | Adds poison detection/handling and updates reconciliation to return a drop reason. |
| src/DurableTask.Core/TaskEntityDispatcher.cs | Adds poison detection/handling for entity messages, plus poison-aware batching/result shaping. |
| src/DurableTask.Core/TaskActivityDispatcher.cs | Adds poison/invalid handling for activity scheduling messages (including failing poisoned tasks). |
| src/DurableTask.Core/Logging/StructuredEventSource.cs | Adds a new structured event for poison message detection. |
| src/DurableTask.Core/Logging/LogHelper.cs | Adds PoisonMessageDetected helper overloads emitting structured logs. |
| src/DurableTask.Core/Logging/LogEvents.cs | Adds a new structured log event type for poison messages. |
| src/DurableTask.Core/Logging/EventIds.cs | Reserves a new event ID for poison message detection. |
| src/DurableTask.Core/IPoisonMessageHandler.cs | New interface defining poison detection and handling hooks. |
| src/DurableTask.Core/History/HistoryEvent.cs | Adds DispatchCount to history events for poisoning heuristics/telemetry. |
| src/DurableTask.Core/History/ExecutionRewoundEvent.cs | Adds a parameterless ctor for JSON deserialization compatibility. |
| src/DurableTask.Core/Entities/OrchestrationEntityContext.cs | Adds AbandonAcquire() to reset lock acquisition state on failure. |
| src/DurableTask.Core/Entities/EventFormat/RequestMessage.cs | Adds poison metadata fields used during entity request processing. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
… combined' Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>
… combined' Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>
cgillum
left a comment
There was a problem hiding this comment.
Some initial comments. I haven't gone through the dispatcher code yet (those are bigger diffs).
cgillum
left a comment
There was a problem hiding this comment.
Adding a few quick comments for now.
| NewMessages = session.CurrentMessageBatch.Select(m => m.TaskMessage).ToList(), | ||
| NewMessages = session.CurrentMessageBatch.Select(m => | ||
| { | ||
| m.TaskMessage.Event.DispatchCount = (int)m.OriginalQueueMessage.DequeueCount; |
There was a problem hiding this comment.
Shouldn't we be updating this in the message dequeue path - i.e. TaskQueue.cs or ControlQueue.cs?
There was a problem hiding this comment.
We can, but this felt the more natural place for me since we are setting it at the point where it will actually be used (this value is only read by the dispatchers, and so it made sense to me to set it when we are generating the work item to be sent to the dispatchers). And this way all the changes are localized in AzureStorageOrchestrationService
That being said I don't feel strongly about this so I can move it if that's better
…ages twice, also changed the logic for invalid work items to just delete the incoming messages
| /// <summary> | ||
| /// A message or event could not be deserialized | ||
| /// </summary> | ||
| DeserializationError, |
There was a problem hiding this comment.
I added all of these for future-proofing (and consistency, to cover all our poison message cases) in case another backend needs them but really the Azure Storage implementation only needs a handful of them for its own poison-message-handling.
I suppose if we ever need all of them, we could add them at that point and then make a new release that the backend could reference. Or we can keep them to be consistent
…ariable' Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>
| if (poisonMessageReason == null && scheduledEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount) | ||
| { | ||
| // start a task to run RenewUntil | ||
| renewTask = Task.Factory.StartNew( | ||
| () => this.RenewUntil(workItem, renewCancellationTokenSource.Token), | ||
| renewCancellationTokenSource.Token); | ||
| poisonMessageReason = $"Activity has received an event with dispatch count {taskMessage.Event.DispatchCount} which exceeds the maximum dispatch " + | ||
| $"count of {this.poisonMessageHandler.MaxDispatchCount}. The task will be failed."; | ||
| } |
|
|
||
| case EventType.EventRaised: | ||
| EventRaisedEvent eventRaisedEvent = (EventRaisedEvent)e; | ||
| bool isPoisonMessage = eventRaisedEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount; |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 18 out of 18 changed files in this pull request and generated 3 comments.
Comments suppressed due to low confidence (6)
src/DurableTask.Core/TaskOrchestrationDispatcher.cs:449
this.poisonMessageHandler?.MaxDispatchCountis nullable, soevt.DispatchCount > this.poisonMessageHandler?.MaxDispatchCountbecomes a lifted comparison that returnsbool?, which cannot be used inWhere(...). This will not compile whenpoisonMessageHandleris null.
Compute a non-null max value (e.g., int.MaxValue when no handler) and compare against that.
var poisonEvents = runtimeState.NewEvents.Where(evt => evt.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount);
src/DurableTask.Core/TaskActivityDispatcher.cs:183
scheduledEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCountperforms a lifted comparison (returnsbool?) whenpoisonMessageHandleris null, which will not compile in anifcondition.
Coalesce MaxDispatchCount to a non-null value (e.g., int.MaxValue when no handler) before comparing.
if (poisonMessageReason == null && scheduledEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount)
{
poisonMessageReason = $"Activity has received an event with dispatch count {taskMessage.Event.DispatchCount} which exceeds the maximum dispatch " +
$"count of {this.poisonMessageHandler.MaxDispatchCount}. The task will be failed.";
}
src/DurableTask.Core/TaskEntityDispatcher.cs:474
request.DispatchCount > this.poisonMessageHandler?.MaxDispatchCountis a lifted comparison that returnsbool?whenpoisonMessageHandleris null, which cannot be assigned tobool.
Gate the comparison on a non-null handler (or coalesce to a non-null max).
bool isPoisonMessage = request.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount;
src/DurableTask.Core/TaskEntityDispatcher.cs:590
eventRaisedEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCountis a lifted comparison that returnsbool?whenpoisonMessageHandleris null, which cannot be assigned tobool.
Gate the comparison on a non-null handler (or coalesce to a non-null max).
bool isPoisonMessage = eventRaisedEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount;
src/DurableTask.Core/TaskEntityDispatcher.cs:1116
workToDoNow.Operations.Any(op => op.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount)uses a lifted comparison in theAnypredicate (returnsbool?when the handler is null), which will not compile.
Only evaluate the Any(...) when the handler is non-null (or compare against a coalesced max).
bool poisonMessagesExist = workToDoNow.Operations.Any(op => op.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount);
Test/DurableTask.AzureStorage.Tests/PoisonMessageHandlingTests.cs:47
- This test file is added under
Test/DurableTask.AzureStorage.Tests/, but the repo’s test projects are undertest/and the referenced AzureStorage test project istest/DurableTask.AzureStorage.Tests/DurableTask.AzureStorage.Tests.csproj. As a result, this file will not be compiled or executed by the existing test project/solution, so it won’t provide the intended coverage.
Move the file into test/DurableTask.AzureStorage.Tests/ (or add/update a Test/*.csproj + solution/project references) so the tests actually run in CI.
[TestClass]
public class PoisonMessageHandlingTests
{
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 18 out of 18 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (9)
Test/DurableTask.AzureStorage.Tests/PoisonMessageHandlingTests.cs:47
- This file is under
Test/DurableTask.AzureStorage.Tests, but the repo’s active test project istest/DurableTask.AzureStorage.Tests(referenced byDurableTask.sln). There is no.csprojunderTest/, so these tests will not be compiled or executed in CI as-is.
Move this test to test/DurableTask.AzureStorage.Tests (or add an actual Test/DurableTask.AzureStorage.Tests/*.csproj and wire it into the solution/pipelines).
[TestClass]
public class PoisonMessageHandlingTests
{
src/DurableTask.Core/TaskOrchestrationDispatcher.cs:449
evt.DispatchCount > this.poisonMessageHandler?.MaxDispatchCountdoes not compile becausethis.poisonMessageHandler?.MaxDispatchCountisint?, making the comparison resultbool?(which can’t be used byWhere). Coalesce the max dispatch count to anint(e.g.int.MaxValuewhen the handler is null) before comparing.
var poisonEvents = runtimeState.NewEvents.Where(evt => evt.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount);
src/DurableTask.Core/TaskEntityDispatcher.cs:464
request.DispatchCount > this.poisonMessageHandler?.MaxDispatchCountwon’t compile because the RHS becomesint?and the comparison producesbool?. Coalesce the handler max to anint(e.g.int.MaxValuewhen no handler is available).
bool isPoisonMessage = request.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount;
src/DurableTask.Core/TaskEntityDispatcher.cs:584
eventRaisedEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCountwon’t compile because the RHS becomesint?and the comparison producesbool?. Coalesce the handler max to anint(e.g.int.MaxValuewhen no handler is available).
bool isPoisonMessage = eventRaisedEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount;
src/DurableTask.Core/TaskEntityDispatcher.cs:764
if (request.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount)won’t compile because the RHS becomesint?and the comparison producesbool?. Coalesce the handler max to anint(e.g.int.MaxValuewhen no handler is available).
if (request.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount)
src/DurableTask.Core/TaskEntityDispatcher.cs:1106
Any(op => op.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount)won’t compile because the RHS becomesint?and the comparison producesbool?. Coalesce the handler max to anint(e.g.int.MaxValuewhen no handler is available).
bool poisonMessagesExist = workToDoNow.Operations.Any(op => op.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount);
src/DurableTask.Core/TaskEntityDispatcher.cs:993
SendLockResponseMessagenow requires a non-nullFailureDetails, but the call site passesnullfor non-poison cases. Make the parameter nullable to matchResponseMessage.FailureDetailsand the call sites.
void SendLockResponseMessage(WorkItemEffects effects, OrchestrationInstance target, Guid requestId, FailureDetails failureDetails)
src/DurableTask.Core/TaskActivityDispatcher.cs:183
- The poison dispatch-count check compares against
this.poisonMessageHandler?.MaxDispatchCount, which isint?. This makes the comparison abool?and won’t compile, and it also makes the failure message rely onpoisonMessageHandlerbeing non-null. Capture a non-nullmaxDispatchCountfirst (usingint.MaxValuewhen the handler is null) and use that for both the comparison and message.
if (poisonMessageReason == null && scheduledEvent.DispatchCount > this.poisonMessageHandler?.MaxDispatchCount)
{
poisonMessageReason = $"Activity has received an event with dispatch count {taskMessage.Event.DispatchCount} which exceeds the maximum dispatch " +
$"count of {this.poisonMessageHandler.MaxDispatchCount}. The task will be failed.";
}
src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs:200
instancePoisonMessageContainer/activityPoisonMessageContainerarereadonlybut are only assigned whenIsPoisonMessageStorageEnabledis true. This fails definite-assignment rules and won’t compile when poison storage is disabled. Initialize the container references unconditionally (it should be a cheap reference object) and continue to guard actual usage behindIsPoisonMessageStorageEnabled.
if (this.settings.IsPoisonMessageStorageEnabled)
{
string prefix = string.IsNullOrEmpty(this.settings.PoisonMessageStorageContainerNamePrefix)
? "durable-task-poison"
: this.settings.PoisonMessageStorageContainerNamePrefix;
this.instancePoisonMessageContainer = this.azureStorageClient.GetBlobContainerReference($"{prefix}-instance-messages");
this.activityPoisonMessageContainer = this.azureStorageClient.GetBlobContainerReference($"{prefix}-activity-messages");
}
This PR introduces poison message handling to the dispatchers. This is done by
IPoisonMessageHandlerthat the any orchestration service which has poison message handling is expected to implementThe orchestration service is otherwise responsible for determining what to do with the poison message(s) and how to store them.