diff --git a/rs/moq-tokio/src/transport.rs b/rs/moq-tokio/src/transport.rs index 19ba714eab..e49fb8f138 100644 --- a/rs/moq-tokio/src/transport.rs +++ b/rs/moq-tokio/src/transport.rs @@ -560,15 +560,21 @@ impl Drop for SendStream { }); } } + // Both deferred actions fired the watch's interrupt when recorded, so + // the salvage task reclaims the stream without waiting on the peer. Some(SendState::Closing(fut)) => { let reset = self.reset.take(); - if reset.is_some() + let finish = std::mem::take(&mut self.finish); + if (reset.is_some() || finish) && let Ok(handle) = tokio::runtime::Handle::try_current() { handle.spawn(async move { let (mut stream, _) = fut.await; - if let Some(code) = reset { - stream.reset(code); + match reset { + Some(code) => stream.reset(code), + None => { + let _ = stream.finish(); + } } }); } @@ -1198,6 +1204,25 @@ mod tests { assert_eq!(fake.resets.lock().unwrap().as_slice(), &[9]); } + // An idle closed watch owns the stream, so a FIN recorded against it and then + // dropped must still go out instead of the backend's drop reset. + #[tokio::test(start_paused = true)] + async fn a_deferred_finish_survives_a_drop_mid_watch() { + let fake = FakeSend::default(); + let mut send = SendStream::new(fake.clone()); + let mut cx = cx(); + + assert!(send.poll_closed(&mut cx).is_pending()); + send.finish().unwrap(); + drop(send); + + for _ in 0..10 { + tokio::task::yield_now().await; + } + assert!(fake.finished.load(Ordering::SeqCst), "the deferred FIN was lost"); + assert!(fake.resets.lock().unwrap().is_empty()); + } + // The receive-side twin: a reader dropped mid-read still sends the deferred // STOP_SENDING with the real code. #[tokio::test]