Down replaced cluster members (#867)

* Make sure replaced cluster members are .down

* Add integration test for ungraceful shutdown

* Remove dead code from integ test

* Remove more dead code

* Update MembershipTests for changed replacement semantics

* remove whitespace that made it in

* fix typo in test

* Fix issues in integration test

* Fix formatting

Co-authored-by: Konrad `ktoso` Malawski <[email protected]>

Author: budde

IntegrationTests/tests_04_cluster/it_Clustered_swim_ungraceful_shutdown/main.swift

@@ -0,0 +1,54 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Distributed Actors open source project
+//
+// Copyright (c) 2022 Apple Inc. and the Swift Distributed Actors project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.md for the list of Swift Distributed Actors project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import DistributedActors
+
+var args = CommandLine.arguments
+args.removeFirst()
+
+guard args.count >= 2, let bindPort = Int(args[1]) else {
+    fatalError("Node name and bind port must be provided")
+}
+
+let nodeName = args[0]
+
+print("Binding to port \(bindPort)")
+
+let system = ActorSystem(nodeName) { settings in
+    settings.logging.logLevel = .info
+
+    settings.cluster.enabled = true
+    settings.cluster.bindPort = bindPort
+
+    settings.cluster.swim.probeInterval = .milliseconds(300)
+    settings.cluster.swim.pingTimeout = .milliseconds(100)
+}
+
+let ref = try system.spawn(
+    "streamWatcher",
+    of: Cluster.Event.self,
+    .receive { context, event in
+        context.log.info("Event: \(event)")
+        return .same
+    }
+)
+system.cluster.events.subscribe(ref)
+
+if args.count >= 4, let joinPort = Int(args[3]) {
+    let joinHost = args[2]
+    print("Joining node \(joinHost):\(joinPort)")
+    system.cluster.join(host: joinHost, port: joinPort)
+}
+
+Thread.sleep(.seconds(120))

IntegrationTests/tests_04_cluster/test_02_ungraceful_shutdown_recovery.sh

@@ -0,0 +1,75 @@
+#!/bin/bash
+##===----------------------------------------------------------------------===##
+##
+## This source file is part of the Swift Distributed Actors open source project
+##
+## Copyright (c) 2022 Apple Inc. and the Swift Distributed Actors project authors
+## Licensed under Apache License v2.0
+##
+## See LICENSE.txt for license information
+## See CONTRIBUTORS.md for the list of Swift Distributed Actors project authors
+##
+## SPDX-License-Identifier: Apache-2.0
+##
+##===----------------------------------------------------------------------===##
+
+set -e
+#set -x # verbose
+
+declare -r my_path="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+declare -r root_path="$my_path/.."
+
+declare -r app_name='it_Clustered_swim_ungraceful_shutdown'
+
+source ${my_path}/shared.sh
+
+declare -r first_logs=/tmp/sact_first.out
+declare -r second_logs=/tmp/sact_second.out
+declare -r killed_logs=/tmp/sact_killed.out
+declare -r replacement_logs=/tmp/sact_replacement.out
+rm -f ${first_logs}
+rm -f ${second_logs}
+rm -f ${killed_logs}
+rm -f ${replacement_logs}
+
+stdbuf -i0 -o0 -e0 swift run it_Clustered_swim_ungraceful_shutdown Frist 7337 > ${first_logs} 2>&1 &
+declare -r first_pid=$(echo $!)
+wait_log_exists ${first_logs} 'Binding to: ' 200 # since it might be compiling again...
+
+stdbuf -i0 -o0 -e0 swift run it_Clustered_swim_ungraceful_shutdown Second 8228 127.0.0.1 7337 > ${second_logs} 2>&1 &
+declare -r second_pid=$(echo $!)
+wait_log_exists ${second_logs} 'Binding to: ' 200 # since it might be compiling again...
+
+stdbuf -i0 -o0 -e0 swift run it_Clustered_swim_ungraceful_shutdown Killed 9119 127.0.0.1 7337 > ${killed_logs} 2>&1 &
+declare -r killed_pid=$(echo $!)
+wait_log_exists ${killed_logs} 'Binding to: ' 200 # since it might be compiling again...
+
+echo "Waiting nodes to become .up..."
+wait_log_exists ${first_logs} 'Event: membershipChange(sact://[email protected]:8228 :: \[joining\] -> \[     up\])' 50
+wait_log_exists ${first_logs} 'Event: membershipChange(sact://[email protected]:9119 :: \[joining\] -> \[     up\])' 50
+echo 'Other two members seen .up, good...'
+
+sleep 1
+
+# SIGKILL the third member, causing ungraceful shutdown
+echo "Killing PID ${killed_pid}"
+kill -9 ${killed_pid}
+
+# Immediately restart the third process
+stdbuf -i0 -o0 -e0 swift run it_Clustered_swim_ungraceful_shutdown Replacement 9119 127.0.0.1 7337 >> ${replacement_logs} 2>&1 &
+declare -r replacement_pid=$(echo $!)
+wait_log_exists ${replacement_logs} 'Binding to: ' 200 # just to be safe...
+
+# The original third node should go .down while the replacement becomes .up
+wait_log_exists ${first_logs} 'Event: membershipChange(sact://[email protected]:9119 :: \[     up\] -> \[   down\])' 50
+echo 'Killed member .down, good...'
+wait_log_exists ${first_logs} 'Event: membershipChange(sact://[email protected]:9119 :: \[joining\] -> \[     up\])' 50
+echo 'Replacement member .up, good...'
+
+# === cleanup ----------------------------------------------------------------------------------------------------------
+
+kill -9 ${first_pid}
+kill -9 ${second_pid}
+kill -9 ${replacement_pid}
+
+_killall ${app_name}

Package.swift

@@ -197,6 +197,13 @@ var targets: [PackageDescription.Target] = [
         ],
         path: "IntegrationTests/tests_04_cluster/it_Clustered_swim_suspension_reachability"
     ),
