From 316a73ac31657c8f00b4da04f5c72ef6e9c92d85 Mon Sep 17 00:00:00 2001 From: Roma Glushko Date: Mon, 9 Mar 2026 17:42:46 +0100 Subject: [PATCH 1/6] Remove potential cross-threads contention by eliminating Mutex on Batch Processor --- src/watcher.rs | 63 +++++++++++++++++++++++--------------------------- 1 file changed, 29 insertions(+), 34 deletions(-) diff --git a/src/watcher.rs b/src/watcher.rs index db3e8a8..d5d7c2f 100644 --- a/src/watcher.rs +++ b/src/watcher.rs @@ -16,7 +16,7 @@ use notify::{ use pyo3::exceptions::{PyException, PyFileNotFoundError, PyOSError, PyPermissionError}; use pyo3::prelude::*; use tokio::{ - sync::{broadcast, oneshot}, + sync::{broadcast, mpsc, oneshot}, time, }; // use crate::file_cache::FileCache; @@ -29,9 +29,10 @@ pyo3::create_exception!(_inotify_toolkit_lib, WatcherError, PyException); pub(crate) struct Watcher { debug: bool, event_buffer_size: usize, + buffering_duration: Duration, inner: RecommendedWatcher, // file_cache: FileCache, - processor: Arc>, // TODO: use the EventProcessor trait instead + event_rx: Arc>>>, tx: broadcast::Sender>, stop_tx: Option>, drain_handle: Option>, @@ -44,13 +45,8 @@ impl Watcher { debug: bool, follow_symlinks: bool, ) -> Result { - // TODO: hide usage of file cache from Watcher - // let file_cache = FileCache::new(); - // let file_cache_c = file_cache.clone(); - let buffering_duration = Duration::from_millis(buffering_time_ms); - let processor = Arc::new(Mutex::new(BatchProcessor::new(buffering_duration))); - let processor_c = processor.clone(); + let (event_tx, event_rx) = mpsc::unbounded_channel(); let (tx, _rx) = broadcast::channel::>(event_buffer_size); @@ -58,21 +54,11 @@ impl Watcher { let inner = RecommendedWatcher::new( move |e: Result| { - let mut event_processor = match processor_c.lock() { - Ok(guard) => guard, - Err(e) => { - eprintln!("notifykit: event processor lock poisoned, dropping event: {e}"); - return; - } - }; - if debug { println!("raw event: {:?}", e); } - - match e { - Ok(e) => event_processor.add_event(e), - Err(e) => event_processor.add_error(e), + if let Err(e) = event_tx.send(e) { + eprintln!("failed to send event: {e}"); } }, config, @@ -81,8 +67,9 @@ impl Watcher { Ok(Self { debug, event_buffer_size, + buffering_duration, inner, - processor, + event_rx: Arc::new(Mutex::new(event_rx)), tx, stop_tx: None, drain_handle: None, @@ -157,7 +144,7 @@ impl Watcher { self.tx = new_tx; } - pub fn start_drain(&mut self, debounce_delay: Duration, event_filter: Option) { + pub fn start_drain(&mut self, tick_duration: Duration, event_filter: Option) { if let Some(tx) = self.stop_tx.take() { let _ = tx.send(()); } @@ -169,27 +156,31 @@ impl Watcher { let (stop_tx, mut stop_rx) = oneshot::channel(); self.stop_tx = Some(stop_tx); - let proc = Arc::clone(&self.processor); + let event_rx = Arc::clone(&self.event_rx); let tx = self.tx.clone(); let debug = self.debug; + let buffering_duration = self.buffering_duration; self.drain_handle = Some(pyo3_async_runtimes::tokio::get_runtime().spawn(async move { - let mut ticker = time::interval(debounce_delay); + let mut processor = BatchProcessor::new(buffering_duration); + let mut ticker = time::interval(tick_duration); loop { tokio::select! { _ = &mut stop_rx => break, _ = ticker.tick() => { - let (raw, errs) = { - let mut p = match proc.lock() { - Ok(guard) => guard, - Err(e) => { - eprintln!("notifykit: event processor lock poisoned, skipping drain tick: {e}"); - continue; + if let Ok(mut rx) = event_rx.lock() { + while let Ok(result) = rx.try_recv() { + match result { + Ok(event) => processor.add_event(event), + Err(error) => processor.add_error(error), } - }; - (p.get_events(), p.get_errors()) - }; + } + } + + let raw = processor.get_events(); + let errs = processor.get_errors(); + if debug && !raw.is_empty() { println!("processed: {:?}", raw); } if !errs.is_empty() { eprintln!("errors: {:?}", errs); } if raw.is_empty() { continue; } @@ -207,7 +198,11 @@ impl Watcher { } } - if !batch.is_empty() { let _ = tx.send(batch); } + if !batch.is_empty() { + if let Err(e) = tx.send(batch) { + eprintln!("failed to broadcast events: {e}"); + } + } } } } From b52fac327076a8f787fd8ffe30dbf388d2989642 Mon Sep 17 00:00:00 2001 From: Roma Glushko Date: Mon, 9 Mar 2026 19:49:16 +0100 Subject: [PATCH 2/6] Use bounded channel & avoid mutex poisoning --- src/watcher.rs | 52 +++++++++++++++++++++++++++++++++----------------- 1 file changed, 34 insertions(+), 18 deletions(-) diff --git a/src/watcher.rs b/src/watcher.rs index d5d7c2f..cee5105 100644 --- a/src/watcher.rs +++ b/src/watcher.rs @@ -1,6 +1,5 @@ use std::io::ErrorKind as IOErrorKind; use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; use std::time::Duration; use crate::events::EventType; @@ -32,7 +31,8 @@ pub(crate) struct Watcher { buffering_duration: Duration, inner: RecommendedWatcher, // file_cache: FileCache, - event_rx: Arc>>>, + event_rx: Option>>, + rx_return: Option>>>, tx: broadcast::Sender>, stop_tx: Option>, drain_handle: Option>, @@ -46,7 +46,7 @@ impl Watcher { follow_symlinks: bool, ) -> Result { let buffering_duration = Duration::from_millis(buffering_time_ms); - let (event_tx, event_rx) = mpsc::unbounded_channel(); + let (event_tx, event_rx) = mpsc::channel(event_buffer_size); let (tx, _rx) = broadcast::channel::>(event_buffer_size); @@ -57,8 +57,8 @@ impl Watcher { if debug { println!("raw event: {:?}", e); } - if let Err(e) = event_tx.send(e) { - eprintln!("failed to send event: {e}"); + if let Err(e) = event_tx.try_send(e) { + eprintln!("event channel full or closed, dropping event: {e}"); } }, config, @@ -69,7 +69,8 @@ impl Watcher { event_buffer_size, buffering_duration, inner, - event_rx: Arc::new(Mutex::new(event_rx)), + event_rx: Some(event_rx), + rx_return: None, tx, stop_tx: None, drain_handle: None, @@ -136,12 +137,22 @@ impl Watcher { let _ = tx.send(()); } + self.recover_event_rx(); + + let (new_tx, _rx) = broadcast::channel::>(self.event_buffer_size); + self.tx = new_tx; + } + + fn recover_event_rx(&mut self) { if let Some(handle) = self.drain_handle.take() { handle.abort(); } - let (new_tx, _rx) = broadcast::channel::>(self.event_buffer_size); - self.tx = new_tx; + if let Some(mut rx_return) = self.rx_return.take() { + if let Ok(rx) = rx_return.try_recv() { + self.event_rx = Some(rx); + } + } } pub fn start_drain(&mut self, tick_duration: Duration, event_filter: Option) { @@ -149,14 +160,19 @@ impl Watcher { let _ = tx.send(()); } - if let Some(handle) = self.drain_handle.take() { - handle.abort(); - } + self.recover_event_rx(); + + let mut event_rx = match self.event_rx.take() { + Some(rx) => rx, + None => return, + }; let (stop_tx, mut stop_rx) = oneshot::channel(); self.stop_tx = Some(stop_tx); - let event_rx = Arc::clone(&self.event_rx); + let (rx_return_tx, rx_return_rx) = oneshot::channel(); + self.rx_return = Some(rx_return_rx); + let tx = self.tx.clone(); let debug = self.debug; let buffering_duration = self.buffering_duration; @@ -169,12 +185,10 @@ impl Watcher { tokio::select! { _ = &mut stop_rx => break, _ = ticker.tick() => { - if let Ok(mut rx) = event_rx.lock() { - while let Ok(result) = rx.try_recv() { - match result { - Ok(event) => processor.add_event(event), - Err(error) => processor.add_error(error), - } + while let Ok(result) = event_rx.try_recv() { + match result { + Ok(event) => processor.add_event(event), + Err(error) => processor.add_error(error), } } @@ -206,6 +220,8 @@ impl Watcher { } } } + + let _ = rx_return_tx.send(event_rx); })); } From 1f4fc259955266456e8fb8d5c12bdb472fef6862 Mon Sep 17 00:00:00 2001 From: Roma Glushko Date: Mon, 9 Mar 2026 20:04:31 +0100 Subject: [PATCH 3/6] Make sure the event timestamp is not drifted by buffering time --- src/processor.rs | 35 +++++++++++++++++------------------ src/watcher.rs | 12 ++++++------ 2 files changed, 23 insertions(+), 24 deletions(-) diff --git a/src/processor.rs b/src/processor.rs index 5b0e402..4d553b1 100644 --- a/src/processor.rs +++ b/src/processor.rs @@ -89,7 +89,7 @@ impl FileEventQueue { pub(crate) trait EventProcessor { fn get_events(&mut self) -> Vec; fn get_errors(&mut self) -> Vec; - fn add_event(&mut self, event: NotifyEvent); + fn add_event(&mut self, event: NotifyEvent, time: Instant); fn add_error(&mut self, error: NotifyError); } @@ -117,8 +117,7 @@ impl CrossPlatformEventProcessor { } } - fn handle_rename_from(&mut self, event: NotifyEvent) { - let time = Instant::now(); + fn handle_rename_from(&mut self, event: NotifyEvent, time: Instant) { let path = &event.paths[0]; // store event @@ -130,7 +129,7 @@ impl CrossPlatformEventProcessor { self.push_event(event, time); } - fn handle_rename_to(&mut self, event: NotifyEvent) { + fn handle_rename_to(&mut self, event: NotifyEvent, time: Instant) { self.file_cache.add_path(&event.paths[0]); let trackers_match = self @@ -155,11 +154,11 @@ impl CrossPlatformEventProcessor { // connect rename let (mut rename_event, _) = self.rename_event.take().unwrap(); // unwrap is safe because `rename_event` must be set at this point let path = rename_event.paths.remove(0); - let time = rename_event.time; - self.push_rename_event(path, event, time); + let rename_time = rename_event.time; + self.push_rename_event(path, event, rename_time); } else { // move in - self.push_event(event, Instant::now()); + self.push_event(event, time); } self.rename_event = None; @@ -351,12 +350,12 @@ impl EventProcessor for CrossPlatformEventProcessor { } /// Add new event to debouncer cache - fn add_event(&mut self, event: NotifyEvent) { + fn add_event(&mut self, event: NotifyEvent, time: Instant) { // log::trace!("raw event: {event:?}"); if event.need_rescan() { self.file_cache.rescan(); - self.rescan_event = Some(event.into()); + self.rescan_event = Some(RawEvent::new(event, time)); return; } @@ -366,22 +365,22 @@ impl EventProcessor for CrossPlatformEventProcessor { EventKind::Create(_) => { self.file_cache.add_path(path); - self.push_event(event, Instant::now()); + self.push_event(event, time); } EventKind::Modify(ModifyKind::Name(rename_mode)) => { match rename_mode { RenameMode::Any => { if event.paths[0].exists() { - self.handle_rename_to(event); + self.handle_rename_to(event, time); } else { - self.handle_rename_from(event); + self.handle_rename_from(event, time); } } RenameMode::To => { - self.handle_rename_to(event); + self.handle_rename_to(event, time); } RenameMode::From => { - self.handle_rename_from(event); + self.handle_rename_from(event, time); } RenameMode::Both => { // ignore and handle `To` and `From` events instead @@ -392,7 +391,7 @@ impl EventProcessor for CrossPlatformEventProcessor { } } EventKind::Remove(_) => { - self.push_remove_event(event, Instant::now()); + self.push_remove_event(event, time); } EventKind::Other => { // ignore meta events @@ -402,7 +401,7 @@ impl EventProcessor for CrossPlatformEventProcessor { self.file_cache.add_path(path); } - self.push_event(event, Instant::now()); + self.push_event(event, time); } } } @@ -458,8 +457,8 @@ impl EventProcessor for BatchProcessor { std::mem::take(&mut self.errors) } - fn add_event(&mut self, event: NotifyEvent) { - self.events.push(RawEvent::new(event, Instant::now())); + fn add_event(&mut self, event: NotifyEvent, time: Instant) { + self.events.push(RawEvent::new(event, time)); } fn add_error(&mut self, error: NotifyError) { diff --git a/src/watcher.rs b/src/watcher.rs index cee5105..3cb1717 100644 --- a/src/watcher.rs +++ b/src/watcher.rs @@ -1,6 +1,6 @@ use std::io::ErrorKind as IOErrorKind; use std::path::{Path, PathBuf}; -use std::time::Duration; +use std::time::{Duration, Instant}; use crate::events::EventType; use crate::events::access::from_access_kind; @@ -31,8 +31,8 @@ pub(crate) struct Watcher { buffering_duration: Duration, inner: RecommendedWatcher, // file_cache: FileCache, - event_rx: Option>>, - rx_return: Option>>>, + event_rx: Option)>>, + rx_return: Option)>>>, tx: broadcast::Sender>, stop_tx: Option>, drain_handle: Option>, @@ -57,7 +57,7 @@ impl Watcher { if debug { println!("raw event: {:?}", e); } - if let Err(e) = event_tx.try_send(e) { + if let Err(e) = event_tx.try_send((Instant::now(), e)) { eprintln!("event channel full or closed, dropping event: {e}"); } }, @@ -185,9 +185,9 @@ impl Watcher { tokio::select! { _ = &mut stop_rx => break, _ = ticker.tick() => { - while let Ok(result) = event_rx.try_recv() { + while let Ok((time, result)) = event_rx.try_recv() { match result { - Ok(event) => processor.add_event(event), + Ok(event) => processor.add_event(event, time), Err(error) => processor.add_error(error), } } From 674f272a530587e8351b3456315e9fc1bc4707f4 Mon Sep 17 00:00:00 2001 From: Roma Glushko Date: Mon, 9 Mar 2026 20:10:32 +0100 Subject: [PATCH 4/6] Linting --- src/watcher.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/watcher.rs b/src/watcher.rs index 3cb1717..3faceef 100644 --- a/src/watcher.rs +++ b/src/watcher.rs @@ -24,6 +24,9 @@ use crate::processor::{BatchProcessor, EventProcessor, RawEvent}; pyo3::create_exception!(_inotify_toolkit_lib, WatcherError, PyException); +type TimestampedEvent = (Instant, Result); +type EventReceiver = mpsc::Receiver; + #[derive(Debug)] pub(crate) struct Watcher { debug: bool, @@ -31,8 +34,8 @@ pub(crate) struct Watcher { buffering_duration: Duration, inner: RecommendedWatcher, // file_cache: FileCache, - event_rx: Option)>>, - rx_return: Option)>>>, + event_rx: Option, + rx_return: Option>, tx: broadcast::Sender>, stop_tx: Option>, drain_handle: Option>, From 15e247c4c2e3027a8423ca5c12bc32765aaba7e2 Mon Sep 17 00:00:00 2001 From: Roma Glushko Date: Mon, 9 Mar 2026 20:15:38 +0100 Subject: [PATCH 5/6] Made sure maturine is running on the new rust toolchain --- .github/workflows/ci.yaml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 08f087c..0724bc0 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -110,6 +110,7 @@ jobs: args: --release --out dist sccache: 'true' manylinux: ${{ matrix.manylinux || 'auto' }} + rust-toolchain: '1.85' - name: Build PyPy wheels if: ${{ matrix.manylinux == 'auto' && (matrix.target == 'x86_64' || matrix.target == 'aarch64') }} @@ -119,6 +120,7 @@ jobs: sccache: 'true' manylinux: ${{ matrix.manylinux || 'auto' }} args: --release --out dist --interpreter pypy3.11 + rust-toolchain: '1.85' - run: ls -lah dist/ @@ -148,6 +150,7 @@ jobs: target: ${{ matrix.target }} args: --release --out dist sccache: 'true' + rust-toolchain: '1.85' - name: Build PyPy wheels if: ${{ matrix.target == 'x86_64' }} @@ -155,6 +158,7 @@ jobs: with: target: ${{ matrix.target }} args: --release --out dist --interpreter pypy3.11 + rust-toolchain: '1.85' - run: ls -lah dist/ @@ -185,6 +189,7 @@ jobs: target: ${{ matrix.target }} args: --release --out dist sccache: 'true' + rust-toolchain: '1.85' - run: dir dist/ @@ -205,6 +210,7 @@ jobs: with: command: sdist args: --out dist + rust-toolchain: '1.85' - name: Upload sdist uses: actions/upload-artifact@v4 From bb84a816d5a69db2c80df0a512d8f8c8f716b242 Mon Sep 17 00:00:00 2001 From: Roma Glushko Date: Mon, 9 Mar 2026 20:27:48 +0100 Subject: [PATCH 6/6] Extract Event Queue Drain logic into a separate testable component --- src/watcher.rs | 301 +++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 264 insertions(+), 37 deletions(-) diff --git a/src/watcher.rs b/src/watcher.rs index 3faceef..9546b06 100644 --- a/src/watcher.rs +++ b/src/watcher.rs @@ -27,6 +27,74 @@ pyo3::create_exception!(_inotify_toolkit_lib, WatcherError, PyException); type TimestampedEvent = (Instant, Result); type EventReceiver = mpsc::Receiver; +#[derive(Debug)] +struct DrainState { + event_rx: Option, + rx_return: Option>, + stop_tx: Option>, + drain_handle: Option>, +} + +impl DrainState { + fn new(event_rx: EventReceiver) -> Self { + Self { + event_rx: Some(event_rx), + rx_return: None, + stop_tx: None, + drain_handle: None, + } + } + + fn send_stop_signal(&mut self) { + if let Some(tx) = self.stop_tx.take() { + let _ = tx.send(()); + } + } + + fn recover_event_rx(&mut self, runtime: &tokio::runtime::Runtime) { + if let Some(handle) = self.drain_handle.take() { + // Wait for the task to finish so it can return the receiver. + // The caller must send the stop signal before calling this. + let _ = runtime.block_on(handle); + } + + if let Some(mut rx_return) = self.rx_return.take() { + match rx_return.try_recv() { + Ok(rx) => self.event_rx = Some(rx), + Err(oneshot::error::TryRecvError::Empty) => { + // Task hasn't returned the receiver yet; keep the handle for later + self.rx_return = Some(rx_return); + } + Err(oneshot::error::TryRecvError::Closed) => { + // Task dropped the sender (e.g. panic); receiver is lost + eprintln!("notifykit: drain task exited without returning event receiver"); + } + } + } + } + + fn stop_and_recover(&mut self, runtime: &tokio::runtime::Runtime) { + self.send_stop_signal(); + self.recover_event_rx(runtime); + } + + fn take_event_rx(&mut self) -> Option { + self.event_rx.take() + } + + fn set_drain( + &mut self, + handle: tokio::task::JoinHandle<()>, + stop_tx: oneshot::Sender<()>, + rx_return: oneshot::Receiver, + ) { + self.drain_handle = Some(handle); + self.stop_tx = Some(stop_tx); + self.rx_return = Some(rx_return); + } + +} + #[derive(Debug)] pub(crate) struct Watcher { debug: bool, @@ -34,11 +102,8 @@ pub(crate) struct Watcher { buffering_duration: Duration, inner: RecommendedWatcher, // file_cache: FileCache, - event_rx: Option, - rx_return: Option>, + drain: DrainState, tx: broadcast::Sender>, - stop_tx: Option>, - drain_handle: Option>, } impl Watcher { @@ -72,11 +137,8 @@ impl Watcher { event_buffer_size, buffering_duration, inner, - event_rx: Some(event_rx), - rx_return: None, + drain: DrainState::new(event_rx), tx, - stop_tx: None, - drain_handle: None, }) } @@ -136,51 +198,33 @@ impl Watcher { } pub fn stop(&mut self) { - if let Some(tx) = self.stop_tx.take() { - let _ = tx.send(()); - } - - self.recover_event_rx(); + let runtime = pyo3_async_runtimes::tokio::get_runtime(); + self.drain.stop_and_recover(runtime); let (new_tx, _rx) = broadcast::channel::>(self.event_buffer_size); self.tx = new_tx; } - fn recover_event_rx(&mut self) { - if let Some(handle) = self.drain_handle.take() { - handle.abort(); - } - - if let Some(mut rx_return) = self.rx_return.take() { - if let Ok(rx) = rx_return.try_recv() { - self.event_rx = Some(rx); - } - } - } - pub fn start_drain(&mut self, tick_duration: Duration, event_filter: Option) { - if let Some(tx) = self.stop_tx.take() { - let _ = tx.send(()); - } + let runtime = pyo3_async_runtimes::tokio::get_runtime(); + self.drain.stop_and_recover(runtime); - self.recover_event_rx(); - - let mut event_rx = match self.event_rx.take() { + let mut event_rx = match self.drain.take_event_rx() { Some(rx) => rx, - None => return, + None => { + eprintln!("notifykit: event receiver lost, drain will not start"); + return; + } }; let (stop_tx, mut stop_rx) = oneshot::channel(); - self.stop_tx = Some(stop_tx); - let (rx_return_tx, rx_return_rx) = oneshot::channel(); - self.rx_return = Some(rx_return_rx); let tx = self.tx.clone(); let debug = self.debug; let buffering_duration = self.buffering_duration; - self.drain_handle = Some(pyo3_async_runtimes::tokio::get_runtime().spawn(async move { + let handle = runtime.spawn(async move { let mut processor = BatchProcessor::new(buffering_duration); let mut ticker = time::interval(tick_duration); @@ -225,7 +269,9 @@ impl Watcher { } let _ = rx_return_tx.send(event_rx); - })); + }); + + self.drain.set_drain(handle, stop_tx, rx_return_rx); } pub fn subscribe(&self) -> broadcast::Receiver> { @@ -287,3 +333,184 @@ fn create_event(event: &RawEvent) -> Option { } }) } + +#[cfg(test)] +mod tests { + use super::*; + use notify::Event as NotifyEvent; + + fn build_runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap() + } + + /// Helper: spawn a minimal drain task that moves the receiver in + /// and returns it via oneshot when the stop signal is received. + fn spawn_drain( + runtime: &tokio::runtime::Runtime, + drain: &mut DrainState, + ) { + let mut event_rx = drain.take_event_rx().expect("event_rx should be available"); + + let (stop_tx, mut stop_rx) = oneshot::channel(); + let (rx_return_tx, rx_return_rx) = oneshot::channel(); + + let handle = runtime.spawn(async move { + // Minimal drain loop: just wait for stop signal + loop { + tokio::select! { + _ = &mut stop_rx => break, + _ = tokio::time::sleep(Duration::from_millis(10)) => { + // drain events to keep channel from filling + while event_rx.try_recv().is_ok() {} + } + } + } + let _ = rx_return_tx.send(event_rx); + }); + + drain.set_drain(handle, stop_tx, rx_return_rx); + } + + #[test] + fn new_drain_state_has_receiver() { + let (_tx, rx) = mpsc::channel::(16); + let drain = DrainState::new(rx); + + assert!(drain.event_rx.is_some()); + assert!(drain.rx_return.is_none()); + assert!(drain.stop_tx.is_none()); + assert!(drain.drain_handle.is_none()); + } + + #[test] + fn stop_and_recover_returns_receiver() { + let rt = build_runtime(); + let (_tx, rx) = mpsc::channel::(16); + let mut drain = DrainState::new(rx); + + // Start a drain task + spawn_drain(&rt, &mut drain); + assert!(drain.event_rx.is_none(), "receiver should be moved into task"); + + // Stop and recover + drain.stop_and_recover(&rt); + assert!(drain.event_rx.is_some(), "receiver should be recovered after stop"); + } + + #[test] + fn stop_and_recover_without_active_drain_is_noop() { + let rt = build_runtime(); + let (_tx, rx) = mpsc::channel::(16); + let mut drain = DrainState::new(rx); + + // No drain running — should not panic + drain.stop_and_recover(&rt); + assert!(drain.event_rx.is_some()); + } + + #[test] + fn multiple_stop_restart_cycles() { + let rt = build_runtime(); + let (tx, rx) = mpsc::channel::(16); + let mut drain = DrainState::new(rx); + + for i in 0..5 { + spawn_drain(&rt, &mut drain); + assert!(drain.event_rx.is_none(), "cycle {i}: receiver should be in task"); + + // Send an event while drain is running + let event = NotifyEvent::default(); + tx.try_send((Instant::now(), Ok(event))).ok(); + + drain.stop_and_recover(&rt); + assert!( + drain.event_rx.is_some(), + "cycle {i}: receiver should be recovered" + ); + } + } + + #[test] + fn events_survive_stop_restart_cycle() { + let rt = build_runtime(); + let (tx, rx) = mpsc::channel::(16); + let mut drain = DrainState::new(rx); + + // Start and stop drain + spawn_drain(&rt, &mut drain); + drain.stop_and_recover(&rt); + + // Send event after recovery + let event = NotifyEvent::default(); + tx.try_send((Instant::now(), Ok(event))).unwrap(); + + // Receiver should still work + let event_rx = drain.event_rx.as_mut().unwrap(); + let received = event_rx.try_recv(); + assert!(received.is_ok(), "should receive event after stop/restart"); + } + + #[test] + fn channel_stays_connected_after_recovery() { + let rt = build_runtime(); + let (tx, rx) = mpsc::channel::(16); + let mut drain = DrainState::new(rx); + + // Multiple cycles with events in between + for _ in 0..3 { + spawn_drain(&rt, &mut drain); + drain.stop_and_recover(&rt); + } + + // The sender should still be connected to the recovered receiver + let event = NotifyEvent::default(); + assert!( + tx.try_send((Instant::now(), Ok(event))).is_ok(), + "sender should still be connected after multiple cycles" + ); + + let event_rx = drain.event_rx.as_mut().unwrap(); + assert!(event_rx.try_recv().is_ok()); + } + + #[test] + fn recover_handles_task_that_dropped_return_sender() { + let rt = build_runtime(); + let (_tx, rx) = mpsc::channel::(16); + let mut drain = DrainState::new(rx); + + let event_rx = drain.take_event_rx().unwrap(); + + let (stop_tx, mut stop_rx) = oneshot::channel(); + let (rx_return_tx, rx_return_rx) = oneshot::channel(); + + // Spawn a task that drops the return sender without sending + let handle = rt.spawn(async move { + let _rx = event_rx; // take ownership + let _ = &mut stop_rx; + drop(rx_return_tx); // simulate panic/early exit + }); + + drain.set_drain(handle, stop_tx, rx_return_rx); + drain.stop_and_recover(&rt); + + // Receiver is lost — this is the expected failure mode + assert!( + drain.event_rx.is_none(), + "receiver should be lost when task drops return sender" + ); + } + + #[test] + fn send_stop_signal_is_idempotent() { + let (_tx, rx) = mpsc::channel::(16); + let mut drain = DrainState::new(rx); + + // No stop_tx set — should not panic + drain.send_stop_signal(); + drain.send_stop_signal(); + } +}