From db1bc23e07a6bab200d08d42b223ccd7b0d2963e Mon Sep 17 00:00:00 2001 From: yexiyue Date: Fri, 24 Jul 2026 17:24:00 +0800 Subject: [PATCH] fix(webrtc-websys): defer data channel callback wakes --- .../src/stream/poll_data_channel.rs | 60 ++++++++++++++----- 1 file changed, 45 insertions(+), 15 deletions(-) diff --git a/transports/webrtc-websys/src/stream/poll_data_channel.rs b/transports/webrtc-websys/src/stream/poll_data_channel.rs index eaeec4d6789..3d1d0eb9205 100644 --- a/transports/webrtc-websys/src/stream/poll_data_channel.rs +++ b/transports/webrtc-websys/src/stream/poll_data_channel.rs @@ -45,12 +45,39 @@ pub(crate) struct PollDataChannel { /// cause the application developer to drop the stream which resets it. overloaded: Rc, - // Store the closures for proper garbage collection. - // These are wrapped in an [`Rc`] so we can implement [`Clone`]. - _on_open_closure: Rc>, - _on_write_closure: Rc>, - _on_close_closure: Rc>, - _on_message_closure: Rc>, + /// Owns the JavaScript callbacks for the shared lifetime of all clones. + _event_handlers: Rc, +} + +/// Keeps the event callbacks alive and detaches them when the final stream clone is dropped. +#[derive(Debug)] +struct EventHandlers { + inner: RtcDataChannel, + _on_open_closure: Closure, + _on_write_closure: Closure, + _on_close_closure: Closure, + _on_message_closure: Closure, +} + +impl Drop for EventHandlers { + fn drop(&mut self) { + // The browser may queue an event after Rust drops a stream. Detach each handler before + // wasm-bindgen releases its Closure so a queued event cannot invoke a dropped closure. + self.inner.set_onopen(None); + self.inner.set_onbufferedamountlow(None); + self.inner.set_onclose(None); + self.inner.set_onmessage(None); + } +} + +/// Wakes outside the Web API callback that triggered the event. +/// +/// Calling `wake` synchronously from an `RtcDataChannel` callback can re-enter the executor +/// while wasm-bindgen still mutably borrows that callback closure. +fn defer_wake(waker: Rc) { + wasm_bindgen_futures::spawn_local(async move { + waker.wake(); + }); } impl PollDataChannel { @@ -61,7 +88,7 @@ impl PollDataChannel { move |_: RtcDataChannelEvent| { tracing::trace!("DataChannel opened"); - open_waker.wake(); + defer_wake(open_waker.clone()); } }); inner.set_onopen(Some(on_open_closure.as_ref().unchecked_ref())); @@ -73,7 +100,7 @@ impl PollDataChannel { move |_: Event| { tracing::trace!("DataChannel available for writing (again)"); - write_waker.wake(); + defer_wake(write_waker.clone()); } }); inner.set_onbufferedamountlow(Some(on_write_closure.as_ref().unchecked_ref())); @@ -84,7 +111,7 @@ impl PollDataChannel { move |_: Event| { tracing::trace!("DataChannel closed"); - close_waker.wake(); + defer_wake(close_waker.clone()); } }); inner.set_onclose(Some(on_close_closure.as_ref().unchecked_ref())); @@ -112,23 +139,26 @@ impl PollDataChannel { } read_buffer.extend_from_slice(&data.to_vec()); - new_data_waker.wake(); + defer_wake(new_data_waker.clone()); } }); inner.set_onmessage(Some(on_message_closure.as_ref().unchecked_ref())); Self { - inner, + inner: inner.clone(), new_data_waker, read_buffer, open_waker, write_waker, close_waker, overloaded, - _on_open_closure: Rc::new(on_open_closure), - _on_write_closure: Rc::new(on_write_closure), - _on_close_closure: Rc::new(on_close_closure), - _on_message_closure: Rc::new(on_message_closure), + _event_handlers: Rc::new(EventHandlers { + inner, + _on_open_closure: on_open_closure, + _on_write_closure: on_write_closure, + _on_close_closure: on_close_closure, + _on_message_closure: on_message_closure, + }), } }