sui_rpc/client/ledger_streams/
types.rs

1use std::fmt;
2use std::num::NonZeroUsize;
3use std::sync::Arc;
4use std::time::Duration;
5
6use super::observability::LedgerStreamEvent;
7use super::observability::LedgerStreamObserver;
8use super::observability::ListEvent;
9use super::observability::ListObserver;
10use crate::proto::sui::rpc::v2::Checkpoint;
11use crate::proto::sui::rpc::v2::Event;
12use crate::proto::sui::rpc::v2::EventFilter;
13use crate::proto::sui::rpc::v2::ExecutedTransaction;
14use crate::proto::sui::rpc::v2::TransactionFilter;
15use prost::bytes::Bytes;
16use prost_types::FieldMask;
17/// How a stream follows the ledger.
18#[non_exhaustive]
19#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
20pub enum Delivery {
21    /// Follows the tip through SubscriptionService, repairing gaps through List.
22    #[default]
23    Subscribe,
24    /// Follows the ledger via the List APIs, useful on endpoints without SubscriptionService.
25    Poll,
26}
27
28/// Starting position for a checkpoint stream.
29#[non_exhaustive]
30#[derive(Clone, Debug, Default, PartialEq, Eq)]
31pub enum CheckpointStreamStart {
32    /// Begins at the chain tip (live for `Subscribe`, indexed for `Poll`) and emits an initial
33    /// progress frame before subsequent items.
34    #[default]
35    Tip,
36    /// Starts at and includes the checkpoint with this sequence number.
37    Checkpoint(u64),
38}
39
40/// Starting position for a transaction stream.
41#[non_exhaustive]
42#[derive(Clone, Debug, Default, PartialEq, Eq)]
43pub enum TransactionStreamStart {
44    /// Begins at the chain tip (live for `Subscribe`, indexed for `Poll`) and emits an initial
45    /// progress frame before subsequent items.
46    #[default]
47    Tip,
48    /// Starts at and includes transactions from this checkpoint.
49    Checkpoint(u64),
50    /// Starts strictly after this server-validated opaque cursor.
51    Resume(Bytes),
52}
53
54/// Starting position for an event stream.
55#[non_exhaustive]
56#[derive(Clone, Debug, Default, PartialEq, Eq)]
57pub enum EventStreamStart {
58    /// Begins at the chain tip (live for `Subscribe`, indexed for `Poll`) and emits an initial
59    /// progress frame before subsequent items.
60    #[default]
61    Tip,
62    /// Starts at and includes events from this checkpoint.
63    Checkpoint(u64),
64    /// Starts strictly after this server-validated opaque cursor.
65    Resume(Bytes),
66}
67
68/// Request for a resumable checkpoint stream.
69#[non_exhaustive]
70#[derive(Clone, Debug, Default, PartialEq)]
71pub struct CheckpointStreamRequest {
72    /// Projection applied to both List and Subscribe requests.
73    pub read_mask: Option<FieldMask>,
74    /// Transaction filter applied to checkpoints.
75    pub filter: Option<TransactionFilter>,
76    /// Logical starting position.
77    pub start: CheckpointStreamStart,
78    /// How the stream follows the ledger.
79    pub delivery: Delivery,
80}
81
82impl CheckpointStreamRequest {
83    /// Creates a request with the default start and delivery.
84    pub fn new() -> Self {
85        Self::default()
86    }
87
88    /// Sets the response projection.
89    pub fn with_read_mask(mut self, read_mask: impl Into<FieldMask>) -> Self {
90        self.read_mask = Some(read_mask.into());
91        self
92    }
93
94    /// Sets the transaction filter.
95    pub fn with_filter(mut self, filter: impl Into<TransactionFilter>) -> Self {
96        self.filter = Some(filter.into());
97        self
98    }
99
100    /// Sets the logical starting position.
101    pub fn with_start(mut self, start: impl Into<CheckpointStreamStart>) -> Self {
102        self.start = start.into();
103        self
104    }
105
106    /// Sets how the stream follows the ledger.
107    pub fn with_delivery(mut self, delivery: impl Into<Delivery>) -> Self {
108        self.delivery = delivery.into();
109        self
110    }
111}
112
113/// Request for a resumable transaction stream.
114#[non_exhaustive]
115#[derive(Clone, Debug, Default, PartialEq)]
116pub struct TransactionStreamRequest {
117    /// Projection applied to both List and Subscribe requests.
118    pub read_mask: Option<FieldMask>,
119    /// Filter applied to transactions.
120    pub filter: Option<TransactionFilter>,
121    /// Logical starting position.
122    pub start: TransactionStreamStart,
123    /// How the stream follows the ledger.
124    pub delivery: Delivery,
125}
126
127impl TransactionStreamRequest {
128    /// Creates a request with the default start and delivery.
129    pub fn new() -> Self {
130        Self::default()
131    }
132
133    /// Sets the response projection.
134    pub fn with_read_mask(mut self, read_mask: impl Into<FieldMask>) -> Self {
135        self.read_mask = Some(read_mask.into());
136        self
137    }
138
139    /// Sets the transaction filter.
140    pub fn with_filter(mut self, filter: impl Into<TransactionFilter>) -> Self {
141        self.filter = Some(filter.into());
142        self
143    }
144
145    /// Sets the logical starting position.
146    pub fn with_start(mut self, start: impl Into<TransactionStreamStart>) -> Self {
147        self.start = start.into();
148        self
149    }
150
151    /// Sets how the stream follows the ledger.
152    pub fn with_delivery(mut self, delivery: impl Into<Delivery>) -> Self {
153        self.delivery = delivery.into();
154        self
155    }
156}
157
158/// Request for a resumable event stream.
159#[non_exhaustive]
160#[derive(Clone, Debug, Default, PartialEq)]
161pub struct EventStreamRequest {
162    /// Projection applied to both List and Subscribe requests.
163    pub read_mask: Option<FieldMask>,
164    /// Filter applied to events.
165    pub filter: Option<EventFilter>,
166    /// Logical starting position.
167    pub start: EventStreamStart,
168    /// How the stream follows the ledger.
169    pub delivery: Delivery,
170}
171
172impl EventStreamRequest {
173    /// Creates a request with the default start and delivery.
174    pub fn new() -> Self {
175        Self::default()
176    }
177
178    /// Sets the response projection.
179    pub fn with_read_mask(mut self, read_mask: impl Into<FieldMask>) -> Self {
180        self.read_mask = Some(read_mask.into());
181        self
182    }
183
184    /// Sets the event filter.
185    pub fn with_filter(mut self, filter: impl Into<EventFilter>) -> Self {
186        self.filter = Some(filter.into());
187        self
188    }
189
190    /// Sets the logical starting position.
191    pub fn with_start(mut self, start: impl Into<EventStreamStart>) -> Self {
192        self.start = start.into();
193        self
194    }
195
196    /// Sets how the stream follows the ledger.
197    pub fn with_delivery(mut self, delivery: impl Into<Delivery>) -> Self {
198        self.delivery = delivery.into();
199        self
200    }
201}
202
203/// A checkpoint frame with an optional payload and inclusive restart cursor.
204///
205/// Process `checkpoint` before persisting `cursor`. To resume after this checkpoint,
206/// start the next stream from `CheckpointStreamStart::Checkpoint(cursor + 1)`.
207#[non_exhaustive]
208#[derive(Clone, Debug)]
209pub struct CheckpointStreamFrame {
210    /// The checkpoint payload, or `None` for a progress-only frame.
211    pub checkpoint: Option<Checkpoint>,
212    /// The inclusive checkpoint cursor represented by this frame.
213    pub cursor: u64,
214}
215
216/// A transaction frame with an optional payload and opaque exclusive restart cursor.
217///
218/// Process `transaction` before persisting `cursor`, then resume with
219/// `TransactionStreamStart::Resume(cursor)`.
220#[non_exhaustive]
221#[derive(Clone, Debug)]
222pub struct TransactionStreamFrame {
223    /// The executed transaction payload, or `None` for a progress-only frame.
224    pub transaction: Option<ExecutedTransaction>,
225    /// The server's opaque resume cursor as of this frame, passed through verbatim.
226    pub cursor: Bytes,
227    /// Checkpoint coverage reported with this frame, when known, for progress observability.
228    pub covered_checkpoint: Option<u64>,
229}
230
231/// An event frame with an optional payload and opaque exclusive restart cursor.
232///
233/// Process `event` before persisting `cursor`, then resume with `EventStreamStart::Resume(cursor)`.
234#[non_exhaustive]
235#[derive(Clone, Debug)]
236pub struct EventStreamFrame {
237    /// The event payload, or `None` for a progress-only frame.
238    pub event: Option<Event>,
239    /// The server's opaque resume cursor as of this frame, passed through verbatim.
240    pub cursor: Bytes,
241    /// Checkpoint coverage reported with this frame, when known, for progress observability.
242    pub covered_checkpoint: Option<u64>,
243}
244
245/// Polling, retry, buffering, and observability controls for `stream_*` operations.
246///
247/// Client-level timeouts bound individual RPCs, not the total duration of the stream.
248/// Finite `list_*` operations use the retry and observer controls in [`ListConfig`].
249#[non_exhaustive]
250#[derive(Clone)]
251pub struct LedgerStreamConfig {
252    /// Delay between polls when using `Poll` delivery. Must be non-zero.
253    pub ledger_tip_poll_interval: Duration,
254    /// Maximum items per internally built List request.
255    pub list_page_limit: Option<u32>,
256    /// Delay before the first transient retry.
257    pub base_retry_delay: Duration,
258    /// Maximum exponential portion of a retry delay.
259    pub max_retry_delay: Duration,
260    /// Maximum random jitter added to a retry delay.
261    pub retry_jitter: Duration,
262    /// Maximum number of live subscription items buffered during gap replay before reconnecting.
263    pub max_buffered_live_items: NonZeroUsize,
264    observer: Option<LedgerStreamObserver>,
265}
266
267impl fmt::Debug for LedgerStreamConfig {
268    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
269        formatter
270            .debug_struct("LedgerStreamConfig")
271            .field("ledger_tip_poll_interval", &self.ledger_tip_poll_interval)
272            .field("list_page_limit", &self.list_page_limit)
273            .field("base_retry_delay", &self.base_retry_delay)
274            .field("max_retry_delay", &self.max_retry_delay)
275            .field("retry_jitter", &self.retry_jitter)
276            .field("max_buffered_live_items", &self.max_buffered_live_items)
277            .field("observer_configured", &self.observer.is_some())
278            .finish()
279    }
280}
281
282impl Default for LedgerStreamConfig {
283    fn default() -> Self {
284        Self {
285            ledger_tip_poll_interval: Duration::from_secs(1),
286            list_page_limit: None,
287            base_retry_delay: Duration::from_millis(250),
288            max_retry_delay: Duration::from_secs(30),
289            retry_jitter: Duration::from_millis(500),
290            max_buffered_live_items: NonZeroUsize::new(1_024).expect("non-zero constant"),
291            observer: None,
292        }
293    }
294}
295
296impl LedgerStreamConfig {
297    /// Installs the synchronous stream observer, replacing any prior callback.
298    ///
299    /// The callback must return quickly and not block; its panic propagates to the polling task.
300    pub fn with_observer(
301        mut self,
302        observer: impl Fn(LedgerStreamEvent) + Send + Sync + 'static,
303    ) -> Self {
304        self.observer = Some(Arc::new(observer));
305        self
306    }
307
308    pub(super) fn observer(&self) -> Option<LedgerStreamObserver> {
309        self.observer.clone()
310    }
311}
312
313/// Retry and observability controls for finite `list_*` operations.
314#[non_exhaustive]
315#[derive(Clone)]
316pub struct ListConfig {
317    /// Delay before the first transient retry.
318    pub base_retry_delay: Duration,
319    /// Maximum exponential portion of a retry delay.
320    pub max_retry_delay: Duration,
321    /// Maximum random jitter added to a retry delay.
322    pub retry_jitter: Duration,
323    observer: Option<ListObserver>,
324}
325
326impl fmt::Debug for ListConfig {
327    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
328        formatter
329            .debug_struct("ListConfig")
330            .field("base_retry_delay", &self.base_retry_delay)
331            .field("max_retry_delay", &self.max_retry_delay)
332            .field("retry_jitter", &self.retry_jitter)
333            .field("observer_configured", &self.observer.is_some())
334            .finish()
335    }
336}
337
338impl Default for ListConfig {
339    fn default() -> Self {
340        let stream_defaults = LedgerStreamConfig::default();
341        Self {
342            base_retry_delay: stream_defaults.base_retry_delay,
343            max_retry_delay: stream_defaults.max_retry_delay,
344            retry_jitter: stream_defaults.retry_jitter,
345            observer: None,
346        }
347    }
348}
349
350impl ListConfig {
351    /// Installs the synchronous List observer, replacing any prior callback.
352    ///
353    /// The callback must return quickly and not block; its panic propagates to the polling task.
354    pub fn with_observer(mut self, observer: impl Fn(ListEvent) + Send + Sync + 'static) -> Self {
355        self.observer = Some(Arc::new(observer));
356        self
357    }
358
359    pub(super) fn into_stream_config(self) -> LedgerStreamConfig {
360        let observer: Option<LedgerStreamObserver> = self.observer.map(|list_observer| {
361            Arc::new(move |event: LedgerStreamEvent| {
362                if let Some(list_event) = ListEvent::from_stream_event(event) {
363                    list_observer(list_event);
364                }
365            }) as LedgerStreamObserver
366        });
367
368        LedgerStreamConfig {
369            base_retry_delay: self.base_retry_delay,
370            max_retry_delay: self.max_retry_delay,
371            retry_jitter: self.retry_jitter,
372            observer,
373            ..LedgerStreamConfig::default()
374        }
375    }
376}
sui_rpc/client/ledger_streams/types.rs

sui_rpc/client/ledger_streams/
types.rs
