Fix #7534

Author: Arkatufus

src/core/Akka.Streams/Implementation/Fusing/Ops.cs

@@ -2547,6 +2547,8 @@ private sealed class Logic : InAndOutGraphStageLogic
         {
             private sealed class Holder<T>(object? message, Result<T> element)
             {
+                private Directive? _cachedDirective = null;
+                
                 public object? Message { get; } = message;
 
                 public Result<T> Element { get; private set; } = element;
@@ -2557,6 +2559,12 @@ public void SetElement(Result<T> result)
                         ? Result.Failure<T>(ReactiveStreamsCompliance.ElementMustNotBeNullException)
                         : result;
                 }
+
+                public Directive SupervisionDirectiveFor(Decider decider, Exception ex)
+                {
+                    _cachedDirective ??= decider(ex);
+                    return _cachedDirective.Value;
+                }
             }
 
             private static readonly Result<TOut> NotYetThere = Result.Failure<TOut>(new Exception());
@@ -2673,7 +2681,7 @@ private void PushOne()
                         {
                             // this could happen if we are looping in PushOne and end up on a failed Task before the
                             // HolderCompleted callback has run
-                            var strategy = _decider(result.Exception);
+                            var strategy = holder.SupervisionDirectiveFor(_decider, result.Exception!);
                             Log.Error(result.Exception, "An exception occured inside SelectAsync while processing message [{0}]. Supervision strategy: {1}", holder.Message, strategy);
                             switch (strategy)
                             {
@@ -2720,7 +2728,7 @@ private void HolderCompleted(Holder<TOut> holder, Result<TOut> result)
                 }
 
                 var exception = result.Exception;
-                var strategy = _decider(exception);
+                var strategy = holder.SupervisionDirectiveFor(_decider, exception!);
                 Log.Error(exception, "An exception occured inside SelectAsync while executing Task. Supervision strategy: {0}", strategy);
                 switch (strategy)
                 {