Akka Stream AlsoTo may not be failing graph when its sink throws exception

[richardjharding]

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

Describe the bug
A simple graph with source, sink and an intermediate AlsoTo stage with its own sink, if the sink referenced by the AlsoTo flow throws an exception the rest of the graph completes instead of failing the complete graph

To Reproduce

The following console app I think demonstrates the issue

// See https://aka.ms/new-console-template for more information
using Akka.Actor;
using Akka.Streams;
using Akka.Streams.Dsl;

Console.WriteLine("Starting Akka Stream");

var system = ActorSystem.Create("MySystem");
var materializer = system.Materializer();

var source = Source.From(Enumerable.Range(1, 100));

var sink = Sink.ForEach<int>(i => Console.WriteLine($"Main sink {i}"));

var alsoSink = Sink.ForEach<int>(i => throw new Exception("sink failed") );


var alsoFlow = Flow.Create<int>()
                .WireTap(i => Console.WriteLine($"also wiretap: {i}"))
                .To(alsoSink);

var task = source
           .AlsoTo(alsoFlow)           
           .ToMaterialized(sink, Keep.Right).Run(materializer);

await task;

Links to working reproductions on Github / Gitlab are very much appreciated

Expected behavior

When the AlsoTo flow fails the complete graph should fail

Actual behavior
No further output is seen from the AlsoTo stage but the other sink continues to receive elements

Environment

Windows 11, Dotnet 8

Additional context

I first noticed this when working with alpakka where the AlsoTo stream sink was an Azure Service bus queue, the main flow reading from an Azure Event Hub and it continued to emit elements even after the service bus sink threw an exception

Note no exception is returned by the failed AlsoTo stage in my Azure example the actor starting the stream has the task piped back to itself - and no fail message is received

[Arkatufus]

Internally, AlsoTo uses a Broadcast stage that can cancel the stream when an exception was thrown, we could expand the API to allow users to opt-in on this behavior

[Aaronontheweb]

@richardjharding this has been fixed - @Arkatufus can provide some color on how to do that via a new parameter we're enabling in v1.5.27

[Arkatufus]

Sure thing. We have a new AlsoTo() API that takes a second bool argument. The second argument is an opt-in flag that if set to true, will let the underlying Broadcast stage to propagate any cancellation/failures to the original stream.