ShardedDaemonProcess push mode (#7229)

* stashing work

* finished first pass at push-based `ShardedDaemonProcess`

close #7195

* fixed issues with message propagation

* made `ShardedDaemonProcessSettings` support `nullable`

* cleanup logging

* added standard function for name formatting

* Fix unit test

* added documentation

---------

Co-authored-by: Gregorius Soedharmo <[email protected]>

Author: Aaronontheweb

docs/articles/clustering/cluster-sharded-daemon-process.md

@@ -32,5 +32,20 @@ when starting up:
 
 ## Scalability  
 
-This cluster tool is intended for small numbers of consumers and will not scale well to a large set. In large clusters
-it is recommended to limit the nodes the sharded daemon process will run on using a role.
+This cluster tool is intended for small numbers of consumers and will not scale well to a large set. In large clusters it is recommended to limit the nodes the sharded daemon process will run on using a role.
+
+## Push-Based Communication
+
+[`ShardedDaemonProcess`](xref:Akka.Cluster.Sharding.ShardedDaemonProcess) also supports push-based communication not too dissimilar from a round-robin `Router`:
+
+[!code-csharp[IActorRef returned by ShardedDaemonProcess](../../../src/contrib/cluster/Akka.Cluster.Sharding.Tests/ShardedDaemonProcessProxySpec.cs?name=PushDaemon)]
+
+The `ShardedDaemonProcess.Init` call returns an `IActorRef` - any messages you send to this `IActorRef` will be routed to one of the `n` worker instances you specified.
+
+## Daemon Proxies
+
+You can also interact with `ShardedDaemonProcess` on nodes that don't use the same [Akka.Cluster role](xref:member-roles) as the `ShardedDaemonProcess` host itself.
+
+[!code-csharp[IActorRef returned by ShardedDaemonProcess](../../../src/contrib/cluster/Akka.Cluster.Sharding.Tests/ShardedDaemonProcessProxySpec.cs?name=PushDaemonProxy)]
+
+Under the covers we use a [`ShardRegionProxy`](xref:Akka.Cluster.Sharding.ClusterSharding#Akka_Cluster_Sharding_ClusterSharding_StartProxy_System_String_System_String_Akka_Cluster_Sharding_IMessageExtractor_) to forward any messages you send to the `IActorRef` returned by the `ShardedDaemonProcess.InitProxy` method.

src/contrib/cluster/Akka.Cluster.Sharding.Tests/ShardedDaemonProcessProxySpec.cs

@@ -0,0 +1,109 @@
+// -----------------------------------------------------------------------
+//  <copyright file="ShardedDaemonProcessProxySpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2009-2024 Lightbend Inc. <http://www.lightbend.com>
+//      Copyright (C) 2013-2024 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using System.Linq;
+using System.Threading.Tasks;
+using Akka.Actor;
+using Akka.Cluster.Tools.Singleton;
+using Akka.Configuration;
+using Akka.TestKit;
+using Xunit;
+using Xunit.Abstractions;
+using FluentAssertions;
+
+namespace Akka.Cluster.Sharding.Tests;
+
+public class ShardedDaemonProcessProxySpec : AkkaSpec
+{
+    private static Config GetConfig()
+    {
+        return ConfigurationFactory.ParseString(@"
+                akka.loglevel = DEBUG
+                akka.actor.provider = cluster
+                akka.remote.dot-netty.tcp.port = 0
+                akka.remote.dot-netty.tcp.hostname = localhost
+
+                # ping often/start fast for test
+                akka.cluster.sharded-daemon-process.keep-alive-interval = 1s
+                akka.cluster.roles = [workers]
+                akka.coordinated-shutdown.terminate-actor-system = off
+                akka.coordinated-shutdown.run-by-actor-system-terminate = off")
+            .WithFallback(ClusterSharding.DefaultConfig())
+            .WithFallback(ClusterSingletonProxy.DefaultConfig())
+            .WithFallback(DistributedData.DistributedData.DefaultConfig());
+    }
+
+    private readonly ActorSystem _proxySystem;
+
+    public ShardedDaemonProcessProxySpec(ITestOutputHelper output)
+        : base(GetConfig(), output: output)
+    {
+        _proxySystem = ActorSystem.Create(Sys.Name,
+            ConfigurationFactory
+                .ParseString("akka.cluster.roles=[proxy]").WithFallback(Sys.Settings.Config));
+        InitializeLogger(_proxySystem, "PROXY");
+    }
+    
+    private class EchoActor : ReceiveActor
+    {
+        public static Props EchoProps(int i) => Props.Create(() => new EchoActor());
+        
+        public EchoActor()
+        {
+            ReceiveAny(msg => Sender.Tell(msg));
+        }
+    }
+    
+    [Fact]
+    public async Task ShardedDaemonProcessProxy_must_start_daemon_process_on_proxy()
+    {
+        // form a cluster with one node
+        Cluster.Get(Sys).Join(Cluster.Get(Sys).SelfAddress);
+        Cluster.Get(_proxySystem).Join(Cluster.Get(Sys).SelfAddress);
+        
+        // validate that we have a 2 node cluster with both members marked as up
+        await AwaitAssertAsync(() =>
+        {
+            Cluster.Get(Sys).State.Members.Count(x => x.Status == MemberStatus.Up).Should().Be(2);
+            Cluster.Get(_proxySystem).State.Members.Count(x => x.Status == MemberStatus.Up).Should().Be(2);
+        });
+        
+        // <PushDaemon>
+        // start the daemon process on the host
+        var name = "daemonTest";
+        var targetRole = "workers";
+        var numWorkers = 10;
+        var settings = ShardedDaemonProcessSettings.Create(Sys).WithRole(targetRole);
+        IActorRef host = ShardedDaemonProcess.Get(Sys).Init(name, numWorkers, EchoActor.EchoProps, settings, PoisonPill.Instance);
+        
+        // ping some of the workers via the host
+        for(var i = 0; i < numWorkers; i++)
+        {
+            var result = await host.Ask<int>(i);
+            result.Should().Be(i);
+        }
+        // </PushDaemon>
+        
+        // <PushDaemonProxy>
+        // start the proxy on the proxy system, which runs on a different role not capable of hosting workers
+        IActorRef proxy = ShardedDaemonProcess.Get(_proxySystem).InitProxy(name, numWorkers, targetRole);
+        
+        // ping some of the workers via the proxy
+        for(var i = 0; i < numWorkers; i++)
+        {
+            var result = await proxy.Ask<int>(i);
+            result.Should().Be(i);
+        }
+        // </PushDaemonProxy>
+    }
+
+    protected override void AfterAll()
+    {
+        Shutdown(_proxySystem);
+        base.AfterAll();
+    }
+}

src/contrib/cluster/Akka.Cluster.Sharding.Tests/ShardedDaemonProcessSpec.cs

@@ -12,6 +12,7 @@
 using Akka.Configuration;
 using Akka.Event;
 using Akka.TestKit;
+using FluentAssertions;
 using Xunit;
 
 namespace Akka.Cluster.Sharding.Tests
@@ -21,7 +22,10 @@ public class ShardedDaemonProcessSpec : AkkaSpec
         private sealed class Stop
         {
             public static Stop Instance { get; } = new();
-            private Stop() { }
+
+            private Stop()
+            {
+            }
         }
 
         private sealed class Started
@@ -81,14 +85,16 @@ private static Config GetConfig()
 
         public ShardedDaemonProcessSpec()
             : base(GetConfig())
-        { }
+        {
+        }
 
         [Fact]
         public void ShardedDaemonProcess_must_have_a_single_node_cluster_running_first()
         {
             var probe = CreateTestProbe();
             Cluster.Get(Sys).Join(Cluster.Get(Sys).SelfAddress);
-            probe.AwaitAssert(() => Cluster.Get(Sys).SelfMember.Status.ShouldBe(MemberStatus.Up), TimeSpan.FromSeconds(3));
+            probe.AwaitAssert(() => Cluster.Get(Sys).SelfMember.Status.ShouldBe(MemberStatus.Up),
+                TimeSpan.FromSeconds(3));
         }
 
         [Fact]
@@ -126,7 +132,9 @@ public void ShardedDaemonProcess_must_not_run_if_the_role_does_not_match_node_ro
 
             var probe = CreateTestProbe();
             var settings = ShardedDaemonProcessSettings.Create(Sys).WithRole("workers");
-            ShardedDaemonProcess.Get(Sys).Init("roles", 3, id => MyActor.Props(id, probe.Ref), settings, PoisonPill.Instance);
+            var actorRef = ShardedDaemonProcess.Get(Sys).Init("roles", 3, id => MyActor.Props(id, probe.Ref), settings,
+                PoisonPill.Instance);
+            actorRef.Should().BeNull();
 
             probe.ExpectNoMsg();
         }
@@ -152,9 +160,11 @@ protected override void PreStart()
         private void DocExample()
         {
             #region tag-processing
+
             var tags = new[] { "tag-1", "tag-2", "tag-3" };
             ShardedDaemonProcess.Get(Sys).Init("TagProcessors", tags.Length, id => TagProcessor.Props(tags[id]));
+
             #endregion
         }
     }
-}
+}