fix: keep publish consistent with a reconnect

CHANGELOG.md

@@ -248,6 +248,14 @@ on `Call` (`get_call_participant_session_metrics`,
   `get_stats()` samples only in the `getstats` records of `rtc_stats`. Before,
   it also sent them as JSON strings in `subscriber_stats` and
   `publisher_stats`.
+- A `publish` that waits for another media operation uses the connection that
+  is current at the end of the wait. Before, it could use the session that a
+  reconnect replaced. After a rejoin, this `publish` failed although the call
+  was connected again.
+- A reconnect restores each publication with one sender. This includes a track
+  that `publish` adds to the new connection before the restore. Before, that
+  track got a second m-line. The SFU rejected the negotiation, and the SDK
+  started one more rejoin.
 
 # v0.1.0-preview.2
 

src/rtc/join/publish.rs

@@ -14,6 +14,8 @@ impl RtcCore {
                 "a video track cannot be published as {track_type:?}"
             )));
         }
+        // A reconnect can replace the connection while `publish` waits for the lock.
+        let mut media = self.media.lock().await;
         let Some((publisher, signal, session_id, publish_options)) = self.publisher_handles().await
         else {
             return Err(RtcError::IllegalState("publish() before join()".to_owned()));
@@ -27,7 +29,6 @@ impl RtcCore {
         {
             return Err(RtcError::PermissionDenied { capability });
         }
-        let mut media = self.media.lock().await;
         let track_id = track.track_id();
         let status = media
             .position(&track_id)

src/rtc/join/reconnect_runtime.rs

@@ -35,6 +35,10 @@ impl RtcCore {
         if !tracks.is_empty() {
             let mut publisher_rtcp_tasks = Vec::new();
             for track in &tracks {
+                // A `publish` after the connection swap can add this sender first.
+                if has_sender_for_track(&publisher, &track.track_id()).await {
+                    continue;
+                }
                 match publisher::add_transceiver_for_track(&publisher, track, &publish_options)
                     .await
                 {
@@ -964,3 +968,18 @@ impl RtcCore {
         Some(reporter.report_once().await)
     }
 }
+
+async fn has_sender_for_track(publisher: &Arc<RTCPeerConnection>, track_id: &str) -> bool {
+    for transceiver in publisher.get_transceivers().await {
+        if transceiver
+            .sender()
+            .await
+            .track()
+            .await
+            .is_some_and(|bound| bound.id() == track_id)
+        {
+            return true;
+        }
+    }
+    false
+}

src/rtc/join/tests.rs

@@ -249,6 +249,71 @@ async fn establish_fake(
     (connection, sfu)
 }
 
+/// A local SFU Twirp endpoint that sends the path of each request to the
+/// channel and answers it with an empty 200.
+async fn recording_twirp_server() -> (String, tokio::sync::mpsc::UnboundedReceiver<String>) {
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind Twirp server");
+    let address = listener.local_addr().expect("Twirp server address");
+    let (paths, received) = tokio::sync::mpsc::unbounded_channel();
+    tokio::spawn(async move {
+        while let Ok((mut stream, _)) = listener.accept().await {
+            let paths = paths.clone();
+            tokio::spawn(async move {
+                let mut head = Vec::new();
+                let mut buffer = [0_u8; 4096];
+                while !head.windows(4).any(|window| window == b"\r\n\r\n") {
+                    match stream.read(&mut buffer).await {
+                        Ok(0) | Err(_) => return,
+                        Ok(read) => head.extend_from_slice(&buffer[..read]),
+                    }
+                }
+                if let Some(path) = String::from_utf8_lossy(&head).split_whitespace().nth(1) {
+                    let _ = paths.send(path.to_owned());
+                }
+                let _ = stream
+                    .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n")
+                    .await;
+            });
+        }
+    });
+    (format!("http://{address}/twirp"), received)
+}
+
+/// [`establish_fake`] whose signal RPCs go to a [`recording_twirp_server`].
+async fn establish_fake_recording_rpcs(
+    core: &Arc<RtcCore>,
+    generation: u64,
+) -> (
+    Connection,
+    tokio::sync::mpsc::UnboundedReceiver<event::SfuRequest>,
+    tokio::sync::mpsc::UnboundedReceiver<String>,
+) {
+    let (mut credentials, sfu) = fake_sfu(true).await;
+    let (twirp_url, rpcs) = recording_twirp_server().await;
+    credentials.server.url = twirp_url;
+    let connection = core
+        .clone()
+        .establish(
+            &credentials,
+            0,
+            ReconnectStrategy::Fast,
+            None,
+            generation,
+            None,
+        )
+        .await
+        .expect("establish against fake SFU");
+    (connection, sfu, rpcs)
+}
+
+fn received_paths(rpcs: &mut tokio::sync::mpsc::UnboundedReceiver<String>) -> Vec<String> {
+    std::iter::from_fn(|| rpcs.try_recv().ok()).collect()
+}
+
 /// The event loop context of the current `connection`.
 fn event_context(core: &Arc<RtcCore>, connection: &Connection) -> EventLoopContext {
     EventLoopContext {
@@ -1205,6 +1270,130 @@ async fn a_failed_publish_does_not_hold_the_track() {
     core.leave("test cleanup").await.expect("leave");
 }
 
+#[tokio::test]
+async fn a_publish_that_waits_for_a_reconnect_negotiates_with_the_new_session() {
+    let core = test_core();
+    let generation = prepare_joined_core(&core, "alice");
+    let (mut old, _old_sfu, mut old_rpcs) = establish_fake_recording_rpcs(&core, generation).await;
+    let (mut new, _new_sfu, mut new_rpcs) = establish_fake_recording_rpcs(&core, generation).await;
+    for connection in [&mut old, &mut new] {
+        connection.publish_options = vec![models::PublishOption {
+            track_type: TrackType::Audio as i32,
+            codec: Some(models::Codec {
+                name: "opus".to_owned(),
+                ..Default::default()
+            }),
+            ..Default::default()
+        }];
+    }
+    *core.connection.lock().await = Some(old);
+    core.own_capabilities
+        .lock()
+        .unwrap_or_else(|error| error.into_inner())
+        .insert("send-audio".to_owned());
+    let audio = LocalAudioTrack::opus().expect("opus track");
+
+    // Another media operation holds the lock while a reconnect replaces the
+    // connection. In a migration, the old SFU session stays reachable.
+    let busy = core.media.lock().await;
+    let publish = tokio::spawn({
+        let core = core.clone();
+        let track = LocalTrack::Audio(audio.clone());
+        async move { core.publish(track).await }
+    });
+    tokio::task::yield_now().await;
+    let old = core
+        .connection
+        .lock()
+        .await
+        .replace(new)
+        .expect("old connection");
+    drop(busy);
+    publish
+        .await
+        .expect("publish task")
+        .expect_err("the fake SFU answers no SDP");
+
+    let set_publisher = "/twirp/stream.video.sfu.signal.SignalServer/SetPublisher";
+    assert!(
+        received_paths(&mut new_rpcs)
+            .iter()
+            .any(|path| path == set_publisher)
+    );
+    assert!(
+        !received_paths(&mut old_rpcs)
+            .iter()
+            .any(|path| path == set_publisher)
+    );
+    old.teardown().await;
+    audio.stop();
+    core.leave("test cleanup").await.expect("leave");
+}
+
+#[tokio::test]
+async fn restoring_publications_adds_no_second_sender_for_a_track_published_meanwhile() {
+    let core = test_core();
+    let generation = prepare_joined_core(&core, "alice");
+    let (mut connection, _sfu) = establish_fake(&core, generation).await;
+    connection.publish_options = ["opus", "vp8"]
+        .into_iter()
+        .zip([TrackType::Audio, TrackType::Video])
+        .map(|(codec, track_type)| models::PublishOption {
+            track_type: track_type as i32,
+            codec: Some(models::Codec {
+                name: codec.to_owned(),
+                ..Default::default()
+            }),
+            ..Default::default()
+        })
+        .collect();
+    let publisher = connection.publisher.clone();
+    let publish_options = connection.publish_options.clone();
+    *core.connection.lock().await = Some(connection);
+    core.own_capabilities
+        .lock()
+        .unwrap_or_else(|error| error.into_inner())
+        .extend(["send-audio".to_owned(), "send-video".to_owned()]);
+    let earlier = LocalTrack::Video {
+        track: LocalVideoTrack::vp8().expect("vp8 track"),
+        track_type: TrackType::Video,
+    };
+    let meanwhile = LocalTrack::Audio(LocalAudioTrack::opus().expect("opus track"));
+    let reader_tasks =
+        publisher::add_transceiver_for_track(&publisher, &meanwhile, &publish_options)
+            .await
+            .expect("audio transceiver");
+    {
+        let mut media = core.media.lock().await;
+        for track in [&earlier, &meanwhile] {
+            media.begin_publish(track.clone(), 0);
+            media.set_status(&track.track_id(), PublicationStatus::Published);
+        }
+    }
+
+    // The fake SFU has no SetPublisher endpoint.
+    core.restore_published_tracks()
+        .await
+        .expect_err("restore without SetPublisher");
+
+    let mut sender_tracks = Vec::new();
+    for transceiver in publisher.get_transceivers().await {
+        if let Some(track) = transceiver.sender().await.track().await {
+            sender_tracks.push(track.id().to_owned());
+        }
+    }
+    sender_tracks.sort();
+    let mut expected = vec![earlier.track_id(), meanwhile.track_id()];
+    expected.sort();
+    assert_eq!(sender_tracks, expected);
+    for task in reader_tasks {
+        task.abort();
+    }
+    earlier.stop();
+    meanwhile.stop();
+    core.leave("test cleanup").await.expect("leave");
+}
+
 #[tokio::test]
 async fn a_stopped_simulcast_publication_is_not_replaced() {
     let core = test_core();