From 9f2aea3b0854c584902db453e44dfd60b796fa2c Mon Sep 17 00:00:00 2001 From: Thomas Kirkman-Wood Date: Sat, 1 Aug 2026 00:55:50 +0100 Subject: [PATCH 1/2] fix: only start one hotplug watcher per emitter start_watching checked whether a watcher was live and then stored a new one, releasing the lock in between. addEventListener starts addAttach and addDetach back-to-back without awaiting, so both ran at once, both saw an empty slot, and each spawned a watcher. Two watchers delivered every hotplug event twice, which surfaced as connect firing twice for a single plug-in. disconnect was unaffected because the JS side already dedupes it. Hold one lock across the check and the store, and take the napi methods from &mut self to &self: napi can poll them concurrently, so &mut self was handing out aliasing mutable references. They no longer need to be unsafe. Regression from the watch_task refactor in #18; v3.0.0 is unaffected. --- src/lib.rs | 53 ++++++++++++++++++++++++++++++++++------------------- 1 file changed, 34 insertions(+), 19 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 3469815..916612f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,10 +24,18 @@ fn callbacks_guard(callbacks: &Mutex) -> MutexGuard<'_, Callbacks> { .unwrap_or_else(|poisoned| poisoned.into_inner()) } +fn watch_task_guard( + watch_task: &Mutex>>, +) -> MutexGuard<'_, Option>> { + watch_task + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + #[napi] pub struct Emitter { callbacks: Arc>, - watch_task: Option>, + watch_task: Arc>>>, } #[napi] @@ -44,16 +52,21 @@ impl Emitter { })); Self { callbacks, - watch_task: None, + watch_task: Arc::new(Mutex::new(None)), } } - async fn start_watching(&mut self) -> Result<()> { - if matches!(self.watch_task.as_ref(), Some(task) if !task.is_finished()) { + // addAttach and addDetach run at the same time, so one lock has to cover + // both the check and the store. Release it in between and each starts a + // watcher, which then reports every plug twice. + fn start_watching(&self) -> Result<()> { + let mut watch_task = watch_task_guard(&self.watch_task); + + if matches!(watch_task.as_ref(), Some(task) if !task.is_finished()) { return Ok(()); } - self.watch_task = None; + *watch_task = None; let callbacks = self.callbacks.clone(); let mut watch_stream = match nusb::watch_devices() { Ok(watch_stream) => watch_stream, @@ -64,7 +77,7 @@ impl Emitter { } }; - self.watch_task = Some(tokio::spawn(async move { + *watch_task = Some(tokio::spawn(async move { while let Some(ev) = watch_stream.next().await { match ev { HotplugEvent::Connected(info) => { @@ -89,61 +102,63 @@ impl Emitter { Ok(()) } - async fn stop_watching(&mut self) { + fn stop_watching(&self) { + // Let the callbacks lock go before taking watch_task: the watcher locks + // callbacks itself, so the reverse order could deadlock. let has_listeners = { let cb = self.callbacks(); cb.attach.is_some() || cb.detach.is_some() }; if !has_listeners { - if let Some(task) = self.watch_task.take() { + if let Some(task) = watch_task_guard(&self.watch_task).take() { task.abort(); } } } #[napi] - pub async unsafe fn addAttach( - &mut self, + pub async fn addAttach( + &self, callback: ThreadsafeFunction, ) -> Result<()> { { self.callbacks().attach = Some(callback); } - self.start_watching().await + self.start_watching() } #[napi] - pub async unsafe fn removeAttach(&mut self) { + pub async fn removeAttach(&self) { { self.callbacks().attach = None; } - self.stop_watching().await; + self.stop_watching(); } #[napi] - pub async unsafe fn addDetach( - &mut self, + pub async fn addDetach( + &self, callback: ThreadsafeFunction, ) -> Result<()> { { self.callbacks().detach = Some(callback); } - self.start_watching().await + self.start_watching() } #[napi] - pub async unsafe fn removeDetach(&mut self) { + pub async fn removeDetach(&self) { { self.callbacks().detach = None; } - self.stop_watching().await; + self.stop_watching(); } } impl Drop for Emitter { fn drop(&mut self) { - if let Some(task) = self.watch_task.take() { + if let Some(task) = watch_task_guard(&self.watch_task).take() { task.abort(); } } From c3029fb471de8d03c30cd51adb35403d1b999cbe Mon Sep 17 00:00:00 2001 From: Thomas Kirkman-Wood Date: Sat, 1 Aug 2026 01:09:20 +0100 Subject: [PATCH 2/2] fix: hold the callbacks lock while deciding to stop watching --- src/lib.rs | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 916612f..12f86b6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -103,12 +103,11 @@ impl Emitter { } fn stop_watching(&self) { - // Let the callbacks lock go before taking watch_task: the watcher locks - // callbacks itself, so the reverse order could deadlock. - let has_listeners = { - let cb = self.callbacks(); - cb.attach.is_some() || cb.detach.is_some() - }; + // Keep the callbacks lock for the whole decision. Drop it after the + // check and a listener registered in the gap would find its watcher + // aborted here, leaving nothing watching. + let callbacks = self.callbacks(); + let has_listeners = callbacks.attach.is_some() || callbacks.detach.is_some(); if !has_listeners { if let Some(task) = watch_task_guard(&self.watch_task).take() {