pgwire_replication/client/
metrics.rs

1//! Observability counters for the replication worker.
2
3use std::sync::atomic::{AtomicU64, Ordering};
4
5use crate::lsn::Lsn;
6
7/// Pull-based metrics for a replication stream.
8///
9/// Shared between the background worker (which updates the counters) and the
10/// consumer (which reads them via [`ReplicationClient::metrics`]). All updates
11/// use relaxed atomics, so the worker's hot path never blocks and reads are
12/// cheap. Counters are monotonic for the lifetime of the connection.
13///
14/// [`ReplicationClient::metrics`]: crate::client::ReplicationClient::metrics
15#[derive(Debug, Default)]
16pub struct ReplicationMetrics {
17    events_forwarded: AtomicU64,
18    feedback_sent: AtomicU64,
19    keepalive_replies: AtomicU64,
20    stall_count: AtomicU64,
21    stall_micros_total: AtomicU64,
22    last_applied_lsn: AtomicU64,
23    last_wal_end: AtomicU64,
24}
25
26impl ReplicationMetrics {
27    /// Replication events successfully forwarded to the consumer channel.
28    #[inline]
29    pub fn events_forwarded(&self) -> u64 {
30        self.events_forwarded.load(Ordering::Relaxed)
31    }
32
33    /// Standby-status (feedback) messages sent to the server.
34    #[inline]
35    pub fn feedback_sent(&self) -> u64 {
36        self.feedback_sent.load(Ordering::Relaxed)
37    }
38
39    /// Feedback messages sent specifically in reply to a reply-requested
40    /// keepalive. A subset of [`feedback_sent`](Self::feedback_sent).
41    #[inline]
42    pub fn keepalive_replies(&self) -> u64 {
43        self.keepalive_replies.load(Ordering::Relaxed)
44    }
45
46    /// Number of times forwarding an event had to wait for channel capacity
47    /// (i.e. the consumer applied backpressure).
48    #[inline]
49    pub fn stall_count(&self) -> u64 {
50        self.stall_count.load(Ordering::Relaxed)
51    }
52
53    /// Cumulative time, in microseconds, the worker spent waiting for channel
54    /// capacity while backpressured. Feedback continues throughout this time.
55    #[inline]
56    pub fn stall_micros_total(&self) -> u64 {
57        self.stall_micros_total.load(Ordering::Relaxed)
58    }
59
60    /// Most recent applied LSN reported to the server via feedback.
61    #[inline]
62    pub fn last_applied_lsn(&self) -> Lsn {
63        Lsn::from_u64(self.last_applied_lsn.load(Ordering::Relaxed))
64    }
65
66    /// Most recent server WAL-end position observed on an incoming message.
67    #[inline]
68    pub fn last_wal_end(&self) -> Lsn {
69        Lsn::from_u64(self.last_wal_end.load(Ordering::Relaxed))
70    }
71
72    // ── worker-side updates (crate-internal) ──
73
74    #[inline]
75    pub(crate) fn record_event_forwarded(&self) {
76        self.events_forwarded.fetch_add(1, Ordering::Relaxed);
77    }
78
79    #[inline]
80    pub(crate) fn record_feedback_sent(&self, applied: Lsn) {
81        self.feedback_sent.fetch_add(1, Ordering::Relaxed);
82        self.last_applied_lsn
83            .store(applied.as_u64(), Ordering::Relaxed);
84    }
85
86    #[inline]
87    pub(crate) fn record_keepalive_reply(&self) {
88        self.keepalive_replies.fetch_add(1, Ordering::Relaxed);
89    }
90
91    /// Record that a backpressure stall has begun. Counted at the start (not
92    /// the end) so an in-progress stall is observable while the worker is still
93    /// parked awaiting capacity.
94    #[inline]
95    pub(crate) fn record_stall_begin(&self) {
96        self.stall_count.fetch_add(1, Ordering::Relaxed);
97    }
98
99    /// Record the duration of a stall once capacity is regained.
100    #[inline]
101    pub(crate) fn record_stall_end(&self, micros: u64) {
102        self.stall_micros_total.fetch_add(micros, Ordering::Relaxed);
103    }
104
105    #[inline]
106    pub(crate) fn set_last_wal_end(&self, wal_end: Lsn) {
107        self.last_wal_end.store(wal_end.as_u64(), Ordering::Relaxed);
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114
115    #[test]
116    fn counters_start_at_zero() {
117        let m = ReplicationMetrics::default();
118        assert_eq!(m.events_forwarded(), 0);
119        assert_eq!(m.feedback_sent(), 0);
120        assert_eq!(m.keepalive_replies(), 0);
121        assert_eq!(m.stall_count(), 0);
122        assert_eq!(m.stall_micros_total(), 0);
123        assert_eq!(m.last_applied_lsn(), Lsn::ZERO);
124        assert_eq!(m.last_wal_end(), Lsn::ZERO);
125    }
126
127    #[test]
128    fn recorders_update_their_counters() {
129        let m = ReplicationMetrics::default();
130
131        m.record_event_forwarded();
132        m.record_event_forwarded();
133        assert_eq!(m.events_forwarded(), 2);
134
135        m.record_feedback_sent(Lsn::from_u64(42));
136        assert_eq!(m.feedback_sent(), 1);
137        assert_eq!(m.last_applied_lsn(), Lsn::from_u64(42));
138
139        m.record_keepalive_reply();
140        assert_eq!(m.keepalive_replies(), 1);
141
142        m.record_stall_begin();
143        assert_eq!(m.stall_count(), 1);
144        assert_eq!(m.stall_micros_total(), 0);
145
146        m.record_stall_end(100);
147        m.record_stall_end(50);
148        assert_eq!(m.stall_micros_total(), 150);
149        // Duration accrues without re-counting the stall.
150        assert_eq!(m.stall_count(), 1);
151
152        m.set_last_wal_end(Lsn::from_u64(7));
153        assert_eq!(m.last_wal_end(), Lsn::from_u64(7));
154    }
155}
pgwire_replication/client/metrics.rs

pgwire_replication/client/
metrics.rs
