webrtc: Fix data channel not closing after FIN/FIN_ACK by detecting substream drop

Author: timwu20

src/transport/webrtc/substream.rs

@@ -212,6 +212,17 @@ impl Stream for SubstreamHandle {
     type Item = Event;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // Check if Substream has been dropped (inbound channel closed)
+        // If so, and we're in a terminal state, signal stream closure
+        if self.inbound_tx.is_closed() {
+            let state = *self.state.lock();
+            // Only close if we're in a terminal state (FinAcked or SendClosed)
+            // This ensures proper cleanup after FIN/FIN_ACK handshake completes
+            if matches!(state, State::FinAcked | State::SendClosed) {
+                return Poll::Ready(None);
+            }
+        }
+
         self.rx.poll_recv(cx)
     }
 }
@@ -1122,4 +1133,45 @@ mod tests {
         assert!(result.is_err());
         assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::BrokenPipe);
     }
+
+    #[tokio::test]
+    async fn handle_signals_closure_after_substream_dropped() {
+        use futures::StreamExt;
+
+        let (mut substream, mut handle) = Substream::new();
+
+        // Complete shutdown handshake
+        let shutdown_task = tokio::spawn(async move {
+            substream.shutdown().await.unwrap();
+            // Substream will be dropped here
+        });
+
+        // Receive FIN
+        assert_eq!(
+            handle.next().await,
+            Some(Event::Message {
+                payload: vec![],
+                flag: Some(Flag::Fin as i32)
+            })
+        );
+
+        // Send FIN_ACK
+        handle
+            .on_message(WebRtcMessage {
+                payload: None,
+                flag: Some(Flag::FinAck as i32),
+            })
+            .await
+            .unwrap();
+
+        // Wait for shutdown to complete and Substream to drop
+        shutdown_task.await.unwrap();
+
+        // Verify handle signals closure (returns None)
+        assert_eq!(
+            handle.next().await,
+            None,
+            "SubstreamHandle should signal closure after Substream is dropped"
+        );
+    }
 }