+    .target(
+        name: "it_Clustered_swim_ungraceful_shutdown",
+        dependencies: [
+            "DistributedActors",
+        ],
+        path: "IntegrationTests/tests_04_cluster/it_Clustered_swim_ungraceful_shutdown"
+    ),
 
     // ==== ----------------------------------------------------------------------------------------------------------------
     // MARK: Performance / Benchmarks

Sources/DistributedActors/Cluster/Cluster+Membership.swift

@@ -286,7 +286,11 @@ extension Cluster.Membership {
 
         if let previousMember = self.member(change.node.node) {
             // we are joining "over" an existing incarnation of a node; causing the existing node to become .down immediately
-            _ = self.removeCompletely(previousMember.uniqueNode) // the replacement event will handle the down notifications
+            if previousMember.status < .down {
+                _ = self.mark(previousMember.uniqueNode, as: .down)
+            } else {
+                _ = self.removeCompletely(previousMember.uniqueNode) // the replacement event will handle the down notifications
+            }
             self._members[change.node] = change.member
 
             // emit a replacement membership change, this will cause down cluster events for previous member

Tests/DistributedActorsTests/MembershipTests.swift

@@ -133,7 +133,8 @@ final class MembershipTests: XCTestCase {
             Cluster.MembershipChange(member: self.memberB, toStatus: .down)
         )
 
-        membership.members(atLeast: .joining).count.shouldEqual(2)
+        membership.members(atLeast: .joining).count.shouldEqual(3)
+        membership.members(atLeast: .down).count.shouldEqual(1)
         let memberNode = membership.uniqueMember(change.member.uniqueNode)
         memberNode?.status.shouldEqual(Cluster.MemberStatus.up)
     }
@@ -349,7 +350,7 @@ final class MembershipTests: XCTestCase {
     // ==== ----------------------------------------------------------------------------------------------------------------
     // MARK: Replacements
 
-    func test_join_overAnExistingMode_replacement() {
+    func test_join_overAnExistingNode_replacement() {
         var membership = self.initialMembership
         let secondReplacement = Cluster.Member(node: UniqueNode(node: self.nodeB.node, nid: .random()), status: .joining)
         let change = membership.join(secondReplacement.uniqueNode)!
@@ -359,9 +360,9 @@ final class MembershipTests: XCTestCase {
         var secondDown = self.memberB
         secondDown.status = .down
 
-        members.count.shouldEqual(3)
+        members.count.shouldEqual(4)
         members.shouldContain(secondReplacement)
-        members.shouldNotContain(self.memberB) // was replaced
+        members.shouldContain(secondDown) // replaced node should be .down
     }
 
     func test_mark_replacement() throws {