Dataflow EnsureOrdered messages skipped when exception thrown
- Dominant language
- C#
- Stars
- 18.3k
- Forks
- 5.6k
- PR merge metrics
- PR metrics pending
Description
### Description
When using the dataflow block option [`EnsureOrdered`](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.dataflow.dataflowblockoptions.ensureordered?view=net-9.0#system-threading-tasks-dataflow-dataflowblockoptions-ensureordered), an exception will not prevent subsequent messages from escaping a block like `TransformBlock<,>`.
I understand this is a gray area in the requirements for `EnsureOrdered`. However, users like me may attempt to execute logic in a downstream block that is not safe if one or more previous messages have not been processed. In my case, the logic was committing a monotonically increasing offset to a Kafka broker.
### Reproduction Steps
```csharp
using System.Threading.Tasks.Dataflow;
var processor = new TransformBlock(
num =>
{
if (num == 1)
{
Thread.Sleep(5);
throw new Exception("Failed processing first message.");
}
return num;
},
new ExecutionDataflowBlockOptions
{
EnsureOrdered = true,
// Only happens when parallelism > 1.
MaxDegreeOfParallelism = 2,
});
// Downstream block cares about order.
var logger = new ActionBlock(x => Console.WriteLine($"Index {x} emitted."));
processor.LinkTo(logger, new DataflowLinkOptions { PropagateCompletion = true });
for (var i = 1; i <= 5; i++)
{
await processor.SendAsync(i);
}
processor.Complete();
await Task.WhenAll(transformer.Completion, logger.Completion);
```
```
Index 2 emitted.
Index 3 emitted.
Index 4 emitted.
Index 5 emitted.
Unhandled exception. System.Exception: Failed processing first message.
at Program.<>c.<$>b__0_0(Int32 num) in C:\Users\kylem\Desktop\HelloDataflow\Program.cs:line 9
at System.Threading.Tasks.Dataflow.TransformBlock`2.ProcessMessage(Func`2 transform, KeyValuePair`2 messageWithId)
at System.Threading.Tasks.Dataflow.TransformBlock`2.<>c__DisplayClass9_0.<.ctor>b__3(KeyValuePair`2 messageWithId)
at System.Threading.Tasks.Dataflow.Internal.TargetCore`1.ProcessMessagesLoopCore()
--- End of stack trace from previous location ---
at Program.$(String[] args) in C:\Users\kylem\Desktop\HelloDataflow\Program.cs:line 29
at Program.(String[] args)
```
### Expected behavior
If message processing throws an exception while order is preserved, subsequently published messages should not be emitted given that a previous message was not successfully processed.
### Actual behavior
The message triggering the exception is skipped. Any subsequent buffered messages are emitted in order prior to the block's completion in the faulted state.
### Regression?
_No response_
### Known Workarounds
_No response_
### Configuration
.NET 9
Windows 11
### Other information
_No response_
Contributor guide
Assessment
This issue has not been assessed yet.