Hardening process isolated, node watcher uses event to detect down

Author: ktoso

IntegrationTests/tests_02_process_isolated/it_ProcessIsolated_escalatingWorkers/main.swift

@@ -30,24 +30,37 @@ let isolated = ProcessIsolated { boot in
 
 pprint("Started process: \(getpid()) with roles: \(isolated.roles)")
 
-try isolated.run(on: .master) {
+struct OnPurposeBoom: Error {}
+
+isolated.run(on: .master) {
     isolated.spawnServantProcess(supervision: .restart(atMost: 1, within: nil), args: ["fatalError"])
+    isolated.spawnServantProcess(supervision: .restart(atMost: 1, within: nil), args: ["escalateError"])
 }
 
 try isolated.run(on: .servant) {
-    isolated.system.log.info("ISOLATED RUNNING")
+    isolated.system.log.info("ISOLATED RUNNING: \(CommandLine.arguments)")
 
-    // TODO: system should be configured to terminate HARD when a failure reaches the guardian
+    // TODO: assert command line arguments are the expected ones
 
-    let _: ActorRef<String> = try isolated.system.spawn("failing",
+    _ = try isolated.system.spawn("failed", of: String.self,
         props: Props().supervision(strategy: .escalate),
         .setup { context in
-            context.log.info("Spawned \(context.path) on servant node, it will fault with a [Boom].")
+            context.log.info("Spawned \(context.path) on servant node it will fail soon...")
             context.timers.startSingle(key: "explode", message: "Boom", delay: .seconds(1))
 
             return .receiveMessage { message in
-                fatalError("Faulting on purpose: \(message)")
-                return .stop
+                if CommandLine.arguments.contains("fatalError") {
+                    context.log.error("Time to crash with: fatalError")
+                    // crashes process since we do not isolate faults
+                    fatalError("FATAL ERROR ON PURPOSE")
+                } else if CommandLine.arguments.contains("escalateError") {
+                    context.log.error("Time to crash with: throwing an error, escalated to top level")
+                    // since we .escalate and are a top-level actor, this will cause the process to die as well
+                    throw OnPurposeBoom()
+                } else {
+                    context.log.error("MISSING FAILURE MODE ARGUMENT!!! Test is constructed not properly, or arguments were not passed properly.")
+                    fatalError("MISSING FAILURE MODE ARGUMENT!!! Test is constructed not properly, or arguments were not passed properly.")
+                }
             }
         })
 }

IntegrationTests/tests_02_process_isolated/test_04_failing_workers_to_cause_servant_restart.sh

@@ -39,7 +39,7 @@ swift build # synchronously ensure built
 
 swift run ${app_name} &
 
-await_n_processes "$app_name" 2
+await_n_processes "$app_name" 3
 
 pid_master=$(ps aux | grep ${app_name} | grep -v grep | grep -v servant | awk '{ print $2 }')
 pid_servant=$(ps aux | grep ${app_name} | grep -v grep | grep servant | head -n1 | awk '{ print $2 }')
@@ -53,12 +53,8 @@ echo '~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~'
 
 sleep 3 # TODO rather, sleep until another proc replaces the servant automatically
 
-echo '~~~~~~~~~~~~~ KILLED KILLED KILLED KILLED KILLED KILLED ~~~~~~~~~~~~~~~~~~~~~~~~~~~'
-ps aux | grep ${app_name}
-echo '~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~'
-
 # the 1 servant should die, but be restarted so we'll be back at two processes
-await_n_processes "$app_name" 2
+await_n_processes "$app_name" 3
 
 if [[ $(ps aux | awk '{print $2}' | grep ${pid_servant}  | grep -v 'grep' | wc -l) -ne 0 ]]; then
     echo "ERROR: Seems the servant was not killed!!!"

Package.swift

@@ -65,7 +65,7 @@ let targets: [PackageDescription.Target] = [
     // ==== ------------------------------------------------------------------------------------------------------------
     // MARK: Integration Tests - `it_` prefixed
     .target(
-        name: "it_ProcessIsolated_faultingWorkers",
+        name: "it_ProcessIsolated_escalatingWorkers",
         dependencies: [
             "DistributedActors",
         ],

Sources/DistributedActors/ActorShell.swift

@@ -138,11 +138,11 @@ internal final class ActorShell<Message>: ActorContext<Message>, AbstractActor {
 
         self.supervisor = Supervision.supervisorFor(system, initialBehavior: behavior, props: props.supervision)
 
-        if let failureDetectorRef = system._cluster?._nodeDeathWatcher {
-            self._deathWatch = DeathWatch(failureDetectorRef: failureDetectorRef)
+        if let nodeDeathWatcher = system._nodeDeathWatcher {
+            self._deathWatch = DeathWatch(nodeDeathWatcher: nodeDeathWatcher)
         } else {
             // FIXME; we could see if `myself` is the right one actually... rather than dead letters; if we know the FIRST actor ever is the failure detector one?
-            self._deathWatch = DeathWatch(failureDetectorRef: system.deadLetters.adapted())
+            self._deathWatch = DeathWatch(nodeDeathWatcher: system.deadLetters.adapted())
         }
 
         self.namingContext = ActorNamingContext()

Sources/DistributedActors/ActorSystem.swift

@@ -91,8 +91,10 @@ public final class ActorSystem {
 
     // initialized during startup
     internal var _cluster: ClusterShell?
-    internal var _clusterEventStream: EventStream<ClusterEvent>?
+    internal var _clusterEvents: EventStream<ClusterEvent>?
+    internal var _nodeDeathWatcher: NodeDeathWatcherShell.Ref?
 
+    // ==== ----------------------------------------------------------------------------------------------------------------
     // MARK: Logging
 
     public var log: Logger {
@@ -199,8 +201,16 @@ public final class ActorSystem {
         do {
             // Cluster MUST be the last thing we initialize, since once we're bound, we may receive incoming messages from other nodes
             if let cluster = self._cluster {
-                self._clusterEventStream = try! EventStream(self, name: "clusterEvents")
+                let clusterEvents = try! EventStream<ClusterEvent>(self, name: "clusterEvents")
+                self._clusterEvents = clusterEvents // TODO: why stored on self here?
                 _ = try cluster.start(system: self, eventStream: self.clusterEvents) // only spawns when cluster is initialized
+
+                // Node watcher MUST be started AFTER cluster and clusterEvents
+                self._nodeDeathWatcher = try self._spawnSystemActor(
+                    NodeDeathWatcherShell.naming,
+                    NodeDeathWatcherShell.behavior(clusterEvents: clusterEvents),
+                    perpetual: true
+                )
             }
         } catch {
             fatalError("Failed while starting cluster subsystem! Error: \(error)")

Sources/DistributedActors/Cluster/ActorSystem+Cluster.swift

@@ -61,6 +61,6 @@ extension ActorSystem {
     }
 
     internal var clusterEvents: EventStream<ClusterEvent> {
-        return self._clusterEventStream ?? EventStream(ref: self.deadLetters.adapted())
+        return self._clusterEvents ?? EventStream(ref: self.deadLetters.adapted())
     }
 }

Sources/DistributedActors/Cluster/ClusterShell.swift

@@ -93,21 +93,6 @@ internal class ClusterShell {
         return it
     }
 
-    // ==== ------------------------------------------------------------------------------------------------------------
-    // MARK: Node-Death Watcher
-
-    // Implementation notes: The `_failureDetectorRef` has to remain internally accessible.
-    // This is in order to solve a chicken-and-egg problem that we face during spawning of
-    // the first system actor that is the *failure detector* so it cannot reach to the systems
-    // value before it started...
-    var _nodeDeathWatcher: NodeDeathWatcherShell.Ref?
-    var nodeDeathWatcher: NodeDeathWatcherShell.Ref {
-        guard let it = self._nodeDeathWatcher else {
-            return fatalErrorBacktrace("Accessing ClusterShell.nodeDeathWatcher failed, was nil! This should never happen as access should only happen after start() was invoked.")
-        }
-        return it
-    }
-
     init() {
         self._associationsLock = Lock()
         self._associationsRegistry = [:]
@@ -116,21 +101,13 @@ internal class ClusterShell {
         // the single thing in the class it will modify is the associations registry, which we do to avoid actor queues when
         // remote refs need to obtain those
         //
-        // TODO: see if we can restructure this to avoid these nil/then-set dance
+        // FIXME: see if we can restructure this to avoid these nil/then-set dance
         self._ref = nil
-        self._nodeDeathWatcher = nil
     }
 
     /// Actually starts the shell which kicks off binding to a port, and all further cluster work
     internal func start(system: ActorSystem, eventStream: EventStream<ClusterEvent>) throws -> ClusterShell.Ref {
         self._serializationPool = try SerializationPool(settings: .default, serialization: system.serialization)
-
-        self._nodeDeathWatcher = try system._spawnSystemActor(
-            NodeDeathWatcherShell.naming,
-            NodeDeathWatcherShell.behavior(),
-            perpetual: true
-        )
-
         self._events = eventStream
 
         // TODO: concurrency... lock the ref as others may read it?
@@ -259,6 +236,7 @@ extension ClusterShell {
                 return self.onJoin(context, state: state, joining: node)
 
             case .handshakeWith(let remoteAddress, let replyTo):
+                state.logMembership()
                 return self.beginHandshake(context, state, with: remoteAddress, replyTo: replyTo)
             case .retryHandshake(let initiated):
                 return self.connectSendHandshakeOffer(context, state, initiated: initiated)
@@ -469,7 +447,7 @@ extension ClusterShell {
         case .initiated(var initiated):
             switch initiated.onHandshakeError(error) {
             case .scheduleRetryHandshake(let delay):
-                state.log.info("Schedule handshake retry to: [\(initiated.remoteNode)] delay: [\(delay)]")
+                state.log.debug("Schedule handshake retry to: [\(initiated.remoteNode)] delay: [\(delay)]")
                 context.timers.startSingle(
                     key: TimerKey("handshake-timer-\(remoteNode)"),
                     message: .command(.retryHandshake(initiated)),
@@ -619,7 +597,8 @@ extension ClusterShell {
         var state = state
 
         if let change = state.onMembershipChange(node, toStatus: .down) {
-            self.nodeDeathWatcher.tell(.forceDown(change.node))
+            // self.nodeDeathWatcher.tell(.forceDown(change.node))
+            self._events.publish(.membership(.memberDown(Member(node: change.node, status: .down))))
 
             if let logChangeLevel = state.settings.logMembershipChanges {
                 context.log.log(level: logChangeLevel, "Cluster membership change: \(reflecting: change), membership: \(state.membership)")

Sources/DistributedActors/Cluster/NodeDeathWatcher.swift

@@ -126,21 +126,31 @@ enum NodeDeathWatcherShell {
     /// it would be possible however to allow implementing the raw protocol by user actors if we ever see the need for it.
     internal enum Message {
         case remoteActorWatched(watcher: AddressableActorRef, remoteNode: UniqueNode)
-        case membershipSnapshot(Membership)
-        case membershipChange(MembershipChange)
-        case forceDown(UniqueNode) // TODO: this should go away with cluster events landing
+        case membershipSnapshot(Membership) // TODO: remove?
+        case membershipChange(MembershipChange) // TODO: remove as well
     }
 
-    static func behavior() -> Behavior<Message> {
+    static func behavior(clusterEvents: EventStream<ClusterEvent>) -> Behavior<Message> {
         return .setup { context in
-            // WARNING: DO NOT TOUCH context.system.cluster; we are started potentially before the cluster (!)
             let instance = NodeDeathWatcherInstance(selfNode: context.system.settings.cluster.uniqueBindNode)
+
+            context.system.cluster.events.subscribe(context.subReceive(ClusterEvent.self) { event in
+                context.log.info("EVENT::::: \(event)")
+                switch event {
+                case .membership(.memberDown(let member)):
+                    let change = MembershipChange(node: member.node, fromStatus: .none, toStatus: .down)
+                    instance.handleAddressDown(change)
+                default:
+                    () // ignore for now...
+                }
+            })
+
             return NodeDeathWatcherShell.behavior(instance)
         }
     }
 
     static func behavior(_ instance: NodeDeathWatcherInstance) -> Behavior<Message> {
-        return .receive { _, message in
+        return .receiveMessage { message in
 
             let lastMembership: Membership = .empty // TODO: To be mutated based on membership changes
 
@@ -157,11 +167,6 @@ enum NodeDeathWatcherShell {
 
             case .membershipChange(let change):
                 _ = instance.onMembershipChanged(change) // TODO: return and interpret directives
-
-            case .forceDown(let node):
-                // TODO: we'd get the change from subscribing to events and applying to local membership
-                let change = MembershipChange(node: node, fromStatus: .none, toStatus: .down)
-                instance.handleAddressDown(change)
             }
             return .same
         }

Sources/DistributedActors/Cluster/SWIM/SWIMSettings.swift

@@ -36,11 +36,11 @@ public struct SWIMSettings {
     /// These logs will contain SWIM.Instance metadata, as offered by `SWIM.Instance.metadata`.
     /// All logs will be prefixed using `[tracelog:SWIM]`, for easier grepping and inspecting only logs related to the SWIM instance.
     // TODO: how to make this nicely dynamically changeable during runtime
-    #if SACT_TRACELOG_SWIM
+//    #if SACT_TRACELOG_SWIM
     var traceLogLevel: Logger.Level? = .warning
-    #else
-    var traceLogLevel: Logger.Level?
-    #endif
+//    #else
+//    var traceLogLevel: Logger.Level?
+//    #endif
 }
 
 extension SWIM {

Sources/DistributedActors/DeathWatch.swift

@@ -27,14 +27,14 @@ import NIO
 // Implementation notes:
 // Care was taken to keep this implementation separate from the ActorCell however not require more storage space.
 @usableFromInline
-internal struct DeathWatch<Message> { // TODO: may want to change to a protocol
+internal struct DeathWatch<Message> {
     private var watching = Set<AddressableActorRef>()
     private var watchedBy = Set<AddressableActorRef>()
 
-    private var failureDetectorRef: NodeDeathWatcherShell.Ref
+    private var nodeDeathWatcher: NodeDeathWatcherShell.Ref
 
-    init(failureDetectorRef: NodeDeathWatcherShell.Ref) {
-        self.failureDetectorRef = failureDetectorRef
+    init(nodeDeathWatcher: NodeDeathWatcherShell.Ref) {
+        self.nodeDeathWatcher = nodeDeathWatcher
     }
 
     // MARK: perform watch/unwatch
@@ -160,7 +160,7 @@ internal struct DeathWatch<Message> { // TODO: may want to change to a protocol
 
     private func subscribeNodeTerminatedEvents(myself: ActorRef<Message>, node: UniqueNode?) {
         if let remoteNode = node {
-            self.failureDetectorRef.tell(.remoteActorWatched(watcher: AddressableActorRef(myself), remoteNode: remoteNode))
+            self.nodeDeathWatcher.tell(.remoteActorWatched(watcher: AddressableActorRef(myself), remoteNode: remoteNode))
         }
     }
 }