Dead letter when remote call target is dead (#989)

yim-lee · ktoso · web-flow · commit 364e3dd21d61 · 2022-07-07T13:53:42.000-07:00
Resolves #943 * Remove logging Co-authored-by: Konrad `ktoso` Malawski <konrad.malawski@project13.pl> * Remove unneeded import Co-authored-by: Konrad `ktoso` Malawski <konrad.malawski@project13.pl> * Rename variable Co-authored-by: Konrad `ktoso` Malawski <konrad.malawski@project13.pl>
diff --git a/Sources/DistributedActors/ClusterSystem.swift b/Sources/DistributedActors/ClusterSystem.swift
@@ -868,13 +868,12 @@ extension ClusterSystem {
             return nil
         }
 
-        return self.namingLock.withLock {
+        return try self.namingLock.withLock {
             guard let managed = self._managedDistributedActors.get(identifiedBy: id) else {
                 log.trace("Resolved as remote reference", metadata: [
                     "actor/id": "\(id)",
                 ])
-                // TODO(distributed): throw here, this should be a dead letter
-                return nil
+                throw DeadLetterError(recipient: id)
             }
 
             if let resolved = managed as? Act {
@@ -1116,11 +1115,6 @@ extension ClusterSystem {
             return
         }
 
-        guard let actor = self.resolve(id: recipient) else {
-            self.log.error("Unable to resolve recipient \(recipient). Message will be dropped: \(invocation)")
-            return
-        }
-
         Task {
             var decoder = ClusterInvocationDecoder(system: self, message: invocation)
 
@@ -1134,6 +1128,11 @@ extension ClusterSystem {
             )
 
             do {
+                guard let actor = self.resolve(id: recipient) else {
+                    self.deadLetters.tell(DeadLetter(invocation, recipient: recipient))
+                    throw DeadLetterError(recipient: recipient)
+                }
+
                 try await executeDistributedTarget(
                     on: actor,
                     target: target,
diff --git a/Sources/DistributedActors/DeadLetters.swift b/Sources/DistributedActors/DeadLetters.swift
@@ -12,6 +12,7 @@
 //
 //===----------------------------------------------------------------------===//
 
+import Distributed
 import Logging
 
 // ==== ----------------------------------------------------------------------------------------------------------------
@@ -255,3 +256,10 @@ extension ActorPath {
     static let _dead: ActorPath = try! ActorPath(root: "dead")
     static let _deadLetters: ActorPath = try! ActorPath._dead.appending("letters")
 }
+
+// ==== ----------------------------------------------------------------------------------------------------------------
+// MARK: Errors
+
+public struct DeadLetterError: DistributedActorSystemError, Codable {
+    public let recipient: ClusterSystem.ActorID
+}
diff --git a/Tests/DistributedActorsTests/ClusterSystemTests.swift b/Tests/DistributedActorsTests/ClusterSystemTests.swift
@@ -213,10 +213,10 @@ final class ClusterSystemTests: ClusterSystemXCTestCase {
         local.cluster.join(node: remote.cluster.uniqueNode)
 
         let greeter = Greeter(actorSystem: local)
-        let remoteGreeter = try Greeter.resolve(id: greeter.id, using: remote)
+        let remoteGreeterRef = try Greeter.resolve(id: greeter.id, using: remote)
 
         let value = try await shouldNotThrow {
-            try await remoteGreeter.hello()
+            try await remoteGreeterRef.hello()
         }
         value.shouldEqual("hello")
     }
@@ -233,10 +233,10 @@ final class ClusterSystemTests: ClusterSystemXCTestCase {
         local.cluster.join(node: remote.cluster.uniqueNode)
 
         let greeter = Greeter(actorSystem: local)
-        let remoteGreeter = try Greeter.resolve(id: greeter.id, using: remote)
+        let remoteGreeterRef = try Greeter.resolve(id: greeter.id, using: remote)
 
         let error = try await shouldThrow {
-            _ = try await remoteGreeter.helloThrow(codable: true)
+            _ = try await remoteGreeterRef.helloThrow(codable: true)
         }
         guard error is GreeterCodableError else {
             throw testKit.fail("Expected GreeterCodableError, got \(error)")
@@ -255,10 +255,10 @@ final class ClusterSystemTests: ClusterSystemXCTestCase {
         local.cluster.join(node: remote.cluster.uniqueNode)
 
         let greeter = Greeter(actorSystem: local)
-        let remoteGreeter = try Greeter.resolve(id: greeter.id, using: remote)
+        let remoteGreeterRef = try Greeter.resolve(id: greeter.id, using: remote)
 
         let error = try await shouldThrow {
-            _ = try await remoteGreeter.helloThrow(codable: false)
+            _ = try await remoteGreeterRef.helloThrow(codable: false)
         }
         guard let remoteCallError = error as? GenericRemoteCallError else {
             throw testKit.fail("Expected GenericRemoteCallError, got \(error)")
@@ -277,10 +277,10 @@ final class ClusterSystemTests: ClusterSystemXCTestCase {
         local.cluster.join(node: remote.cluster.uniqueNode)
 
         let greeter = Greeter(actorSystem: local)
-        let remoteGreeter = try Greeter.resolve(id: greeter.id, using: remote)
+        let remoteGreeterRef = try Greeter.resolve(id: greeter.id, using: remote)
 
         try await shouldNotThrow {
-            try await remoteGreeter.muted()
+            try await remoteGreeterRef.muted()
         }
     }
 
@@ -295,10 +295,10 @@ final class ClusterSystemTests: ClusterSystemXCTestCase {
         local.cluster.join(node: remote.cluster.uniqueNode)
 
         let greeter = Greeter(actorSystem: local)
-        let remoteGreeter = try Greeter.resolve(id: greeter.id, using: remote)
+        let remoteGreeterRef = try Greeter.resolve(id: greeter.id, using: remote)
 
         let error = try await shouldThrow {
-            try await remoteGreeter.mutedThrow(codable: true)
+            try await remoteGreeterRef.mutedThrow(codable: true)
         }
         guard error is GreeterCodableError else {
             throw testKit.fail("Expected GreeterCodableError, got \(error)")
@@ -316,10 +316,10 @@ final class ClusterSystemTests: ClusterSystemXCTestCase {
         local.cluster.join(node: remote.cluster.uniqueNode)
 
         let greeter = Greeter(actorSystem: local)
-        let remoteGreeter = try Greeter.resolve(id: greeter.id, using: remote)
+        let remoteGreeterRef = try Greeter.resolve(id: greeter.id, using: remote)
 
         let error = try await shouldThrow {
-            try await remoteGreeter.mutedThrow(codable: false)
+            try await remoteGreeterRef.mutedThrow(codable: false)
         }
         guard let remoteCallError = error as? GenericRemoteCallError else {
             throw testKit.fail("Expected GenericRemoteCallError, got \(error)")
@@ -337,11 +337,11 @@ final class ClusterSystemTests: ClusterSystemXCTestCase {
         local.cluster.join(node: remote.cluster.uniqueNode)
 
         let greeter = Greeter(actorSystem: local)
-        let remoteGreeter = try Greeter.resolve(id: greeter.id, using: remote)
+        let remoteGreeterRef = try Greeter.resolve(id: greeter.id, using: remote)
 
         let error = try await shouldThrow {
             try await RemoteCall.with(timeout: .milliseconds(200)) {
-                _ = try await remoteGreeter.hello(delayNanos: 3_000_000_000)
+                _ = try await remoteGreeterRef.hello(delayNanos: 3_000_000_000)
             }
         }
 
@@ -363,11 +363,11 @@ final class ClusterSystemTests: ClusterSystemXCTestCase {
         local.cluster.join(node: remote.cluster.uniqueNode)
 
         let greeter = Greeter(actorSystem: local)
-        let remoteGreeter = try Greeter.resolve(id: greeter.id, using: remote)
+        let remoteGreeterRef = try Greeter.resolve(id: greeter.id, using: remote)
 
         let error = try await shouldThrow {
             try await RemoteCall.with(timeout: .milliseconds(200)) {
-                try await remoteGreeter.muted(delayNanos: 3_000_000_000)
+                try await remoteGreeterRef.muted(delayNanos: 3_000_000_000)
             }
         }
 
@@ -381,7 +381,6 @@ final class ClusterSystemTests: ClusterSystemXCTestCase {
 }
 
 private distributed actor Greeter {
-    typealias ID = ClusterSystem.ActorID
     typealias ActorSystem = ClusterSystem
 
     distributed func hello() async throws -> String {
diff --git a/Tests/DistributedActorsTests/DeadLetterTests.swift b/Tests/DistributedActorsTests/DeadLetterTests.swift
@@ -12,6 +12,7 @@
 //
 //===----------------------------------------------------------------------===//
 
+import Distributed
 @testable import DistributedActors
 import DistributedActorsTestKit
 @testable import Logging
@@ -80,4 +81,82 @@ final class DeadLetterTests: ClusterSystemXCTestCase {
         try self.logCapture.awaitLogContaining(self.testKit, text: "This is a question")
         try self.logCapture.awaitLogContaining(self.testKit, text: "/user/ludwig")
     }
+
+    func test_remoteCallTerminatedTarget_shouldResultInDeadLetter() async throws {
+        let local = await setUpNode("local") { settings in
+            settings.enabled = true
+        }
+        let remote = await setUpNode("remote") { settings in
+            settings.enabled = true
+        }
+        local.cluster.join(node: remote.cluster.uniqueNode)
+
+        var greeter: Greeter? = Greeter(actorSystem: local)
+        let remoteGreeterRef = try Greeter.resolve(id: greeter!.id, using: remote)
+
+        let p = self.testKit.makeTestProbe(expecting: String.self)
+        let watcher = GreeterWatcher(probe: p, actorSystem: local)
+        try await watcher.watch(greeter!)
+
+        greeter = nil
+        try p.expectMessage(prefix: "Received terminated: /user/Greeter")
+
+        let error = try await shouldThrow {
+            _ = try await remoteGreeterRef.greet(name: "world")
+        }
+
+        guard error is DeadLetterError else {
+            throw self.testKit.fail("Expected DeadLetterError, got \(error)")
+        }
+
+        try self.capturedLogs(of: local).awaitLogContaining(self.testKit, text: "was not delivered to")
+    }
+
+    func test_resolveTerminatedTarget_shouldResultInDeadLetter() async throws {
+        var greeter: Greeter? = Greeter(actorSystem: self.system)
+        let greeterID = greeter!.id
+
+        let p = self.testKit.makeTestProbe(expecting: String.self)
+        let watcher = GreeterWatcher(probe: p, actorSystem: self.system)
+        try await watcher.watch(greeter!)
+
+        greeter = nil
+        try p.expectMessage(prefix: "Received terminated: /user/Greeter")
+
+        let error = try shouldThrow {
+            _ = try self.system.resolve(id: greeterID, as: Greeter.self)
+        }
+
+        guard error is DeadLetterError else {
+            throw self.testKit.fail("Expected DeadLetterError, got \(error)")
+        }
+    }
+}
+
+private distributed actor Greeter {
+    typealias ActorSystem = ClusterSystem
+
+    distributed func greet(name: String) -> String {
+        "hello \(name)!"
+    }
+}
+
+private distributed actor GreeterWatcher: LifecycleWatch {
+    typealias ActorSystem = ClusterSystem
+
+    let probe: ActorTestProbe<String>
+
+    init(probe: ActorTestProbe<String>, actorSystem: ActorSystem) {
+        self.actorSystem = actorSystem
+        self.probe = probe
+    }
+
+    distributed func watch(_ greeter: Greeter) {
+        watchTermination(of: greeter)
+    }
+
+    // FIXME(distributed): Should not need to be distributed: https://github.com/apple/swift/pull/59397
+    public distributed func terminated(actor id: ActorID) async { // not REALLY distributed...
+        self.probe.tell("Received terminated: \(id)")
+    }
 }
