pgwire_replication/client/
feedback.rs

1//! Observing standby status feedback: how long an acknowledged position
2//! waited before the server was told about it.
3
4use std::fmt;
5use std::sync::Arc;
6use std::time::Duration;
7
8/// A standby status update that told the server about a newer applied
9/// position, reported to a [`FeedbackObserver`] after the write succeeded.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11#[non_exhaustive]
12pub struct FeedbackSent {
13    /// From the oldest raise of the applied position that this update is the
14    /// first to report, until the update was written. Raises coalesced into
15    /// one update keep the oldest, so this is how long an acknowledged
16    /// position was withheld from the server, including across a failed
17    /// write or a new session that carried it over
18    /// ([`ReplicationClient::carry_feedback_pending`](crate::ReplicationClient::carry_feedback_pending)).
19    pub pending_for: Duration,
20}
21
22/// Called after each successful standby status update that advanced the
23/// position reported to the server; never for one that repeated it. Runs on
24/// the replication worker, so it must be cheap and must not block.
25#[derive(Clone)]
26pub struct FeedbackObserver(Arc<dyn Fn(FeedbackSent) + Send + Sync>);
27
28impl FeedbackObserver {
29    pub fn new(observer: impl Fn(FeedbackSent) + Send + Sync + 'static) -> Self {
30        Self(Arc::new(observer))
31    }
32
33    pub(crate) fn observe(&self, sent: FeedbackSent) {
34        (self.0)(sent)
35    }
36}
37
38impl fmt::Debug for FeedbackObserver {
39    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
40        f.write_str("FeedbackObserver(..)")
41    }
42}
43
44/// Two observers are equal when they are the same callback.
45impl PartialEq for FeedbackObserver {
46    fn eq(&self, other: &Self) -> bool {
47        Arc::ptr_eq(&self.0, &other.0)
48    }
49}
50
51impl Eq for FeedbackObserver {}
52
53// The callback is only ever called by the replication worker
54// (`observe` is crate-private), so a reference held across a
55// `catch_unwind` boundary cannot observe state that a panic left broken.
56// This keeps `ReplicationConfig` unwind safe, as it was before observers.
57impl std::panic::UnwindSafe for FeedbackObserver {}
58impl std::panic::RefUnwindSafe for FeedbackObserver {}
pgwire_replication/client/feedback.rs

pgwire_replication/client/
feedback.rs
