Akka.Remote: log all layers of wrapped messages during errors (#6818)

Make sure we capture what the real underlying message types are when we encounter issues like serialization errors, oversized payload exceptions, and so on.

Author: Aaronontheweb

src/core/Akka.Remote.Tests/RemotingSpec.cs

@@ -11,6 +11,7 @@
 using Akka.Actor;
 using Akka.Actor.Dsl;
 using Akka.Configuration;
+using Akka.Event;
 using Akka.Remote.Transport;
 using Akka.Routing;
 using Akka.TestKit;
@@ -651,6 +652,21 @@ await EventFilter.Exception<OversizedPayloadException>(start: "Discarding oversi
                 });
         }
 
+        /// <summary>
+        /// Validate that we can accurately log wrapped messages that fail to be delivered
+        /// </summary>
+        [Fact]
+        public void Log_Wrapped_messages_that_fail_to_Send()
+        {
+            // 2x wrapped message
+            var wrappedMessage =
+                new DeadLetter(new ActorSelectionMessage("hit", Array.Empty<SelectionPathElement>(), false), TestActor,
+                    TestActor);
+
+            var loggedType = EndpointWriter.LogPossiblyWrappedMessageType(wrappedMessage);
+            loggedType.Should().Contain("DeadLetter").And.Contain("ActorSelectionMessage").And.Contain("String");
+        }
+
         [Fact]
         public async Task Drop_received_messages_over_payload_size()
         {

src/core/Akka.Remote.Tests/RemotingTerminatorSpecs.cs

@@ -119,9 +119,13 @@ await EventFilter.Exception<ShutDownAssociation>().ExpectAsync(0,
                     (await associated.Ask<ActorIdentity>(new Identify("foo"), RemainingOrDefault)).MessageId.ShouldBe("foo");
 
                     // terminate the DEPLOYED system
-                    Assert.True(await _sys2.Terminate().AwaitWithTimeout(10.Seconds()), "Expected to terminate within 10 seconds, but didn't.");
-                    await ExpectTerminatedAsync(associated); // expect that the remote deployed actor is dead
-                    
+                    await WithinAsync(TimeSpan.FromSeconds(10), async () =>
+                    {
+                        var terminationTask = _sys2.Terminate(); // start termination process
+                        await ExpectTerminatedAsync(associated);  // expect that the remote deployed actor is dead
+                        Assert.True(await terminationTask.AwaitWithTimeout(RemainingOrDefault), "Expected to terminate within 10 seconds, but didn't.");
+                    });
+
                     // now terminate the DEPLOYER system
                     Assert.True(await Sys.Terminate().AwaitWithTimeout(10.Seconds()), "Expected to terminate within 10 seconds, but didn't.");
                 });

src/core/Akka.Remote/Endpoint.cs

@@ -11,6 +11,7 @@
 using System.Diagnostics;
 using System.Linq;
 using System.Runtime.Serialization;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Akka.Actor;
@@ -1462,6 +1463,41 @@ private void BecomeWritingOrSendBufferedMessages()
             }
         }
 
+        /// <summary>
+        /// Unwraps <see cref="IWrappedMessage"/> in order to help make it easier to troubleshoot
+        /// which oversized message was sent.
+        /// </summary>
+        /// <returns>The formatted type string.</returns>
+        /// <remarks>
+        /// Internal for testing purposes only.
+        /// </remarks>
+        internal static string LogPossiblyWrappedMessageType(object failedMsg)
+        {
+            if (failedMsg is IWrappedMessage wrappedMessage)
+            {
+                static void LogWrapped(StringBuilder builder, IWrappedMessage nextMsg)
+                {
+                    builder.Append($"{nextMsg.GetType()}-->");
+                    if (nextMsg.Message is IWrappedMessage wrappedAgain)
+                    {
+                        builder.Append('(');
+                        LogWrapped(builder, wrappedAgain); // recursively iterate through all layers of wrapping
+                        builder.Append(')');
+                    }
+                    else
+                    {
+                        builder.Append(nextMsg.Message.GetType());
+                    }
+                }
+                
+                var builder = new StringBuilder();
+                LogWrapped(builder, wrappedMessage);
+                return builder.ToString();
+            }
+
+            return failedMsg.GetType().ToString();
+        }
+
         private bool WriteSend(EndpointManager.Send send)
         {
             try
@@ -1486,7 +1522,7 @@ private bool WriteSend(EndpointManager.Send send)
                         string.Format("Discarding oversized payload sent to {0}: max allowed size {1} bytes, actual size of encoded {2} was {3} bytes.",
                             send.Recipient,
                             Transport.MaximumPayloadBytes,
-                            send.Message.GetType(),
+                            LogPossiblyWrappedMessageType(send.Message),
                             pdu.Length));
                     _log.Error(reason, "Transient association error (association remains live)");
                     return true;
@@ -1509,15 +1545,15 @@ private bool WriteSend(EndpointManager.Send send)
                 _log.Error(
                   ex,
                   "Serialization failed for message [{0}]. Transient association error (association remains live)",
-                  send.Message.GetType());
+                  LogPossiblyWrappedMessageType(send.Message));
                 return true;
             }
             catch (ArgumentException ex)
             {
                 _log.Error(
                   ex,
                   "Serializer threw ArgumentException for message type [{0}]. Transient association error (association remains live)",
-                  send.Message.GetType());
+                  LogPossiblyWrappedMessageType(send.Message));
                 return true;
             }
             catch (EndpointException ex)