diff --git a/crates/lib/pager/src/client/pager.rs b/crates/lib/pager/src/client/pager.rs index 91c17b54b9..c4fb9ec7db 100644 --- a/crates/lib/pager/src/client/pager.rs +++ b/crates/lib/pager/src/client/pager.rs @@ -16,15 +16,8 @@ use crate::worker::pager_worker_thread::PagerWorkerThread; /// - **Immediate**: `send_page_immediately()` sends inline (blocks until API responds). /// - **Queued**: `enqueue_page()` pushes to a background worker (non-blocking). /// -/// Build via `PagerBuilder`: -/// ```ignore -/// let pager = PagerBuilder::new() -/// .application_name("seedance2-pro-job".to_string()) -/// .environment("production".to_string()) -/// .rootly(api_key) -/// .notification_target("User".to_string(), user_id) -/// .build()?; -/// ``` +/// Build an instance via `PagerBuilder`. +/// pub struct Pager { client: PagerClient, queue: Option, @@ -33,6 +26,8 @@ pub struct Pager { } impl Pager { + + /// External users will need to create instances via `PagerBuilder`. pub(crate) fn new( client: PagerClient, queue: Option, @@ -47,7 +42,9 @@ impl Pager { &self, notification: NotificationDetails, ) -> Result { - self.client.send_page(¬ification).await.map_err(PagerError::Client) + self.client.send_page(¬ification) + .await + .map_err(PagerError::Client) } /// Enqueue a page to be sent by the background worker thread. @@ -61,7 +58,8 @@ impl Pager { let queue = self.queue.as_ref() .ok_or(PagerServiceError::WorkerNotAvailable)?; - let dropped = queue.push(notification); + let dropped = queue.push(notification) + .map_err(PagerError::Service)?; if let Some(dropped_summary) = dropped { // We still enqueued the new item — just warn that we lost an old one. diff --git a/crates/lib/pager/src/error/pager_service_error.rs b/crates/lib/pager/src/error/pager_service_error.rs index 1887fb8f56..798b987f88 100644 --- a/crates/lib/pager/src/error/pager_service_error.rs +++ b/crates/lib/pager/src/error/pager_service_error.rs @@ -9,6 +9,9 @@ pub enum PagerServiceError { /// The message queue is full and the oldest item was dropped. QueueFull { dropped_summary: String }, + + /// A mutex lock was poisoned (another thread panicked while holding it). + MutexPoisoned(String), } impl Error for PagerServiceError {} @@ -20,6 +23,7 @@ impl Display for PagerServiceError { Self::QueueFull { dropped_summary } => { write!(f, "Pager queue is full. Dropped oldest item: {}", dropped_summary) } + Self::MutexPoisoned(msg) => write!(f, "Pager mutex poisoned: {}", msg), } } } diff --git a/crates/lib/pager/src/worker/pager_worker_message_queue.rs b/crates/lib/pager/src/worker/pager_worker_message_queue.rs index cda6f3f00f..97e25a8743 100644 --- a/crates/lib/pager/src/worker/pager_worker_message_queue.rs +++ b/crates/lib/pager/src/worker/pager_worker_message_queue.rs @@ -3,6 +3,7 @@ use std::sync::{Arc, Mutex, Condvar}; use log::warn; +use crate::error::pager_service_error::PagerServiceError; use crate::notification::notification_details::NotificationDetails; const DEFAULT_MAX_SIZE: usize = 256; @@ -33,9 +34,11 @@ impl PagerWorkerMessageQueue { /// Push a notification onto the queue. /// /// If the queue is full, the oldest item is dropped and a warning is logged. - /// Returns the dropped item's summary if one was evicted. - pub fn push(&self, notification: NotificationDetails) -> Option { - let mut queue = self.inner.lock().unwrap(); + /// Returns `Ok(Some(summary))` if an old item was evicted, `Ok(None)` otherwise. + pub fn push(&self, notification: NotificationDetails) -> Result, PagerServiceError> { + let mut queue = self.inner.lock() + .map_err(|e| PagerServiceError::MutexPoisoned(format!("push: {}", e)))?; + let dropped = if queue.len() >= self.max_size { let old = queue.pop_front(); let dropped_summary = old.as_ref().map(|n| n.summary.clone()); @@ -46,27 +49,33 @@ impl PagerWorkerMessageQueue { } else { None }; + queue.push_back(notification); self.condvar.notify_one(); - dropped + Ok(dropped) } /// Block until at least one item is available, then drain all items. /// - /// Returns `None` if the queue was poisoned (shouldn't happen in practice). - pub fn wait_and_drain(&self) -> Option> { - let mut queue = self.inner.lock().ok()?; + /// Returns an error if the mutex is poisoned. + pub fn wait_and_drain(&self) -> Result, PagerServiceError> { + let mut queue = self.inner.lock() + .map_err(|e| PagerServiceError::MutexPoisoned(format!("wait_and_drain lock: {}", e)))?; + while queue.is_empty() { - queue = self.condvar.wait(queue).ok()?; + queue = self.condvar.wait(queue) + .map_err(|e| PagerServiceError::MutexPoisoned(format!("wait_and_drain wait: {}", e)))?; } + let items: Vec = queue.drain(..).collect(); - Some(items) + Ok(items) } /// Non-blocking drain of all currently queued items. - pub fn drain_available(&self) -> Vec { - let mut queue = self.inner.lock().unwrap(); - queue.drain(..).collect() + pub fn drain_available(&self) -> Result, PagerServiceError> { + let mut queue = self.inner.lock() + .map_err(|e| PagerServiceError::MutexPoisoned(format!("drain_available: {}", e)))?; + Ok(queue.drain(..).collect()) } /// Wake up any thread blocked on `wait_and_drain()`. @@ -76,12 +85,14 @@ impl PagerWorkerMessageQueue { self.condvar.notify_all(); } - pub fn len(&self) -> usize { - self.inner.lock().unwrap().len() + pub fn len(&self) -> Result { + let queue = self.inner.lock() + .map_err(|e| PagerServiceError::MutexPoisoned(format!("len: {}", e)))?; + Ok(queue.len()) } - pub fn is_empty(&self) -> bool { - self.len() == 0 + pub fn is_empty(&self) -> Result { + Ok(self.len()? == 0) } } diff --git a/crates/lib/pager/src/worker/pager_worker_thread.rs b/crates/lib/pager/src/worker/pager_worker_thread.rs index 772aa970e7..dd90f255b8 100644 --- a/crates/lib/pager/src/worker/pager_worker_thread.rs +++ b/crates/lib/pager/src/worker/pager_worker_thread.rs @@ -54,9 +54,9 @@ impl PagerWorkerThread { while !self.shutdown.load(Ordering::Relaxed) { // Block until items are available (or we're woken up for shutdown). let items = match self.queue.wait_and_drain() { - Some(items) => items, - None => { - warn!("Pager worker queue lock was poisoned. Retrying in 5s."); + Ok(items) => items, + Err(err) => { + error!("Pager worker queue error: {}. Retrying in 5s.", err); tokio::time::sleep(Duration::from_secs(5)).await; continue; } @@ -94,12 +94,17 @@ impl PagerWorkerThread { } // Drain any remaining items on shutdown. - let remaining = self.queue.drain_available(); - if !remaining.is_empty() { - warn!( - "Pager worker shutting down with {} unsent notification(s) in queue.", - remaining.len() - ); + match self.queue.drain_available() { + Ok(remaining) if !remaining.is_empty() => { + warn!( + "Pager worker shutting down with {} unsent notification(s) in queue.", + remaining.len() + ); + } + Err(err) => { + error!("Pager worker could not drain queue on shutdown: {}", err); + } + _ => {} } info!("Pager worker thread stopped.");