WiP: Cluster Singletons - Keep alive

src/Proto.Cluster/Cluster.cs

@@ -5,7 +5,6 @@
 // -----------------------------------------------------------------------
 using System;
 using System.Collections.Generic;
-using System.Collections.Immutable;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -15,6 +14,7 @@
 using Proto.Cluster.Identity;
 using Proto.Cluster.Metrics;
 using Proto.Cluster.PubSub;
+using Proto.Cluster.Singleton;
 using Proto.Extensions;
 using Proto.Remote;
 
@@ -43,11 +43,14 @@ public Cluster(ActorSystem system, ClusterConfig config)
             Gossip = new Gossiper(this);
             PidCache = new PidCache();
             PubSub = new PubSubManager(this);
+            Singleton = new SingletonManager(this);
 
             SubscribeToTopologyEvents();
         }
 
         public PubSubManager PubSub { get; }
+
+        public SingletonManager Singleton { get; }
 
         public static ILogger Logger { get; } = Log.CreateLogger<Cluster>();
 

src/Proto.Cluster/Gossip/GossipActor.cs

@@ -66,6 +66,8 @@ private Task OnGossipRequest(IContext context, GossipRequest gossipRequest)
                 {
                     context.System.EventStream.Publish(update);
                 }
+
+                context.System.EventStream.Publish(new Gossip(newState));
 
                 _state = newState;
                 CheckConsensus(context);

src/Proto.Cluster/Gossip/Gossiper.cs

@@ -13,16 +13,6 @@
 
 namespace Proto.Cluster.Gossip
 {
-    public record GossipUpdate(string MemberId, string Key, Any Value, long SequenceNumber);
-    public record GetGossipStateRequest(string Key);
-
-    public record GetGossipStateResponse(ImmutableDictionary<string,Any> State);
-
-    public record SetGossipStateKey(string Key, IMessage Value);
-
-    public record SendGossipStateRequest;
-    public record SendGossipStateResponse;
-
     public class Gossiper
     {
         public const string GossipActorName = "gossip";

src/Proto.Cluster/Gossip/Messages.cs

@@ -0,0 +1,24 @@
+// -----------------------------------------------------------------------
+// <copyright file="Messages.cs" company="Asynkron AB">
+//      Copyright (C) 2015-2021 Asynkron AB All rights reserved
+// </copyright>
+// -----------------------------------------------------------------------
+using System;
+using System.Collections.Immutable;
+using Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
+
+namespace Proto.Cluster.Gossip
+{
+    public record GossipUpdate(string MemberId, string Key, Any Value, long SequenceNumber);
+
+    public record Gossip(GossipState State);
+    public record GetGossipStateRequest(string Key);
+
+    public record GetGossipStateResponse(ImmutableDictionary<string,Any> State);
+
+    public record SetGossipStateKey(string Key, IMessage Value);
+
+    public record SendGossipStateRequest;
+    public record SendGossipStateResponse;
+}

src/Proto.Cluster/Singleton/SingletonManager.cs

@@ -0,0 +1,65 @@
+// -----------------------------------------------------------------------
+// <copyright file="SingletonManager.cs" company="Asynkron AB">
+//      Copyright (C) 2015-2021 Asynkron AB All rights reserved
+// </copyright>
+// -----------------------------------------------------------------------
+using System.Collections.Concurrent;
+using System.Collections.Immutable;
+using System.Linq;
+using Grpc.Core;
+using Proto.Cluster.Gossip;
+using Proto.Utils;
+
+namespace Proto.Cluster.Singleton
+{
+    public class SingletonManager
+    {
+        private readonly Cluster _cluster;
+        private ImmutableHashSet<string> _tracked = ImmutableHashSet<string>.Empty;
+        private readonly object _lock = new();
+
+        public SingletonManager(Cluster cluster)
+        {
+            _cluster = cluster;
+
+            cluster.System.EventStream.Subscribe<Gossip.Gossip>(g => {
+                    var tracked = _tracked;
+                    var existingKeys = (
+                            from member in g.State.Members
+                            from entry in member.Value.Values
+                            where entry.Key.StartsWith("Singleton-")
+                            select entry.Key)
+                        //.Distinct()
+                        .ToImmutableHashSet();
+
+                    var missing = tracked.Except(existingKeys);
+
+                    //iterate over all missing actors
+                    //send a touch message to them to activate
+                    foreach (var m in missing)
+                    {
+                        _ = cluster.RequestAsync<Touched>("", "", new Touch(), CancellationTokens.FromSeconds(5));
+                    }
+                }
+            );
+        }
+
+        public void Track(ClusterIdentity identity)
+        {
+            lock (_lock)
+            {
+                _tracked = _tracked.Add(Key(identity));
+            }
+        }
+
+        public void Untrack(ClusterIdentity identity)
+        {
+            lock (_lock)
+            {
+                _tracked = _tracked.Remove(Key(identity));
+            }
+        }
+
+        private static string Key(ClusterIdentity identity) => "Singleton-" + identity.ToDiagnosticString();
+    }
+}