=cluster #52 Leader actions on convergence, including .removals

Samples/Sources/SampleCluster/main.swift

@@ -15,13 +15,9 @@
 import Dispatch
 import DistributedActors
 
-print("Getting args")
-
 var args = CommandLine.arguments
 args.removeFirst()
 
-print("got args")
-
 print("\(args)")
 
 guard args.count >= 1 else {
@@ -31,7 +27,7 @@ guard args.count >= 1 else {
 let system = ActorSystem("System") { settings in
     settings.cluster.enabled = true
     settings.cluster.bindPort = Int(args[0])!
-    settings.cluster.downingStrategy = .none
+    settings.cluster.downingStrategy = .timeout(.default)
     settings.defaultLogLevel = .debug
 }
 
@@ -42,11 +38,8 @@ let ref = try system.spawn("hello", of: Cluster.Event.self, .receive { context,
 system.cluster.events.subscribe(ref)
 
 if args.count >= 3 {
-    print("getting host")
     let host = args[1]
-    print("parsing port")
     let port = Int(args[2])!
-    print("Joining")
     system.cluster.join(node: Node(systemName: "System", host: host, port: port))
 }
 

Samples/Sources/SampleDiningPhilosophers/Philosopher.swift

@@ -11,7 +11,6 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 //===----------------------------------------------------------------------===//
-
 import DistributedActors
 
 public class Philosopher {
@@ -115,7 +114,7 @@ public class Philosopher {
             case .forkReply(.busy(let fork)):
                 fatalError("Received fork busy response from an unexpected fork: \(fork)! Already in hand: \(inHand), and pending: \(pending)")
 
-            // Ignore others...
+                // Ignore others...
             case .think: return .ignore // since we'll decide to become thinking ourselves
             case .eat: return .ignore // since we'll decide to become eating ourselves
             }

Sources/DistributedActors/ActorAddress.swift

@@ -564,19 +564,19 @@ public struct Node: Hashable {
 
 extension Node: CustomStringConvertible, CustomDebugStringConvertible {
     public var description: String {
-        return "\(self.protocol)://\(self.systemName)@\(self.host):\(self.port)"
+        "\(self.protocol)://\(self.systemName)@\(self.host):\(self.port)"
     }
 
     public var debugDescription: String {
-        return self.description
+        self.description
     }
 }
 
 extension Node: Comparable {
     // Silly but good enough comparison for deciding "who is lower node"
     // as we only use those for "tie-breakers" any ordering is fine to be honest here.
     public static func < (lhs: Node, rhs: Node) -> Bool {
-        return "\(lhs)" < "\(rhs)"
+        "\(lhs)" < "\(rhs)"
     }
 
     public func hash(into hasher: inout Hasher) {
@@ -586,7 +586,7 @@ extension Node: Comparable {
     }
 
     public static func == (lhs: Node, rhs: Node) -> Bool {
-        return lhs.protocol == rhs.protocol && lhs.host == rhs.host && lhs.port == rhs.port
+        lhs.protocol == rhs.protocol && lhs.host == rhs.host && lhs.port == rhs.port
     }
 }
 
@@ -624,7 +624,7 @@ public struct UniqueNode: Hashable {
             self.node.host = newValue
         }
         get {
-            return self.node.host
+            self.node.host
         }
     }
 
@@ -633,14 +633,14 @@ public struct UniqueNode: Hashable {
             self.node.port = newValue
         }
         get {
-            return self.node.port
+            self.node.port
         }
     }
 }
 
 extension UniqueNode: CustomStringConvertible, CustomDebugStringConvertible {
     public var description: String {
-        return "\(self.node)"
+        "\(self.node)"
     }
 
     public var debugDescription: String {
@@ -652,7 +652,7 @@ extension UniqueNode: CustomStringConvertible, CustomDebugStringConvertible {
 extension UniqueNode: Comparable {
     public static func == (lhs: UniqueNode, rhs: UniqueNode) -> Bool {
         // we first compare the NodeIDs since they're quicker to compare and for diff systems always would differ, even if on same physical address
-        return lhs.nid == rhs.nid && lhs.node == rhs.node
+        lhs.nid == rhs.nid && lhs.node == rhs.node
     }
 
     // Silly but good enough comparison for deciding "who is lower node"

Sources/DistributedActors/ActorSystem.swift

@@ -267,6 +267,12 @@ public final class ActorSystem {
 
         /// Starts plugins after the system is fully initialized
         self.settings.plugins.startAll(self)
+
+        self.log.info("Actor System [\(self.name)] initialized.")
+        if settings.cluster.enabled {
+            self.log.info("Actor System Settings in effect: Cluster.autoLeaderElection: \(self.settings.cluster.autoLeaderElection)")
+            self.log.info("Actor System Settings in effect: Cluster.downingStrategy: \(self.settings.cluster.downingStrategy)")
+        }
     }
 
     public convenience init() {

Sources/DistributedActors/CRDT/CRDTReplicatorShell.swift

@@ -100,7 +100,7 @@ extension CRDT.Replicator {
                 self.remoteReplicators.remove(remoteReplicatorRef)
 
             case .snapshot(let snapshot):
-                Cluster.Membership.diff(from: .empty, to: snapshot).changes.forEach { change in
+                Cluster.Membership._diff(from: .empty, to: snapshot).changes.forEach { change in
                     self.receiveClusterEvent(context, event: .membershipChange(change))
                 }
 

Sources/DistributedActors/Clocks/VersionVector.swift

@@ -39,6 +39,8 @@ public struct VersionVector {
     // Internal state is a dictionary of replicas and their corresponding version
     internal var state: [ReplicaId: Int] = [:]
 
+    public static let empty: VersionVector = .init()
+
     public static func first(at replicaId: ReplicaId) -> Self {
         .init((replicaId, 1))
     }
@@ -89,6 +91,13 @@ public struct VersionVector {
         self.state.merge(other.state, uniquingKeysWith: max)
     }
 
+    /// Prune any trace of the passed in replica id.
+    public func pruneReplica(_ replicaId: ReplicaId) -> Self {
+        var s = self
+        s.state.removeValue(forKey: replicaId)
+        return s
+    }
+
     /// Obtain current version at the given replica. If the replica is unknown, the default version is 0.
     ///
     /// - Parameter replicaId: The replica whose version is being queried.