MSRP: propagate leave to talker only if complementary attribute unregistered

When propagating listener leave back to talker port after talker
deregistration, check if the complementary talker attribute type
is still registered (e.g., if talkerAdvertise exists when processing
talkerFailed leave). Only propagate if no other talker attribute is
registered, avoiding race conditions during talker attribute type
transitions.

Author: lhoward

Sources/MRP/Applications/MSRP/MSRPApplication.swift

@@ -787,7 +787,8 @@ extension MSRPApplication {
 
       if let existingTalkerRegistration = await _findTalkerRegistration(
         for: streamID,
-        participant: participant
+        participant: participant,
+        attributeType: .talkerAdvertise
       ), existingTalkerRegistration.dataFrameParameters != dataFrameParameters {
         _logger
           .error(
@@ -1110,32 +1111,42 @@ extension MSRPApplication {
 
   private func _findTalkerRegistration(
     for streamID: MSRPStreamID,
-    participant: Participant<MSRPApplication>
+    participant: Participant<MSRPApplication>,
+    attributeType: MSRPAttributeType?
   ) async -> (any MSRPTalkerValue)? {
-    if let value = await participant.findAttribute(
-      attributeType: MSRPAttributeType.talkerAdvertise.rawValue,
-      matching: .matchAnyIndex(streamID.index)
-    ) {
+    if attributeType == nil || attributeType == .talkerAdvertise,
+       let value = await participant.findAttribute(
+         attributeType: MSRPAttributeType.talkerAdvertise.rawValue,
+         matching: .matchAnyIndex(streamID.index)
+       )
+    {
       value.1 as? (any MSRPTalkerValue)
-    } else if let value = await participant.findAttribute(
-      attributeType: MSRPAttributeType.talkerFailed.rawValue,
-      matching: .matchAnyIndex(streamID.index)
-    ) {
+    } else if attributeType == nil || attributeType == .talkerFailed,
+              let value = await participant.findAttribute(
+                attributeType: MSRPAttributeType.talkerFailed.rawValue,
+                matching: .matchAnyIndex(streamID.index)
+              )
+    {
       value.1 as? (any MSRPTalkerValue)
     } else {
       nil
     }
   }
 
   private func _findTalkerRegistration(
-    for streamID: MSRPStreamID
+    for streamID: MSRPStreamID,
+    attributeType: MSRPAttributeType? = nil
   ) async throws -> TalkerRegistration? {
     var talkerRegistration: TalkerRegistration?
 
+    // note: it's safe to enumerate all participants, because we only expect so
+    // see a talker for a given stream ID registered on a single port
+    // (otherwise, it is an error)
     await apply { participant in
       guard let participantTalker = await _findTalkerRegistration(
         for: streamID,
-        participant: participant
+        participant: participant,
+        attributeType: attributeType
       ) else {
         return
       }
@@ -1550,14 +1561,25 @@ extension MSRPApplication {
     // behavior described in 35.2.4.4.1.
     guard let talkerParticipant = try? findParticipant(port: port) else { return }
 
+    let otherTalkerRegistration =
+      await _findTalkerRegistration(
+        for: streamID,
+        participant: talkerParticipant,
+        attributeType: talkerValue.declarationType!.talkerComplement
+      )
+
     try await apply { participant in
       guard participant.port != port else { return } // don't propagate to source port
 
-      // If this participant has active listeners, propagate a leave back to the talker
-      if let listenerRegistration = await _findListenerRegistration(
-        for: streamID,
-        participant: participant
-      ) {
+      // if a different talker attribute is registered on the talker port,
+      // don't propagate the listener leave back to the talker to avoid a race
+      // whilst deregistering
+      if otherTalkerRegistration == nil,
+         let listenerRegistration = await _findListenerRegistration(
+           for: streamID,
+           participant: participant
+         )
+      {
         try await talkerParticipant.leave(
           attributeType: MSRPAttributeType.listener.rawValue,
           attributeSubtype: listenerRegistration.1.rawValue,