Multicam seams: source peer on inbound messages, per-peer frame acks

RemoteCam/MultipeerService.swift

@@ -12,7 +12,10 @@ import Combine
 import UIKit
 
 protocol MultipeerServiceDelegate: AnyObject {
-    func didReceiveMessage(_ message: Message)
+    /// `peer` is the message's source. The 1:1 SessionCoordinator can ignore
+    /// it (there is only one peer a message can come from); a multicam
+    /// director needs it to route responses to the right camera.
+    func didReceiveMessage(_ message: Message, from peer: MCPeerID)
     func didReceiveFrameRequest(_ request: RemoteCmd.RequestFrame)
     func didReceiveFrame(_ frame: RemoteCmd.SendFrame, from peer: MCPeerID)
     func peerDidConnect(_ peer: MCPeerID)
@@ -205,7 +208,7 @@ class MultipeerService: NSObject, MCSessionDelegate,
         case let frame as RemoteCmd.SendFrame:
             delegate?.didReceiveFrame(frame, from: peerID)
         default:
-            delegate?.didReceiveMessage(inboundMessage)
+            delegate?.didReceiveMessage(inboundMessage, from: peerID)
         }
     }
 

RemoteCam/SessionCoordinator.swift

@@ -302,23 +302,27 @@ public actor SessionCoordinator {
     }
 
     @discardableResult
