sui_rpc/client/ledger_streams/adapter/
mod.rs

1use std::future::Future;
2use std::pin::Pin;
3
4use prost::Message;
5use prost::bytes::Bytes;
6use prost_types::FieldMask;
7use tonic::Request;
8use tonic::Status;
9use tonic::codegen::BoxStream;
10
11use super::super::Client;
12use super::super::Result;
13use super::observability::LedgerStreamFamily;
14use crate::proto::sui::rpc::v2::Ordering;
15use crate::proto::sui::rpc::v2::QueryEnd;
16use crate::proto::sui::rpc::v2::QueryOptions;
17use crate::proto::sui::rpc::v2::Watermark;
18
19mod checkpoint;
20mod event;
21mod progress;
22mod transaction;
23
24pub(super) use checkpoint::CheckpointAdapter;
25pub(super) use event::EventAdapter;
26pub(super) use progress::CursorDomain;
27pub(super) use progress::CursorGapUpper;
28pub(super) use progress::Progress;
29pub(super) use progress::ProgressAdvance;
30pub(super) use progress::Recovery;
31pub(super) use progress::RecoveryGap;
32pub(super) use transaction::TransactionAdapter;
33
34const CHECKPOINT_CURSOR_OVERFLOW: &str = "checkpoint subscription cursor cannot be resumed";
35
36pub(super) type RpcFuture<T> = Pin<Box<dyn Future<Output = Result<BoxStream<T>>> + Send + 'static>>;
37
38pub(super) struct PositionedItem<I, P> {
39    payload: I,
40    position: P,
41}
42
43impl<I, P> PositionedItem<I, P> {
44    fn new(payload: I, position: P) -> Self {
45        Self { payload, position }
46    }
47
48    pub(super) fn position(&self) -> &P {
49        &self.position
50    }
51
52    fn into_payload(self) -> I {
53        self.payload
54    }
55}
56
57pub(super) struct ListResponseParts<A: SubscriptionAdapter + ?Sized> {
58    pub(super) item: Option<A::Item>,
59    pub(super) watermark: Option<Watermark>,
60}
61
62pub(super) struct LiveFrame<I, P> {
63    pub(super) item: Option<I>,
64    pub(super) progress: P,
65}
66
67pub(super) fn validate_caller_read_mask<A: SubscriptionAdapter>(
68    read_mask: Option<&FieldMask>,
69) -> Result<()> {
70    let Some(read_mask) = read_mask else {
71        return Err(Status::invalid_argument(A::READ_MASK_REQUIREMENT));
72    };
73    if read_mask.paths.is_empty() {
74        return Err(Status::invalid_argument(A::READ_MASK_REQUIREMENT));
75    }
76    if read_mask.paths.iter().any(|path| path == "*") {
77        return Ok(());
78    }
79    if A::REQUIRED_READ_MASK_FIELDS.iter().all(|required| {
80        read_mask
81            .paths
82            .iter()
83            .any(|path| path.as_str() == *required)
84    }) {
85        Ok(())
86    } else {
87        Err(Status::invalid_argument(A::READ_MASK_REQUIREMENT))
88    }
89}
90
91/// Connects one checkpoint, transaction, or event protocol family to the shared driver.
92pub(super) trait SubscriptionAdapter: Send + Sync + 'static {
93    const FAMILY: LedgerStreamFamily;
94    /// Scalar payload fields the state machine needs for ordering, deduplication, and restart.
95    const REQUIRED_READ_MASK_FIELDS: &'static [&'static str];
96    /// Error returned when a caller mask omits required identity fields.
97    const READ_MASK_REQUIREMENT: &'static str;
98
99    type Item: Send + 'static;
100    /// Ledger identity used for ordering, deduplication, and resume.
101    type ItemPosition: Clone + Eq + Ord + Send + 'static;
102    /// Cursor carried by payload and progress-only frames.
103    type Cursor: CursorDomain;
104    type Output: Send + 'static;
105    type ListRequest: Clone + Send + Sync + 'static;
106    type ListResponse: Send + 'static;
107    type SubscribeRequest: Clone + Send + 'static;
108    type SubscribeResponse: Message + Send + 'static;
109
110    /// Returns the caller's List projection unchanged.
111    fn list_read_mask(request: &Self::ListRequest) -> Option<&FieldMask>;
112    /// Builds an unbounded ascending recovery request from a live request.
113    fn list_request_from_subscribe(request: &Self::SubscribeRequest) -> Self::ListRequest;
114    /// Returns the List request's pagination and ordering options when present.
115    fn options(request: &Self::ListRequest) -> Option<&QueryOptions>;
116    /// Returns mutable List pagination and ordering options, inserting defaults when absent.
117    fn options_mut(request: &mut Self::ListRequest) -> &mut QueryOptions;
118    /// Returns the List request's inclusive lower checkpoint bound.
119    fn start_checkpoint(request: &Self::ListRequest) -> Option<u64>;
120    /// Sets the List request's inclusive lower checkpoint bound.
121    fn set_start_checkpoint(request: &mut Self::ListRequest, checkpoint: Option<u64>);
122    /// Returns the List request's exclusive upper checkpoint bound.
123    fn end_checkpoint(request: &Self::ListRequest) -> Option<u64>;
124    /// Sets the List request's exclusive upper checkpoint bound.
125    fn set_end_checkpoint(request: &mut Self::ListRequest, checkpoint: Option<u64>);
126    /// Advances an ascending List request strictly past `progress`.
127    fn set_ascending_resume(
128        request: &mut Self::ListRequest,
129        progress: &Progress<Self::Cursor>,
130    ) -> Result<()>;
131    /// Returns progress covered by the request's resume bound, if any.
132    fn request_resume_position(request: &Self::ListRequest) -> Option<Progress<Self::Cursor>>;
133    /// Validates a `CheckpointBound` ending against the request bounds.
134    fn validate_checkpoint_bound(
135        request: &Self::ListRequest,
136        direction: ListScanDirection,
137        checkpoint: Option<u64>,
138    ) -> Result<()>;
139    /// Returns the item identity used for ordering and deduplication.
140    fn item_position(item: &Self::Item) -> &Self::ItemPosition;
141    /// Extracts `(has_item, watermark, query_end)` metadata from a list response frame.
142    fn extract_metadata(
143        response: &Self::ListResponse,
144    ) -> (bool, Option<&Watermark>, Option<&QueryEnd>);
145    /// Splits payload from progress and validates projected identity fields.
146    fn split_list(response: Self::ListResponse) -> Result<ListResponseParts<Self>>;
147    /// Whether successful live frames must contain a payload item.
148    fn item_required(_request: &Self::SubscribeRequest) -> bool {
149        false
150    }
151    /// Validates and converts a Subscribe response to a normalized live frame.
152    fn parse_live(
153        response: Self::SubscribeResponse,
154        item_required: bool,
155    ) -> Result<LiveFrame<Self::Item, Progress<Self::Cursor>>>;
156    /// Converts an optional payload and committed progress to a public frame.
157    fn into_output(item: Option<Self::Item>, progress: Progress<Self::Cursor>) -> Self::Output;
158    fn dispatch_list(
159        client: Client,
160        request: Request<Self::ListRequest>,
161    ) -> RpcFuture<Self::ListResponse>;
162    fn dispatch_subscribe(
163        client: Client,
164        request: Request<Self::SubscribeRequest>,
165    ) -> RpcFuture<Self::SubscribeResponse>;
166}
167
168fn parse_opaque_live_frame<I, P>(
169    item: Option<PositionedItem<I, P>>,
170    watermark: Option<Watermark>,
171) -> Result<LiveFrame<PositionedItem<I, P>, Progress<Bytes>>> {
172    let watermark = watermark
173        .ok_or_else(|| Status::data_loss("subscription frame is missing its watermark"))?;
174    let cursor = watermark
175        .cursor
176        .ok_or_else(|| Status::data_loss("subscription watermark is missing its cursor"))?;
177    Ok(LiveFrame {
178        item,
179        progress: Progress {
180            cursor,
181            checkpoint: watermark.checkpoint,
182        },
183    })
184}
185
186#[derive(Clone, Copy, Debug, PartialEq, Eq)]
187pub(super) enum ListScanDirection {
188    Ascending,
189    Descending,
190}
191
192impl ListScanDirection {
193    pub(super) fn from_request<A: SubscriptionAdapter>(request: &A::ListRequest) -> Result<Self> {
194        match A::options(request).and_then(|options| options.ordering) {
195            None => Ok(Self::Ascending),
196            Some(ordering) => match Ordering::try_from(ordering) {
197                Ok(Ordering::Ascending) => Ok(Self::Ascending),
198                Ok(Ordering::Descending) => Ok(Self::Descending),
199                Err(_) => Err(Status::invalid_argument("List ordering is unknown")),
200            },
201        }
202    }
203
204    pub(super) fn resume_bound(self, options: &QueryOptions) -> &Option<Bytes> {
205        match self {
206            Self::Ascending => &options.after,
207            Self::Descending => &options.before,
208        }
209    }
210
211    pub(super) fn set_resume_bound(self, options: &mut QueryOptions, cursor: Bytes) {
212        match self {
213            Self::Ascending => options.after = Some(cursor),
214            Self::Descending => options.before = Some(cursor),
215        }
216    }
217}
218
219pub(super) fn validate_typed_checkpoint_bound(
220    start_checkpoint: Option<u64>,
221    end_checkpoint: Option<u64>,
222    direction: ListScanDirection,
223    watermark_checkpoint: Option<u64>,
224) -> Result<()> {
225    let expected_checkpoint = match direction {
226        ListScanDirection::Ascending => end_checkpoint.and_then(|end| end.checked_sub(1)),
227        ListScanDirection::Descending => Some(start_checkpoint.unwrap_or(0)),
228    };
229    if watermark_checkpoint.is_some() && watermark_checkpoint != expected_checkpoint {
230        Err(Status::data_loss(
231            "List CheckpointBound watermark does not match the requested bound",
232        ))
233    } else {
234        Ok(())
235    }
236}
sui_rpc/client/ledger_streams/adapter/mod.rs

sui_rpc/client/ledger_streams/adapter/
mod.rs
