improvements

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<SharedMessageQueue>,
@@ -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<SharedMessageQueue>,
@@ -47,7 +42,9 @@ impl Pager {
     &self,
     notification: NotificationDetails,
   ) -> Result<PageSentResult, PagerError> {
-    self.client.send_page(&notification).await.map_err(PagerError::Client)
+    self.client.send_page(&notification)
+        .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.

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),
     }
   }
 }

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<String> {
-    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<Option<String>, 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<Vec<NotificationDetails>> {
-    let mut queue = self.inner.lock().ok()?;
+  /// Returns an error if the mutex is poisoned.
+  pub fn wait_and_drain(&self) -> Result<Vec<NotificationDetails>, 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<NotificationDetails> = queue.drain(..).collect();
-    Some(items)
+    Ok(items)
   }
 
   /// Non-blocking drain of all currently queued items.
-  pub fn drain_available(&self) -> Vec<NotificationDetails> {
-    let mut queue = self.inner.lock().unwrap();
-    queue.drain(..).collect()
+  pub fn drain_available(&self) -> Result<Vec<NotificationDetails>, 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<usize, PagerServiceError> {
+    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<bool, PagerServiceError> {
+    Ok(self.len()? == 0)
   }
 }
 

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.");