Cancelled sinks are blocking other `BroadcastHub` consumers

[Arkatufus]

Version Information
Version of Akka.NET? 1.5.40
Which Akka.NET Modules? Streams

Describe the bug

Cancelled sinks are blocking other BroadcastHub consumers. This is the same exact BroadcastHub problem as described here: akka/akka-core#23205 (comment)

To Reproduce

[Fact]
public async Task BroadcastHub_must_handle_cancelled_Sink()
{
    await this.AssertAllStagesStoppedAsync(async () =>
    {
        var upstream = this.CreatePublisherProbe<int>();
        var hubSource = Source.FromPublisher(upstream).RunWith(BroadcastHub.Sink<int>(4), Materializer);
        var downstream = this.CreateSubscriberProbe<int>();
        
        hubSource.RunWith(Sink.Cancelled<int>(), Materializer);
        hubSource.RunWith(Sink.FromSubscriber(downstream), Materializer);
        
        await downstream.EnsureSubscriptionAsync();
        
        await downstream.RequestAsync(10);
        await upstream.ExpectRequestAsync();
        await upstream.SendNextAsync(1);
        await downstream.ExpectNextAsync(1);
        await upstream.SendNextAsync(2);
        await downstream.ExpectNextAsync(2);
        await upstream.SendNextAsync(3);
        await downstream.ExpectNextAsync(3);
        await upstream.SendNextAsync(4);
        await downstream.ExpectNextAsync(4);
        await upstream.SendNextAsync(5);
        await downstream.ExpectNextAsync(5);
        
        await upstream.SendCompleteAsync();
        await downstream.ExpectCompleteAsync();
    }, Materializer);
}

Expected behavior

downstream should receive the 5th element

Actual behavior

Timeout while waiting for message exception on the await downstream.ExpectNextAsync(5); line.

[Arkatufus]

Might be related to #7253

[Arkatufus]

Latest PR for our code: akka/akka-core#21855

To fix this, we might need to port akka/akka-core#23185, akka/akka-core#24073, and akka/akka-core#23970

[Arkatufus]

Some notes on porting akka/akka-core#23185:

• All async inputs are tagged with a Promise in scala; TaskCompletionSource (tcs) in C#.
• Tcs is stored inside a concurrent collection for each callback invocation. This is the invoke() or invokeWithFeedback() method in scala trait AsyncCallback. We don't have this in C#, the closest code would be this code in the ActorGraphInterpreter. So to port this, we would need to create a variant of GetAsyncCallback that returns a Task that can be awaited. The old code needs to be augmented to pass in a marker completed tcs that gets skipped for the "fast path".
• Tcs is marked as done or exception is set when the scala runAsyncInput() method is executed.
• The equivalent runAsyncInput() code in our code base is this code inside the GraphInterpreter.
• Tcs is then removed from the collection inside the onFeedbackDispatched() method in the GraphStageLogic
• All tcs is set with an exception when the graph completes.

[Arkatufus]

One thing that is different between scala and .NET is the was async callback is stored and invoked. In scala, all future invocations are stored inside a data structure called AsyncCallback, in our code, we don't have this data structure, we just return an Action<T> or a Func<T, Task> with the new API introduced in #7579.

The biggest difference is that in the scala implementation, they can store state information about the async callback (pending, initializing, initialized, etc) and modify how async callback invocations are handled based on those states. We don't have this luxury and we might need to defer the callback invocations with other means.

[Aaronontheweb]

So I definitely thing we probably need a data structure around the AsyncCallback bits - and that doesn't necessarily mean breaking the API signature, btw. We can maintain the existing one but just change the internals around how the GraphInterpreter stores this data.