▮ REFACTORED PATCH
// 1) Hold the receiver; use a BOUNDED channel (backpressure).
pub struct WatcherManager {
watchers: Arc<RwLock<HashMap<u32, WatcherState>>>,
consumer_tx: mpsc::Sender<WatcherEvent>, // bounded
join: Arc<Mutex<Vec<JoinHandle<()>>>>, // track tasks for clean shutdown
shutdown_token: CancellationToken,
}
impl WatcherManager {
pub fn new(bound: usize) -> (Self, mpsc::Receiver<WatcherEvent>) {
let (tx, rx) = mpsc::channel::<WatcherEvent>(bound);
(Self { watchers: Arc::new(RwLock::new(HashMap::new())),
consumer_tx: tx, join: Arc::new(Mutex::new(Vec::new())),
shutdown_token: CancellationToken::new() }, rx)
}
pub async fn shutdown(&self) {
self.shutdown_token.cancel();
let mut handles = self.join.lock().await;
for h in handles.drain(..) { let _ = tokio::time::timeout(Duration::from_secs(1), h).await; }
}
}
// 2) Consumer that actually drains the bounded channel:
async fn consumer_task(mut rx: mpsc::Receiver<WatcherEvent>, shutdown: CancellationToken) {
loop {
tokio::select! {
_ = shutdown.cancelled() => break,
ev = rx.recv() => match ev {
Some(WatcherEvent::Items(id, items)) => { /* aggregate */ }
None => break,
}
}
}
}
// 3) Deterministic unhealthy test — inject failures instead of relying on rand:
async fn failing_fetch(_id: u32) -> Result<Vec<String>, FetchError> { Err(FetchError::MockFetchFailed) }