Supress ActorSelectionMessage with DeadLetterSuppression (migrated from

akka/akka#28341) (#4889)

* for example the Cluster InitJoin message is marked with DeadLetterSuppression
  but was anyway logged because sent with actorSelection
* for other WrappedMessage than ActorSelectionMessage we shouldn't unwrap and publish
  the inner in SuppressedDeadLetter because that might loose some information
* therefore those are silenced in the DeadLetterListener instead

Better deadLetter logging of wrapped messages (migrated from akka/akka#28253)

Logging of UnhandledMessage (migrated from akka/akka#28414)
* make use of the existing logging of dead letter
  also for UnhandledMessage

Add Dropped to Akka.Actor (migrated partially from akka/akka#27160)
Log Dropped from DeadLetterListener

src/contrib/cluster/Akka.Cluster.Sharding/ShardedDaemonProcess.cs

@@ -78,7 +78,7 @@ public MessageExtractor(int maxNumberOfShards)
     /// and have the message types themselves carry identifiers.
     /// </para>
     /// </summary>
-    public sealed class ShardingEnvelope
+    public sealed class ShardingEnvelope: IWrappedMessage
     {
         public string EntityId { get; }
         public object Message { get; }

src/contrib/cluster/Akka.Cluster.Tools/PublishSubscribe/DistributedMessages.cs

@@ -353,7 +353,7 @@ public override string ToString()
     /// TBD
     /// </summary>
     [Serializable]
-    public sealed class Publish : IDistributedPubSubMessage, IEquatable<Publish>
+    public sealed class Publish : IDistributedPubSubMessage, IEquatable<Publish>, IWrappedMessage
     {
         /// <summary>
         /// TBD
@@ -420,7 +420,7 @@ public override string ToString()
     /// TBD
     /// </summary>
     [Serializable]
-    public sealed class Send : IDistributedPubSubMessage, IEquatable<Send>
+    public sealed class Send : IDistributedPubSubMessage, IEquatable<Send>, IWrappedMessage
     {
         /// <summary>
         /// TBD
@@ -487,7 +487,7 @@ public override string ToString()
     /// TBD
     /// </summary>
     [Serializable]
-    public sealed class SendToAll : IDistributedPubSubMessage, IEquatable<SendToAll>
+    public sealed class SendToAll : IDistributedPubSubMessage, IEquatable<SendToAll>, IWrappedMessage
     {
         /// <summary>
         /// TBD

src/core/Akka.API.Tests/CoreAPISpec.ApproveClusterSharding.approved.txt

@@ -253,7 +253,7 @@ namespace Akka.Cluster.Sharding
         public Akka.Cluster.Sharding.ShardedDaemonProcessSettings WithRole(string role) { }
         public Akka.Cluster.Sharding.ShardedDaemonProcessSettings WithShardingSettings(Akka.Cluster.Sharding.ClusterShardingSettings shardingSettings) { }
     }
-    public sealed class ShardingEnvelope
+    public sealed class ShardingEnvelope : Akka.Actor.IWrappedMessage
     {
         public ShardingEnvelope(string entityId, object message) { }
         public string EntityId { get; }

src/core/Akka.API.Tests/CoreAPISpec.ApproveClusterTools.approved.txt

@@ -229,7 +229,7 @@ namespace Akka.Cluster.Tools.PublishSubscribe
         public static Akka.Cluster.Tools.PublishSubscribe.GetTopics Instance { get; }
     }
     public interface IDistributedPubSubMessage { }
-    public sealed class Publish : Akka.Cluster.Tools.PublishSubscribe.IDistributedPubSubMessage, System.IEquatable<Akka.Cluster.Tools.PublishSubscribe.Publish>
+    public sealed class Publish : Akka.Actor.IWrappedMessage, Akka.Cluster.Tools.PublishSubscribe.IDistributedPubSubMessage, System.IEquatable<Akka.Cluster.Tools.PublishSubscribe.Publish>
     {
         public Publish(string topic, object message, bool sendOneMessageToEachGroup = False) { }
         public object Message { get; }
@@ -258,7 +258,7 @@ namespace Akka.Cluster.Tools.PublishSubscribe
         public override int GetHashCode() { }
         public override string ToString() { }
     }
-    public sealed class Send : Akka.Cluster.Tools.PublishSubscribe.IDistributedPubSubMessage, System.IEquatable<Akka.Cluster.Tools.PublishSubscribe.Send>
+    public sealed class Send : Akka.Actor.IWrappedMessage, Akka.Cluster.Tools.PublishSubscribe.IDistributedPubSubMessage, System.IEquatable<Akka.Cluster.Tools.PublishSubscribe.Send>
     {
         public Send(string path, object message, bool localAffinity = False) { }
         public bool LocalAffinity { get; }
@@ -269,7 +269,7 @@ namespace Akka.Cluster.Tools.PublishSubscribe
         public override int GetHashCode() { }
         public override string ToString() { }
     }
-    public sealed class SendToAll : Akka.Cluster.Tools.PublishSubscribe.IDistributedPubSubMessage, System.IEquatable<Akka.Cluster.Tools.PublishSubscribe.SendToAll>
+    public sealed class SendToAll : Akka.Actor.IWrappedMessage, Akka.Cluster.Tools.PublishSubscribe.IDistributedPubSubMessage, System.IEquatable<Akka.Cluster.Tools.PublishSubscribe.SendToAll>
     {
         public SendToAll(string path, object message, bool excludeSelf = False) { }
         public bool ExcludeSelf { get; }

src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt

@@ -320,7 +320,7 @@ namespace Akka.Actor
         public void Tell(object message, Akka.Actor.IActorRef sender = null) { }
         public override string ToString() { }
     }
-    public class ActorSelectionMessage : Akka.Actor.IAutoReceivedMessage, Akka.Actor.IPossiblyHarmful
+    public class ActorSelectionMessage : Akka.Actor.IAutoReceivedMessage, Akka.Actor.IPossiblyHarmful, Akka.Actor.IWrappedMessage
     {
         public ActorSelectionMessage(object message, Akka.Actor.SelectionPathElement[] elements, bool wildCardFanOut = False) { }
         public Akka.Actor.SelectionPathElement[] Elements { get; }
@@ -1145,6 +1145,10 @@ namespace Akka.Actor
         Akka.Actor.ITimerScheduler Timers { get; set; }
     }
     public interface IWithUnboundedStash : Akka.Actor.IActorStash, Akka.Dispatch.IRequiresMessageQueue<Akka.Dispatch.IUnboundedDequeBasedMessageQueueSemantics> { }
+    public interface IWrappedMessage
+    {
+        object Message { get; }
+    }
     public sealed class Identify : Akka.Actor.IAutoReceivedMessage, Akka.Actor.INotInfluenceReceiveTimeout
     {
         public Identify(object messageId) { }
@@ -1797,6 +1801,10 @@ namespace Akka.Actor
         protected void RunTask(System.Func<System.Threading.Tasks.Task> action) { }
     }
     public delegate void UntypedReceive(object message);
+    public class static WrappedMessage
+    {
+        public static object Unwrap(object message) { }
+    }
 }
 namespace Akka.Actor.Dsl
 {
@@ -2879,7 +2887,7 @@ namespace Akka.Event
     {
         protected ActorEventBus() { }
     }
-    public abstract class AllDeadLetters
+    public abstract class AllDeadLetters : Akka.Actor.IWrappedMessage
     {
         protected AllDeadLetters(object message, Akka.Actor.IActorRef sender, Akka.Actor.IActorRef recipient) { }
         public object Message { get; }
@@ -2933,6 +2941,12 @@ namespace Akka.Event
         protected virtual void Print(Akka.Event.LogEvent logEvent) { }
         protected override bool Receive(object message) { }
     }
+    public sealed class Dropped : Akka.Event.AllDeadLetters
+    {
+        public Dropped(object message, string reason, Akka.Actor.IActorRef sender, Akka.Actor.IActorRef recipient) { }
+        public Dropped(object message, string reason, Akka.Actor.IActorRef recipient) { }
+        public string Reason { get; }
+    }
     public class DummyClassForStringSources
     {
         public DummyClassForStringSources() { }
@@ -3155,12 +3169,9 @@ namespace Akka.Event
         public TraceLogger() { }
         protected override void OnReceive(object message) { }
     }
-    public sealed class UnhandledMessage
+    public sealed class UnhandledMessage : Akka.Event.AllDeadLetters, Akka.Actor.IWrappedMessage
     {
         public UnhandledMessage(object message, Akka.Actor.IActorRef sender, Akka.Actor.IActorRef recipient) { }
-        public object Message { get; }
-        public Akka.Actor.IActorRef Recipient { get; }
-        public Akka.Actor.IActorRef Sender { get; }
     }
     public class Warning : Akka.Event.LogEvent
     {
@@ -4088,7 +4099,7 @@ namespace Akka.Routing
             public Akka.Util.ISurrogated FromSurrogate(Akka.Actor.ActorSystem system) { }
         }
     }
-    public sealed class ConsistentHashableEnvelope : Akka.Routing.RouterEnvelope, Akka.Routing.IConsistentHashable
+    public sealed class ConsistentHashableEnvelope : Akka.Routing.RouterEnvelope, Akka.Actor.IWrappedMessage, Akka.Routing.IConsistentHashable
     {
         public ConsistentHashableEnvelope(object message, object hashKey) { }
         public object ConsistentHashKey { get; }

src/core/Akka.Tests/Actor/DeadLetterSupressionSpec.cs

@@ -82,6 +82,19 @@ public void Must_suppress_message_from_default_dead_letters_logging_sent_to_dead
             allDeadLetter.Recipient.Should().Be(deadActor);
 
             allListener.ExpectNoMsg(200.Milliseconds());
+
+            // unwrap for ActorSelection
+            Sys.ActorSelection(deadActor.Path).Tell(new SuppressedMessage());
+            Sys.ActorSelection(deadActor.Path).Tell(new NormalMessage());
+
+            // the recipient ref isn't the same as deadActor here so only checking the message
+            deadLetter = deadListener.ExpectMsg<DeadLetter>();//
+            deadLetter.Message.Should().BeOfType<NormalMessage>();
+            suppressedDeadLetter = suppressedListener.ExpectMsg<SuppressedDeadLetter>();
+            suppressedDeadLetter.Message.Should().BeOfType<SuppressedMessage>();
+
+            deadListener.ExpectNoMsg(200.Milliseconds());
+            suppressedListener.ExpectNoMsg(200.Milliseconds());
         }
 
         [Fact]
@@ -123,6 +136,22 @@ public void Must_suppress_message_from_default_dead_letters_logging_sent_to_dead
             deadListener.ExpectNoMsg(TimeSpan.Zero);
             suppressedListener.ExpectNoMsg(TimeSpan.Zero);
             allListener.ExpectNoMsg(TimeSpan.Zero);
+
+            // unwrap for ActorSelection
+            Sys.ActorSelection(Sys.DeadLetters.Path).Tell(new SuppressedMessage());
+            Sys.ActorSelection(Sys.DeadLetters.Path).Tell(new NormalMessage());
+
+            deadLetter = deadListener.ExpectMsg<DeadLetter>();
+            deadLetter.Message.Should().BeOfType<NormalMessage>();
+            deadLetter.Sender.Should().Be(TestActor);
+            deadLetter.Recipient.Should().Be(Sys.DeadLetters);
+            suppressedDeadLetter = suppressedListener.ExpectMsg<SuppressedDeadLetter>();
+            suppressedDeadLetter.Message.Should().BeOfType<SuppressedMessage>();
+            suppressedDeadLetter.Sender.Should().Be(TestActor);
+            suppressedDeadLetter.Recipient.Should().Be(Sys.DeadLetters);
+
+            deadListener.ExpectNoMsg(200.Milliseconds());
+            suppressedListener.ExpectNoMsg(200.Milliseconds());
         }
     }
 }

src/core/Akka.Tests/Actor/DeadLetterSuspensionSpec.cs

@@ -8,19 +8,54 @@
 using System.Threading;
 using Akka.Actor;
 using Akka.Configuration;
+using Akka.Event;
 using Akka.TestKit;
 using Xunit;
 
 namespace Akka.Tests.Actor
 {
     public class DeadLetterSuspensionSpec : AkkaSpec
     {
+        private class Dropping : ActorBase
+        {
+            public static Props Props() => Akka.Actor.Props.Create(() => new Dropping());
+
+            protected override bool Receive(object message)
+            {
+                switch (message)
+                {
+                    case int n:
+                        Context.System.EventStream.Publish(new Dropped(n, "Don't like numbers", Self));
+                        return true;
+                }
+                return false;
+            }
+        }
+
+        private class Unandled : ActorBase
+        {
+            public static Props Props() => Akka.Actor.Props.Create(() => new Unandled());
+
+            protected override bool Receive(object message)
+            {
+                switch (message)
+                {
+                    case int n:
+                        Unhandled(n);
+                        return true;
+                }
+                return false;
+            }
+        }
+
         private static readonly Config Config = ConfigurationFactory.ParseString(@"
             akka.loglevel = INFO
-            akka.log-dead-letters = 3
+            akka.log-dead-letters = 4
             akka.log-dead-letters-suspend-duration = 2s");
 
         private readonly IActorRef _deadActor;
+        private readonly IActorRef _droppingActor;
+        private readonly IActorRef _unhandledActor;
 
         public DeadLetterSuspensionSpec()
             : base(Config)
@@ -29,10 +64,20 @@ public DeadLetterSuspensionSpec()
             Watch(_deadActor);
             _deadActor.Tell(PoisonPill.Instance);
             ExpectTerminated(_deadActor);
+
+            _droppingActor = Sys.ActorOf(Dropping.Props(), "droppingActor");
+            _unhandledActor = Sys.ActorOf(Unandled.Props(), "unhandledActor");
         }
 
         private string ExpectedDeadLettersLogMessage(int count) =>
-            $"Message [{count.GetType().Name}] from {TestActor.Path} to {_deadActor.Path} was not delivered. [{count}] dead letters encountered";
+            $"Message [{count.GetType().Name}] from {TestActor} to {_deadActor} was not delivered. [{count}] dead letters encountered";
+
+        private string ExpectedDroppedLogMessage(int count) =>
+            $"Message [{count.GetType().Name}] to {_droppingActor} was dropped. Don't like numbers. [{count}] dead letters encountered";
+
+        private string ExpectedUnhandledLogMessage(int count) =>
+            $"Message [{count.GetType().Name}] from {TestActor} to {_unhandledActor} was unhandled. [{count}] dead letters encountered";
+
 
         [Fact]
         public void Must_suspend_dead_letters_logging_when_reaching_akka_log_dead_letters_and_then_re_enable()
@@ -42,28 +87,31 @@ public void Must_suspend_dead_letters_logging_when_reaching_akka_log_dead_letter
                 .Expect(1, () => _deadActor.Tell(1));
 
             EventFilter
-                .Info(start: ExpectedDeadLettersLogMessage(2))
-                .Expect(1, () => _deadActor.Tell(2));
+                .Info(start: ExpectedDroppedLogMessage(2))
+                .Expect(1, () => _droppingActor.Tell(2));
 
             EventFilter
-                .Info(start: ExpectedDeadLettersLogMessage(3) + ", no more dead letters will be logged in next")
-                .Expect(1, () => _deadActor.Tell(3));
+                .Info(start: ExpectedUnhandledLogMessage(3))
+                .Expect(1, () => _unhandledActor.Tell(3));
 
-            _deadActor.Tell(4);
+            EventFilter
+                .Info(start: ExpectedDeadLettersLogMessage(4) + ", no more dead letters will be logged in next")
+                .Expect(1, () => _deadActor.Tell(4));
             _deadActor.Tell(5);
+            _droppingActor.Tell(6);
 
             // let suspend-duration elapse
             Thread.Sleep(2050);
 
             // re-enabled
             EventFilter
-                .Info(start: ExpectedDeadLettersLogMessage(6) + ", of which 2 were not logged")
-                .Expect(1, () => _deadActor.Tell(6));
+                .Info(start: ExpectedDeadLettersLogMessage(7) + ", of which 2 were not logged")
+                .Expect(1, () => _deadActor.Tell(7));
 
             // reset count
             EventFilter
                 .Info(start: ExpectedDeadLettersLogMessage(1))
-                .Expect(1, () => _deadActor.Tell(7));
+                .Expect(1, () => _deadActor.Tell(8));
         }
     }
 }

src/core/Akka.Tests/Event/EventStreamSpec.cs

@@ -296,6 +296,7 @@ private static string GetDebugUnhandledMessagesConfig()
                 akka {
                     actor.serialize-messages = off
                     actor.debug.unhandled = on
+                    log-dead-letters = off
                     stdout-loglevel = DEBUG
                     loglevel = DEBUG
                     loggers = [""%logger%""]