[Streams] Fix null exceptions being propagated on downstream completion (#7497)

* [Streams] Fix null exceptions being propagated on downstream completion

* Add missing exception code

* Update API Approval list

Author: Arkatufus

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveStreams.DotNet.verified.txt

@@ -4128,6 +4128,10 @@ namespace Akka.Streams.Implementation.Fusing
         protected override Akka.Streams.Stage.GraphStageLogic CreateLogic(Akka.Streams.Attributes inheritedAttributes) { }
         public override string ToString() { }
     }
+    public class DownstreamCompletedWithNoCauseException : System.Exception
+    {
+        public DownstreamCompletedWithNoCauseException() { }
+    }
     [Akka.Annotations.InternalApiAttribute()]
     public sealed class Expand<TIn, TOut> : Akka.Streams.Stage.GraphStage<Akka.Streams.FlowShape<TIn, TOut>>
     {

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveStreams.Net.verified.txt

@@ -4102,6 +4102,10 @@ namespace Akka.Streams.Implementation.Fusing
         protected override Akka.Streams.Stage.GraphStageLogic CreateLogic(Akka.Streams.Attributes inheritedAttributes) { }
         public override string ToString() { }
     }
+    public class DownstreamCompletedWithNoCauseException : System.Exception
+    {
+        public DownstreamCompletedWithNoCauseException() { }
+    }
     [Akka.Annotations.InternalApiAttribute()]
     public sealed class Expand<TIn, TOut> : Akka.Streams.Stage.GraphStage<Akka.Streams.FlowShape<TIn, TOut>>
     {

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

@@ -0,0 +1,17 @@
+// -----------------------------------------------------------------------
+//  <copyright file="DownstreamCompletedWithNoCauseException.cs" company="Akka.NET Project">
+//      Copyright (C) 2009-2025 Lightbend Inc. <http://www.lightbend.com>
+//      Copyright (C) 2013-2025 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using System;
+
+namespace Akka.Streams.Implementation.Fusing;
+
+public class DownstreamCompletedWithNoCauseException: Exception
+{
+    public DownstreamCompletedWithNoCauseException() : base("Downstream stage/flow completed with no cause")
+    {
+    }
+}

src/core/Akka.Streams/Stage/GraphStage.cs

@@ -1297,10 +1297,25 @@ public void InternalOnDownstreamFinish(Exception cause)
         {
             try
             {
-                if (cause == null)
-                    throw new ArgumentException("Cancellation cause must not be null", nameof(cause));
                 if (_lastCancellationCause != null)
                     throw new ArgumentException("OnDownstreamFinish must not be called recursively", nameof(cause));
+                
+                // Some stages might propagate null exceptions due to improper Task continuation handling
+                // (see https://github.com/akkadotnet/Akka.Persistence.Sql/issues/498)
+                // This is a stop gap solution to make sure that Akka.Streams doesn't behave improperly
+                // until we can fix all of those
+                if (cause is null)
+                {
+                    try
+                    {
+                        throw new DownstreamCompletedWithNoCauseException();
+                    }
+                    catch (DownstreamCompletedWithNoCauseException e)
+                    {
+                        cause = e;
+                    }
+                }
+                
                 _lastCancellationCause = cause;
                 CancelStage(_lastCancellationCause);
             }