connpool/observability/
events.rs

1//! `PoolListener`: observability hooks, and the internal dispatch group that isolates panics.
2
3use std::error::Error;
4use std::panic::{AssertUnwindSafe, catch_unwind};
5use std::sync::Arc;
6
7use crate::core::connection::{CloseReason, PooledConnection};
8use crate::resilience::breaker::BreakerState;
9
10/// Observability hooks for metrics, tracing, logging. Implement the ones you care about; all
11/// default to no-ops. Hooks run synchronously on the calling thread, outside the pool's lock,
12/// so keep them fast. Panics raised by a listener are caught and logged — a broken metrics
13/// exporter must never break the pool.
14pub trait PoolListener<T>: Send + Sync {
15    /// A new connection was created and registered.
16    fn on_connection_created(&self, _conn: &PooledConnection<T>, _duration: f64) {}
17    /// A connection was closed, with the reason it left the pool.
18    fn on_connection_closed(&self, _conn: &PooledConnection<T>, _reason: CloseReason) {}
19    /// A `create` attempt failed (fires per attempt inside the retry loop).
20    fn on_create_failed(&self, _error: &(dyn Error + 'static)) {}
21    /// A connection was acquired, with how long the caller waited.
22    fn on_acquired(&self, _conn: &PooledConnection<T>, _wait_time: f64) {}
23    /// A connection was released, with how long it was held.
24    fn on_released(&self, _conn: &PooledConnection<T>, _hold_time: f64) {}
25    /// An acquire timed out, with how long the caller waited.
26    fn on_acquire_timeout(&self, _waited: f64) {}
27    /// A borrowed connection was flagged as possibly leaked (held too long).
28    fn on_connection_leak(&self, _conn: &PooledConnection<T>, _held_for: f64) {}
29    /// The circuit breaker changed state.
30    fn on_breaker_state_change(&self, _old: BreakerState, _new: BreakerState) {}
31}
32
33/// An internal group of listeners with panic-isolating dispatch.
34pub(crate) struct Listeners<T> {
35    listeners: Vec<Arc<dyn PoolListener<T>>>,
36}
37
38impl<T> Listeners<T> {
39    pub(crate) fn new(listeners: Vec<Arc<dyn PoolListener<T>>>) -> Self {
40        Self { listeners }
41    }
42
43    /// Calls `f` for every listener, catching and logging any panic.
44    pub(crate) fn safe<F>(&self, f: F)
45    where
46        F: Fn(&dyn PoolListener<T>),
47    {
48        for listener in &self.listeners {
49            let l = listener.as_ref();
50            if catch_unwind(AssertUnwindSafe(|| f(l))).is_err() {
51                eprintln!("connpool: a pool listener panicked (ignored)");
52            }
53        }
54    }
55
56    /// Dispatches a breaker state change to every listener (a named entry point so the pool's
57    /// breaker callback is a plain call rather than a nested closure).
58    pub(crate) fn breaker_change(&self, old: BreakerState, new: BreakerState) {
59        self.safe(|l| l.on_breaker_state_change(old, new));
60    }
61}
62
63#[cfg(test)]
64mod tests {
65    use super::*;
66    use std::sync::atomic::{AtomicU32, Ordering};
67
68    struct Counting(Arc<AtomicU32>);
69    impl PoolListener<u32> for Counting {
70        fn on_acquire_timeout(&self, _waited: f64) {
71            self.0.fetch_add(1, Ordering::SeqCst);
72        }
73    }
74
75    struct Panicky;
76    impl PoolListener<u32> for Panicky {
77        fn on_acquire_timeout(&self, _waited: f64) {
78            panic!("boom");
79        }
80    }
81
82    #[test]
83    fn default_hooks_are_noops() {
84        struct Noop;
85        impl PoolListener<u32> for Noop {}
86        let n = Noop;
87        let pc = PooledConnection::new(1u32, 0.0, None);
88        // Exercise every default method.
89        n.on_connection_created(&pc, 1.0);
90        n.on_connection_closed(&pc, CloseReason::Idle);
91        let e: Box<dyn Error + Send + Sync> = "x".into();
92        n.on_create_failed(&*e);
93        n.on_acquired(&pc, 1.0);
94        n.on_released(&pc, 1.0);
95        n.on_acquire_timeout(1.0);
96        n.on_connection_leak(&pc, 1.0);
97        n.on_breaker_state_change(BreakerState::Closed, BreakerState::Open);
98    }
99
100    #[test]
101    fn dispatch_isolates_panics_and_reaches_all() {
102        let count = Arc::new(AtomicU32::new(0));
103        let group: Listeners<u32> =
104            Listeners::new(vec![Arc::new(Panicky), Arc::new(Counting(count.clone()))]);
105        group.safe(|l| l.on_acquire_timeout(2.0));
106        // The counting listener still ran despite the panicky one.
107        assert_eq!(count.load(Ordering::SeqCst), 1);
108    }
109}
connpool/observability/events.rs

connpool/observability/
events.rs