-    func sendMessage(_ msg: Message, mode: MCSessionSendDataMode = .reliable) -> Bool {
+    func sendMessage(_ msg: Message, to peers: [MCPeerID]? = nil,
+                     mode: MCSessionSendDataMode = .reliable) -> Bool {
         guard let multipeerService else {
             // Watch Remote mode never starts a multipeer session.
             return false
         }
-        return multipeerService.send(msg, to: connectedPeers, mode: mode)
+        return multipeerService.send(msg, to: peers ?? connectedPeers, mode: mode)
     }
 
     /// Send, or pop to scanning with a connection-error alert on failure —
-    /// the old `sendCommandOrGoToScanning`.
-    func sendOrGoToScanning(_ msg: Message, mode: MCSessionSendDataMode = .reliable) async {
+    /// the old `sendCommandOrGoToScanning`. `peers` nil = all connected
+    /// (identical for this 1:1 coordinator; the param exists so per-peer
+    /// sends like frame acks address only their source).
+    func sendOrGoToScanning(_ msg: Message, to peers: [MCPeerID]? = nil,
+                            mode: MCSessionSendDataMode = .reliable) async {
         guard multipeerService != nil else {
             // Watch Remote mode: there is no peer and no scanning state to fall back to.
             debugLog("sendOrGoToScanning: no multipeer session, dropping \(type(of: msg))")
             return
         }
-        if !sendMessage(msg, mode: mode) {
+        if !sendMessage(msg, to: peers, mode: mode) {
             await popToScanning()
             let presenter = alertPresenter
             OperationQueue.main.addOperation {
@@ -542,6 +546,14 @@ public actor SessionCoordinator {
         await sendOrGoToScanning(RemoteCmd.RequestFrame(sender: nil))
     }
 
+    /// Ack the camera that sent this frame so only its credit window
+    /// advances. With one connected peer this is identical to the broadcast
+    /// form; with several cameras a broadcast ack would let every camera
+    /// send on one camera's consumed frame.
+    private func requestFrame(acking frame: RemoteCmd.OnFrame) async {
+        await sendOrGoToScanning(RemoteCmd.RequestFrame(sender: nil), to: [frame.peerId])
+    }
+
     /// Pop to scanning (stops at the lobby floor like the old machine) and
     /// restart discovery via `.scanning`'s entry behavior.
     func popToScanning() async {
@@ -1590,7 +1602,7 @@ public actor SessionCoordinator {
         case let frame as RemoteCmd.OnFrame:
             noteMonitorFrame(frame)
             monitor?.show(frame: frame)
-            await requestFrame()
+            await requestFrame(acking: frame)
 
         case is UICmd.StreamStalled:
             await requestFrame()
@@ -1893,7 +1905,7 @@ public actor SessionCoordinator {
         case let frame as RemoteCmd.OnFrame:
             noteMonitorFrame(frame)
             monitor?.show(frame: frame)
-            await requestFrame()
+            await requestFrame(acking: frame)
 
         case is UICmd.StreamStalled:
             await requestFrame()
@@ -2426,7 +2438,10 @@ public actor SessionCoordinator {
 
 extension SessionCoordinator: MultipeerServiceDelegate {
 
-    public nonisolated func didReceiveMessage(_ message: Message) {
+    public nonisolated func didReceiveMessage(_ message: Message, from peer: MCPeerID) {
+        // `peer` is deliberately unused: this coordinator links exactly one
+        // peer, so every message's source is unambiguous. The multicam
+        // director's controller is the consumer that routes by it.
         tell(UICmd.PeerTrafficObserved())
         tell(message)
     }

RemoteCamTests/LoopbackSessionTests.swift

@@ -82,7 +82,7 @@ class LoopbackMultipeerService: MultipeerServiceProtocol {
         case let frame as RemoteCmd.SendFrame:
             remoteDelegate.didReceiveFrame(frame, from: localPeerID)
         default:
-            remoteDelegate.didReceiveMessage(decoded)
+            remoteDelegate.didReceiveMessage(decoded, from: localPeerID)
         }
         return true
     }

RemoteCamTests/RemoteCamSessionTests.swift

@@ -260,6 +260,33 @@ class SessionCoordinatorTests: XCTestCase {
         XCTAssertEqual(frameRequests[0].mode, .reliable)
     }
 
+    /// Seam B: a frame's ack goes only to the camera that sent it. With two
+    /// peers connected, a broadcast ack would advance the credit window of a
+    /// camera whose frame was never consumed.
+    func testFrameAckTargetsOnlyTheSendingPeer() async {
+        let secondCamera = MCPeerID(displayName: "SecondCamera")
+        await enterMonitor(.Photo)
+        harness.fakeMP.connectedPeers.append(secondCamera)
+        harness.fakeMP.sendResult = true
+
+        await harness.deliver(RemoteCmd.OnFrame(
+            data: Data([1, 2, 3]), sender: nil, peerId: secondCamera,
+            fps: 30, camPosition: .back, camOrientation: .portrait,
+            codec: .jpeg, sequenceNumber: 1))
+
+        let acks = sent(RemoteCmd.RequestFrame.self)
+        XCTAssertEqual(acks.count, 1)
+        XCTAssertEqual(acks[0].peers, [secondCamera],
+                       "ack must address the frame's source, not all connected peers")
+
+        harness.fakeMP.sentMessages.removeAll()
+        await harness.deliver(RemoteCmd.OnFrame(
+            data: Data([4, 5, 6]), sender: nil, peerId: harness.peer,
+            fps: 30, camPosition: .back, camOrientation: .portrait,
+            codec: .jpeg, sequenceNumber: 2))
+        XCTAssertEqual(sent(RemoteCmd.RequestFrame.self).map(\.peers), [[harness.peer]])
+    }
+
     func testMonitorPhotoModeUnbecomeMonitorPopsToConnected() async {
         await enterMonitor(.Photo)
         await harness.deliver(UICmd.UnbecomeMonitor(sender: nil))

RemoteCamTests/StormoLoopbackTests.swift

@@ -57,7 +57,7 @@ final class StormoLoopbackTests: XCTestCase {
 
         func peerDidConnect(_ peer: PeerID) { connected.fulfill() }
 
-        func didReceiveMessage(_ message: Message) {
+        func didReceiveMessage(_ message: Message, from peer: PeerID) {
             lock.lock(); _receivedMessages.append(message); lock.unlock()
             messageReceived.fulfill()
         }