sui_rpc/client/ledger_streams/
observability.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use tonic::Code;
5use tonic::Status;
6
7use super::super::Result;
8
9/// The ledger protocol family associated with an observability event.
10#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
11#[non_exhaustive]
12pub enum LedgerStreamFamily {
13    Checkpoint,
14    Transaction,
15    Event,
16}
17
18/// The RPC operation associated with an observability event.
19#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
20#[non_exhaustive]
21pub enum LedgerStreamOperation {
22    List,
23    GetServiceInfo,
24    Subscribe,
25}
26
27/// List operations use `List`; stream operations use the remaining stages.
28#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
29#[non_exhaustive]
30pub enum LedgerStreamStage {
31    List,
32    InitialReplay,
33    LiveTipStartup,
34    PollingBaseline,
35    PollingTail,
36    GapRecovery,
37    LiveSubscription,
38}
39
40/// Observability events emitted by ledger streams.
41///
42/// Label metrics by family, operation, stage, and status code. Record counts and durations as
43/// values; keep status messages and metadata in logs.
44#[derive(Clone, Debug)]
45#[non_exhaustive]
46pub enum LedgerStreamEvent {
47    /// A generated RPC future resolved.
48    ///
49    /// For List and Subscribe, `elapsed` measures dispatch-to-response-headers latency, not
50    /// response-body processing or subscription lifetime.
51    #[non_exhaustive]
52    RpcResponse {
53        family: LedgerStreamFamily,
54        operation: LedgerStreamOperation,
55        stage: LedgerStreamStage,
56        code: Code,
57        elapsed: Duration,
58    },
59    /// A classified transient failure scheduled another attempt.
60    #[non_exhaustive]
61    RetryScheduled {
62        family: LedgerStreamFamily,
63        operation: LedgerStreamOperation,
64        stage: LedgerStreamStage,
65        status: Status,
66        consecutive_failures: u32,
67        delay: Duration,
68    },
69    /// Operation progress ended a period of consecutive transient failures.
70    #[non_exhaustive]
71    RetryRecovered {
72        family: LedgerStreamFamily,
73        operation: LedgerStreamOperation,
74        started_in: LedgerStreamStage,
75        consecutive_failures: u32,
76        elapsed: Duration,
77    },
78    /// An established Subscribe response body returned an error or ended unexpectedly.
79    #[non_exhaustive]
80    SubscriptionStreamInterrupted {
81        family: LedgerStreamFamily,
82        stage: LedgerStreamStage,
83        status: Status,
84    },
85    /// The stream began a bounded List replay for a subscription gap.
86    #[non_exhaustive]
87    GapRecoveryStarted { family: LedgerStreamFamily },
88    /// List gap buffering reached its configured item-bearing frame limit.
89    #[non_exhaustive]
90    SubscriptionBufferLimitReached {
91        family: LedgerStreamFamily,
92        buffered_items: usize,
93        limit: usize,
94    },
95    /// A non-retryable terminal error that will be yielded as the next stream item.
96    #[non_exhaustive]
97    TerminalError {
98        family: LedgerStreamFamily,
99        status: Status,
100    },
101}
102
103/// Observability events emitted by finite List operations.
104///
105/// Label metrics by family and status code. Record counts and durations as values; keep status
106/// messages and metadata in logs.
107#[derive(Clone, Debug)]
108#[non_exhaustive]
109pub enum ListEvent {
110    /// A generated List RPC future resolved.
111    ///
112    /// `elapsed` measures dispatch-to-response-headers latency, not response-body processing.
113    #[non_exhaustive]
114    RpcResponse {
115        family: LedgerStreamFamily,
116        code: Code,
117        elapsed: Duration,
118    },
119    /// A classified transient failure scheduled another attempt.
120    #[non_exhaustive]
121    RetryScheduled {
122        family: LedgerStreamFamily,
123        status: Status,
124        consecutive_failures: u32,
125        delay: Duration,
126    },
127    /// Operation progress ended a period of consecutive transient failures.
128    #[non_exhaustive]
129    RetryRecovered {
130        family: LedgerStreamFamily,
131        consecutive_failures: u32,
132        elapsed: Duration,
133    },
134    /// A non-retryable terminal error that will be yielded as the next stream item.
135    #[non_exhaustive]
136    TerminalError {
137        family: LedgerStreamFamily,
138        status: Status,
139    },
140}
141
142impl ListEvent {
143    pub(super) fn from_stream_event(event: LedgerStreamEvent) -> Option<Self> {
144        match event {
145            LedgerStreamEvent::RpcResponse {
146                family,
147                operation: LedgerStreamOperation::List,
148                code,
149                elapsed,
150                ..
151            } => Some(Self::RpcResponse {
152                family,
153                code,
154                elapsed,
155            }),
156            LedgerStreamEvent::RetryScheduled {
157                family,
158                operation: LedgerStreamOperation::List,
159                status,
160                consecutive_failures,
161                delay,
162                ..
163            } => Some(Self::RetryScheduled {
164                family,
165                status,
166                consecutive_failures,
167                delay,
168            }),
169            LedgerStreamEvent::RetryRecovered {
170                family,
171                operation: LedgerStreamOperation::List,
172                consecutive_failures,
173                elapsed,
174                ..
175            } => Some(Self::RetryRecovered {
176                family,
177                consecutive_failures,
178                elapsed,
179            }),
180            LedgerStreamEvent::TerminalError { family, status } => {
181                Some(Self::TerminalError { family, status })
182            }
183            _ => None,
184        }
185    }
186}
187
188pub(super) type ListObserver = Arc<dyn Fn(ListEvent) + Send + Sync + 'static>;
189
190pub(super) type LedgerStreamObserver = Arc<dyn Fn(LedgerStreamEvent) + Send + Sync + 'static>;
191
192#[derive(Clone, Default)]
193pub(super) struct LedgerStreamObservability {
194    observer: Option<LedgerStreamObserver>,
195}
196
197impl LedgerStreamObservability {
198    pub(super) fn new(observer: Option<LedgerStreamObserver>) -> Self {
199        Self { observer }
200    }
201
202    pub(super) fn emit(&self, event: impl FnOnce() -> LedgerStreamEvent) {
203        if let Some(observer) = &self.observer {
204            observer(event());
205        }
206    }
207
208    pub(super) fn start_timer(&self) -> Option<tokio::time::Instant> {
209        self.observer.as_ref().map(|_| tokio::time::Instant::now())
210    }
211
212    pub(super) fn emit_rpc_response<T>(
213        &self,
214        started_at: Option<tokio::time::Instant>,
215        family: LedgerStreamFamily,
216        operation: LedgerStreamOperation,
217        stage: LedgerStreamStage,
218        result: &Result<T>,
219    ) {
220        let Some(started_at) = started_at else {
221            return;
222        };
223        let code = result
224            .as_ref()
225            .map_or_else(|status| status.code(), |_| Code::Ok);
226        self.emit(|| LedgerStreamEvent::RpcResponse {
227            family,
228            operation,
229            stage,
230            code,
231            elapsed: started_at.elapsed(),
232        });
233    }
234}
sui_rpc/client/ledger_streams/observability.rs

sui_rpc/client/ledger_streams/
observability.rs
