[Receptionist] Fix ordering bug and improve diagnostics

Author: ktoso

Sources/DistributedActors/Cluster/Reception/OperationLogDistributedReceptionist.swift

@@ -344,28 +344,24 @@ extension OpLogDistributedReceptionist: LifecycleWatch {
     }
 
     public nonisolated func listing<Guest>(
-        of key: DistributedReception.Key<Guest>
+        of key: DistributedReception.Key<Guest>,
+        file: String = #fileID, line: UInt = #line
     ) async -> DistributedReception.GuestListing<Guest>
         where Guest: DistributedActor, Guest.ActorSystem == ClusterSystem
     {
-        let res = await self.whenLocal { _ in
-            DistributedReception.GuestListing<Guest>(receptionist: self, key: key)
-        }
-
-        guard let r = res else {
-            return .init(receptionist: self, key: key)
-        }
-
-        return r
+        return DistributedReception.GuestListing<Guest>(receptionist: self, key: key, file: file, line: line)
     }
 
+    // 'local' impl for 'listing'
     func _listing(
-        subscription: AnyDistributedReceptionListingSubscription
+        subscription: AnyDistributedReceptionListingSubscription,
+        file: String = #fileID, line: UInt = #line
     ) {
         if self.storage.addSubscription(key: subscription.key, subscription: subscription) {
             // self.instrumentation.actorSubscribed(key: anyKey, id: self.id._unwrapActorID) // FIXME: remove the address parameter, it does not make sense anymore
-            self.log.trace("Subscribed async sequence to \(subscription.key) actors", metadata: [
+            self.log.trace("Subscribed async sequence to \(subscription.key)", metadata: [
                 "subscription/key": "\(subscription.key)",
+                "subscription/callSite": "\(file):\(line)",
             ])
         }
     }
@@ -705,7 +701,18 @@ extension OpLogDistributedReceptionist {
 
     /// Receive an Ack and potentially continue streaming ops to peer if still pending operations available.
     distributed func ackOps(until: UInt64, by peer: ReceptionistRef) {
-        guard var replayer = self.peerReceptionistReplayers[peer] else {
+        var replayer = self.peerReceptionistReplayers[peer]
+
+        if replayer == nil, until == 0 {
+            self.log.debug("Received message from \(peer), but no replayer available, create one ad-hoc now", metadata: [
+                "peer": "\(peer.id.uniqueNode)",
+            ])
+            // TODO: Generally we should trigger a `onNewClusterMember` but seems we got a message before that triggered
+            // Seems ordering became less strict here with DA unfortunately...?
+            replayer = self.ops.replay(from: .beginning)
+        }
+
+        guard var replayer = replayer else {
             self.log.trace("Received a confirmation until \(until) from \(peer) but no replayer available for it, ignoring", metadata: [
                 "receptionist/peer/confirmed": "\(until)",
                 "receptionist/peer": "\(peer.id)",

Sources/DistributedActors/Receptionist/DistributedReceptionist.swift

@@ -43,7 +43,7 @@ public protocol DistributedReceptionist: DistributedActor {
     ///
     /// It emits both values for already existing, checked-in before the listing was created,
     /// actors; as well as new actors which are checked-in while the listing was already subscribed to.
-    func listing<Guest>(of key: DistributedReception.Key<Guest>) async -> DistributedReception.GuestListing<Guest>
+    func listing<Guest>(of key: DistributedReception.Key<Guest>, file: String, line: UInt) async -> DistributedReception.GuestListing<Guest>
         where Guest: DistributedActor, Guest.ActorSystem == ClusterSystem
 
     /// Perform a *single* lookup for a distributed actor identified by the passed in `key`.
@@ -54,26 +54,49 @@ public protocol DistributedReceptionist: DistributedActor {
         where Guest: DistributedActor, Guest.ActorSystem == ClusterSystem
 }
 
+extension DistributedReceptionist {
+    /// Returns a "listing" asynchronous sequence which will emit actor references,
+    /// for every distributed actor that the receptionist discovers for the specific key.
+    ///
+    /// It emits both values for already existing, checked-in before the listing was created,
+    /// actors; as well as new actors which are checked-in while the listing was already subscribed to.
+    func listing<Guest>(of key: DistributedReception.Key<Guest>, file: String = #file, line: UInt = #line) async -> DistributedReception.GuestListing<Guest>
+        where Guest: DistributedActor, Guest.ActorSystem == ClusterSystem
+    {
+        await self.listing(of: key, file: file, line: line)
+    }
+}
+
 extension DistributedReception {
     public struct GuestListing<Guest: DistributedActor>: AsyncSequence, Sendable where Guest.ActorSystem == ClusterSystem {
         public typealias Element = Guest
 
         let receptionist: OpLogDistributedReceptionist
         let key: DistributedReception.Key<Guest>
 
-        init(receptionist: OpLogDistributedReceptionist, key: DistributedReception.Key<Guest>) {
+        // Location where the subscription was created
+        let file: String
+        let line: UInt
+
+        init(receptionist: OpLogDistributedReceptionist, key: DistributedReception.Key<Guest>,
+             file: String, line: UInt)
+        {
             self.receptionist = receptionist
             self.key = key
+            self.file = file
+            self.line = line
         }
 
         public func makeAsyncIterator() -> AsyncIterator {
-            AsyncIterator(receptionist: self.receptionist, key: self.key)
+            AsyncIterator(receptionist: self.receptionist, key: self.key, file: self.file, line: self.line)
         }
 
         public class AsyncIterator: AsyncIteratorProtocol {
             var underlying: AsyncStream<Element>.Iterator!
 
-            init(receptionist __secretlyKnownToBeLocal: OpLogDistributedReceptionist, key: DistributedReception.Key<Guest>) {
+            init(receptionist __secretlyKnownToBeLocal: OpLogDistributedReceptionist, key: DistributedReception.Key<Guest>,
+                 file: String, line: UInt)
+            {
                 self.underlying = AsyncStream<Element> { continuation in
                     let anySubscribe = AnyDistributedReceptionListingSubscription(
                         subscriptionID: ObjectIdentifier(self),
@@ -89,7 +112,7 @@ extension DistributedReception {
                     )
 
                     Task {
-                        await __secretlyKnownToBeLocal._listing(subscription: anySubscribe)
+                        await __secretlyKnownToBeLocal._listing(subscription: anySubscribe, file: file, line: line)
                     }
 
                     continuation.onTermination = { @Sendable termination in

Tests/DistributedActorsTests/Cluster/Reception/OpLogDistributedReceptionistClusteredTests.swift

@@ -85,28 +85,29 @@ final class OpLogDistributedReceptionistClusteredTests: ClusteredActorSystemsXCT
     // MARK: Sync
 
     func test_shouldReplicateRegistrations() async throws {
-        let (local, remote) = await self.setUpPair()
-        let testKit: ActorTestKit = self.testKit(local)
-        try await self.joinNodes(node: local, with: remote)
+        let (first, second) = await self.setUpPair()
+        let testKit = self.testKit(first)
+        try await self.joinNodes(node: first, with: second)
 
         let probe = testKit.makeTestProbe(expecting: String.self)
 
-        // Create forwarder on 'local'
-        let forwarder = StringForwarder(probe: probe, actorSystem: local)
+        // Create forwarder on 'first'
+        let forwarder = StringForwarder(probe: probe, actorSystem: first)
 
         // subscribe on `remote`
         let subscriberProbe = testKit.makeTestProbe("subscriber", expecting: StringForwarder.self)
         let subscriptionTask = Task {
-            for try await forwarder in await remote.receptionist.listing(of: .stringForwarders) {
+            for try await forwarder in await second.receptionist.listing(of: .stringForwarders) {
                 subscriberProbe.tell(forwarder)
             }
         }
         defer {
             subscriptionTask.cancel()
         }
 
-        // checkIn on `local`
-        await local.receptionist.checkIn(forwarder, with: .stringForwarders)
+        // checkIn on `first`
+        await first.receptionist.checkIn(forwarder, with: .stringForwarders)
+        first.log.notice("Checked in: \(forwarder)")
 
         try await Task {
             let found = try subscriberProbe.expectMessage()