dotnet / dotnet/runtime

Dataflow EnsureOrdered messages skipped when exception thrown

Open
#121,368 3 comments 0 reactions 0 assignees View on GitHub
area-System.Threading.Tasks
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.