Fix Akka.Remote serialization exception bubbling and messages (#5072)

* Fix exception bubbling and messages

* Remove reference to DData project

* Remove reference to Akka.Cluster

* Remove generic exception handling

* Add a very specific exception handling for failed payload deserialization

Co-authored-by: Aaron Stannard <aaron@petabridge.com>

src/core/Akka.Remote.Tests/Serialization/BugFix5062Spec.cs

@@ -0,0 +1,88 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reflection;
+using System.Runtime.Serialization;
+using System.Text;
+using System.Threading.Tasks;
+using Akka.Actor;
+using Akka.Configuration;
+using Akka.Remote.Configuration;
+using Akka.Serialization;
+using Akka.TestKit;
+using FluentAssertions;
+using Google.Protobuf;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Akka.Remote.Tests.Serialization
+{
+    public class BugFix5062Spec: AkkaSpec
+    {
+        private static Config DDataConfig => ConfigurationFactory.ParseString($@"
+            akka.actor {{
+                serializers {{
+                    dummyWithManifest = ""{typeof(DummySerializerWithStringManifest).AssemblyQualifiedName}""
+                }}
+                serialization-bindings {{
+                    ""{typeof(SomeMessage).AssemblyQualifiedName}"" = dummyWithManifest
+                }}
+                serialization-identifiers {{
+	                ""{typeof(DummySerializerWithStringManifest).AssemblyQualifiedName}"" = 13
+                }}
+            }}")
+            .WithFallback(RemoteConfigFactory.Default());
+
+        public BugFix5062Spec(ITestOutputHelper output) : base(output, DDataConfig)
+        { }
+
+        [Fact]
+        public void Failed_serialization_should_give_proper_exception_message()
+        {
+            var childName = "dummy";
+            var message = new ActorSelectionMessage(
+                new SomeMessage(), 
+                new SelectionPathElement[] { new SelectChildName(childName) }, 
+                true);
+
+            var node1 = new Address("akka.tcp", "Sys", "localhost", 2551);
+            var serialized = MessageSerializer.Serialize((ExtendedActorSystem)Sys, node1, message);
+
+            var o = new object();
+            o.Invoking(s => MessageSerializer.Deserialize((ExtendedActorSystem)Sys, serialized)).Should()
+                .Throw<SerializationException>()
+                .WithMessage($"Failed to deserialize payload object when deserializing {nameof(ActorSelectionMessage)} addressed to [{childName}]")
+                .WithInnerExceptionExactly<NotImplementedException>();
+        }
+
+        public class SomeMessage
+        {
+        }
+
+        public class DummySerializerWithStringManifest : SerializerWithStringManifest
+        {
+            public DummySerializerWithStringManifest(ExtendedActorSystem system) : base(system)
+            {
+            }
+
+            public override byte[] ToBinary(object obj)
+            {
+                return Array.Empty<byte>();
+            }
+
+            public override object FromBinary(byte[] bytes, string manifest)
+            {
+                throw new NotImplementedException();
+            }
+
+            public override string Manifest(object o)
+            {
+                if (o is SomeMessage)
+                    return "SM";
+                throw new Exception("Unknown object type");
+            }
+        }
+
+
+    }
+}

src/core/Akka.Remote/Serialization/MessageContainerSerializer.cs

@@ -7,7 +7,9 @@
 
 using System;
 using System.Linq;
+using System.Runtime.Serialization;
 using Akka.Actor;
+using Akka.Remote.Serialization.Proto.Msg;
 using Akka.Serialization;
 using Akka.Util;
 using Google.Protobuf;
@@ -71,7 +73,6 @@ public override byte[] ToBinary(object obj)
         public override object FromBinary(byte[] bytes, Type type)
         {
             var selectionEnvelope = Proto.Msg.SelectionEnvelope.Parser.ParseFrom(bytes);
-            var message = _payloadSupport.PayloadFrom(selectionEnvelope.Payload);
 
             var elements = new SelectionPathElement[selectionEnvelope.Pattern.Count];
             for (var i = 0; i < selectionEnvelope.Pattern.Count; i++)
@@ -85,6 +86,17 @@ public override object FromBinary(byte[] bytes, Type type)
                     elements[i] = new SelectParent();
             }
 
+            object message;
+            try
+            {
+                message = _payloadSupport.PayloadFrom(selectionEnvelope.Payload);
+            }
+            catch (Exception ex)
+            {
+                throw new SerializationException(
+                    $"Failed to deserialize payload object when deserializing {nameof(ActorSelectionMessage)} addressed to [{string.Join(",", elements.Select(e => e.ToString()))}]", ex);
+            }
+
             return new ActorSelectionMessage(message, elements);
         }