Make ActorLeakTests green again

Author: yim-lee

Sources/DistributedActors/Cluster/ClusterEventStream.swift

@@ -25,8 +25,15 @@ public struct ClusterEventStream: AsyncSequence {
 
     private let shell: ClusterEventStreamShell?
 
-    internal init(_ system: ClusterSystem) {
-        self.shell = ClusterEventStreamShell(actorSystem: system)
+    internal init(_ system: ClusterSystem, customName: String? = nil) async {
+        var props = ClusterEventStreamShell.props
+        if let customName = customName {
+            props._knownActorName = customName
+        }
+
+        self.shell = await _Props.$forSpawn.withValue(props) {
+            ClusterEventStreamShell(actorSystem: system)
+        }
     }
 
     // For testing only
@@ -108,6 +115,14 @@ public struct ClusterEventStream: AsyncSequence {
 internal distributed actor ClusterEventStreamShell: LifecycleWatch {
     typealias ActorSystem = ClusterSystem
 
+    static var props: _Props {
+        var ps = _Props()
+        ps._knownActorName = "clustEventStream"
+        ps._systemActor = true
+        ps._wellKnown = true
+        return ps
+    }
+
     // We maintain a snapshot i.e. the "latest version of the membership",
     // in order to eagerly publish it to anyone who subscribes immediately,
     // followed by joining them to the subsequent ``Cluster/Event`` publishes.

Sources/DistributedActors/ClusterSystem.swift

@@ -305,7 +305,7 @@ public class ClusterSystem: DistributedActorSystem, @unchecked Sendable {
         }
 
         if !settings.enabled {
-            let clusterEvents = ClusterEventStream(self)
+            let clusterEvents = await ClusterEventStream(self)
             _ = self._clusterStore.storeIfNilThenLoad(Box(nil))
             _ = self._clusterControlStore.storeIfNilThenLoad(Box(ClusterControl(settings, cluster: nil, clusterRef: self.deadLetters.adapted(), eventStream: clusterEvents)))
         }
@@ -317,7 +317,7 @@ public class ClusterSystem: DistributedActorSystem, @unchecked Sendable {
         if let cluster = self._cluster {
             // try!-safe, this will spawn under /system/... which we have full control over,
             // and there /system namespace and it is known there will be no conflict for this name
-            let clusterEvents = ClusterEventStream(self)
+            let clusterEvents = await ClusterEventStream(self)
             let clusterRef = try! cluster.start(system: self, clusterEvents: clusterEvents) // only spawns when cluster is initialized
             _ = self._clusterControlStore.storeIfNilThenLoad(Box(ClusterControl(settings, cluster: cluster, clusterRef: clusterRef, eventStream: clusterEvents)))
 

Tests/DistributedActorsTests/Cluster/ClusterEventStreamTests.swift

@@ -35,7 +35,7 @@ final class ClusterEventStreamTests: ClusterSystemXCTestCase, @unchecked Sendabl
         let p1 = self.testKit.makeTestProbe(expecting: Cluster.Event.self)
         let p2 = self.testKit.makeTestProbe(expecting: Cluster.Event.self)
 
-        let eventStream = ClusterEventStream(system)
+        let eventStream = await ClusterEventStream(system, customName: "testClusterEvents")
 
         await eventStream.subscribe(p1.ref) // sub before first -> up was published
         await eventStream.publish(.membershipChange(.init(member: self.memberA, toStatus: .up)))
@@ -83,7 +83,7 @@ final class ClusterEventStreamTests: ClusterSystemXCTestCase, @unchecked Sendabl
     func test_clusterEventStream_collapseManyEventsIntoSnapshot() async throws {
         let p1 = self.testKit.makeTestProbe(expecting: Cluster.Event.self)
 
-        let eventStream = ClusterEventStream(system)
+        let eventStream = await ClusterEventStream(system, customName: "testClusterEvents")
 
         await eventStream.publish(.membershipChange(.init(member: self.memberA, toStatus: .joining)))
         await eventStream.publish(.membershipChange(.init(member: self.memberA, toStatus: .up)))
@@ -106,7 +106,7 @@ final class ClusterEventStreamTests: ClusterSystemXCTestCase, @unchecked Sendabl
     }
 
     func test_clusterEventStream_collapseManyEventsIntoSnapshot_async() async throws {
-        let eventStream = ClusterEventStream(system)
+        let eventStream = await ClusterEventStream(system, customName: "testClusterEvents")
 
         // Publish events to change membership
         await eventStream.publish(.membershipChange(.init(member: self.memberA, toStatus: .joining)))