From 8a91d0def649ff432c6cc00b6c35217b32c2b174 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Fri, 9 Oct 2026 17:53:39 -0700
Subject: [PATCH] fix(tokio): keep a deferred FIN when a send stream drops
 mid-watch

An idle closed watch owns the stream, so finish() only records the FIN.
Dropping the stream before the next poll discarded the watch and let the
backend reset instead. Reclaim the interrupted watch and apply a deferred
finish as well as a deferred reset, like the in-flight write path.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-tokio/src/transport.rs | 31 ++++++++++++++++++++++++++++---
 1 file changed, 28 insertions(+), 3 deletions(-)

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<S: web_transport_trait::SendStream + 'static> Drop for SendStream<S> {
 					});
 				}
 			}
+			// 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]