sui_rpc_api/ledger_history/
watermark.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Shared `Watermark` construction for the v2 list APIs.
5//!
6//! Both ledger-history backends — the fullnode (`sui-rpc-api`) and bigtable
7//! (`sui-kv-rpc`) — and all three list handlers (`list_transactions`,
8//! `list_events`, `list_checkpoints`) emit the same wire `Watermark`: a resume
9//! cursor plus a completion boundary (`checkpoint`, the inclusive boundary
10//! checkpoint the scan has fully covered in the request's ordering direction).
11//! The cursor encoding and the boundary bookkeeping are identical; what differs
12//! per API is how a scan position resolves into a completion-boundary candidate:
13//!
14//! - `list_transactions` / `list_events` scan within a checkpoint, so an
15//!   item at checkpoint `C` does NOT prove `C` complete (more matches may sit
16//!   at higher/lower transaction or event positions). Their covered bound is
17//!   advanced before `C` — see [`advance_covered_bound_before_checkpoint`].
18//! - `list_checkpoints` dedupes checkpoint numbers, so "checkpoint `C`
19//!   emitted" means "checkpoint `C` complete." Its item path directly records
20//!   `C`; independently resolved frontier candidates are folded with
21//!   [`merge_covered_checkpoint_bound`].
22//!
23//! This module owns the shared pieces; each handler keeps only its
24//! API-specific frontier-to-candidate adapter.
25
26use sui_inverted_index::ScanDirection;
27use sui_rpc::proto::sui::rpc::v2::QueryEndReason;
28use sui_rpc::proto::sui::rpc::v2::Watermark;
29use sui_rpc_cursor::{CursorKind, CursorToken, Position};
30
31use crate::RpcError;
32use crate::ledger_history::query_options::{QueryOptions, RangeExhaustion, below_serving_floor};
33
34/// Populate the completion-boundary `checkpoint` field of a `Watermark` from
35/// the per-scan boundary value. The value already carries the direction-correct
36/// meaning (inclusive upper bound ascending, inclusive lower bound descending);
37/// the single wire field records it regardless of ordering.
38fn set_checkpoint_bound(wm: &mut Watermark, boundary: Option<u64>) {
39    wm.checkpoint = boundary;
40}
41
42/// Merge a fully covered checkpoint candidate into the accumulated inclusive
43/// bound. The bound advances by max in ascending scans and min in descending
44/// scans.
45pub fn merge_covered_checkpoint_bound(
46    covered_checkpoint_bound: Option<u64>,
47    candidate_bound: u64,
48    options: &QueryOptions,
49) -> Option<u64> {
50    Some(match covered_checkpoint_bound {
51        None => candidate_bound,
52        Some(bound) if options.is_ascending() => bound.max(candidate_bound),
53        Some(bound) => bound.min(candidate_bound),
54    })
55}
56
57/// Advance the inclusive covered bound using a checkpoint that is not itself
58/// proven complete. Transactions, events, and scan frontiers can leave more
59/// matches within checkpoint `C`, so the candidate excludes `C`: `C - 1`
60/// ascending and `C + 1` descending. The adjusted candidate is then merged by
61/// max ascending or min descending.
62///
63/// `entry_checkpoint` is the window's first checkpoint in scan direction
64/// (raised to the serving floor ascending). A candidate strictly before it
65/// lies outside the window, and is discarded. Within the window, a candidate
66/// counts even before the scan reads its checkpoint's start: the request's
67/// starting cursor vouches for every position before it.
68///
69/// When the `C ∓ 1` adjustment would overflow (`C == 0` ascending or
70/// `u64::MAX` descending), the previously covered bound is preserved.
71pub fn advance_covered_bound_before_checkpoint(
72    covered_checkpoint_bound: Option<u64>,
73    incomplete_checkpoint: u64,
74    entry_checkpoint: u64,
75    options: &QueryOptions,
76) -> Option<u64> {
77    let candidate_bound = if options.is_ascending() {
78        incomplete_checkpoint
79            .checked_sub(1)
80            .filter(|candidate| *candidate >= entry_checkpoint)
81    } else {
82        incomplete_checkpoint
83            .checked_add(1)
84            .filter(|candidate| *candidate <= entry_checkpoint)
85    };
86    match candidate_bound {
87        Some(candidate_bound) => {
88            merge_covered_checkpoint_bound(covered_checkpoint_bound, candidate_bound, options)
89        }
90        None => covered_checkpoint_bound,
91    }
92}
93
94/// Build the embedded `Watermark` for an item: the cursor encodes this
95/// item's position (so the next request's `after`/`before` resumes past it)
96/// plus the current direction-matching checkpoint boundary. `cp` /
97/// `position` are the item's cursor coordinates (`list_checkpoints` passes
98/// its cp_seq for both).
99pub fn item_watermark(position: Position, boundary: Option<u64>) -> Watermark {
100    let mut wm = Watermark::default();
101    wm.cursor = Some(CursorToken::item(position).encode());
102    set_checkpoint_bound(&mut wm, boundary);
103    wm
104}
105
106/// Build a standalone scan-frontier `Watermark`. `cursor_cp` / `position`
107/// are the boundary cursor coordinates the caller has already resolved for
108/// its scan domain (see [`boundary_cursor_cp`] for the per-checkpoint
109/// scanners' direction adjustment); `boundary` is the accumulated
110/// completion boundary.
111pub fn boundary_watermark(position: Position, boundary: Option<u64>) -> Watermark {
112    cursor_watermark(position, boundary, sui_rpc_cursor::CursorKind::Boundary)
113}
114
115/// Build a watermark whose cursor kind has been resolved by query-range
116/// bookkeeping (see [`RangeExhaustion::CursorBound`]): an `after` Item cursor
117/// at the last position names no boundary after it, so its closing cursor keeps
118/// Item kind where a Boundary at its position would re-include the item on resume.
119fn cursor_watermark(
120    position: Position,
121    boundary: Option<u64>,
122    cursor_kind: sui_rpc_cursor::CursorKind,
123) -> Watermark {
124    let cursor = match cursor_kind {
125        sui_rpc_cursor::CursorKind::Item => CursorToken::item(position),
126        sui_rpc_cursor::CursorKind::Boundary => CursorToken::boundary(position),
127    };
128    let mut wm = Watermark::default();
129    wm.cursor = Some(cursor.encode());
130    set_checkpoint_bound(&mut wm, boundary);
131    wm
132}
133
134/// Resolve the boundary-cursor checkpoint coordinate for a `list_transactions`
135/// / `list_events` scan frontier. The cursor encoding is asymmetric:
136/// ascending `Boundary` cursors advance the cp-range start, so the frontier
137/// cp is used directly; descending `Boundary` cursors treat the cp
138/// coordinate as an EXCLUSIVE upper bound, so `cp + 1` is needed to keep
139/// `cp` itself included on resume. Sent back in the other direction, either
140/// coordinate can sit a checkpoint away from the one the position needs, so
141/// [`ResolvedScan::resolve`] bounds a scan by cursor positions and reads the
142/// coordinate only to label checkpoints.
143///
144/// [`ResolvedScan::resolve`]: crate::ledger_history::query_options::ResolvedScan::resolve
145pub fn boundary_cursor_cp(cp: u64, direction: ScanDirection) -> u64 {
146    if direction.is_ascending() {
147        cp
148    } else {
149        cp.saturating_add(1)
150    }
151}
152
153/// Resolve the checkpoint coordinate embedded in a transaction/event/checkpoint
154/// scan-frontier cursor independently from the optional completed-checkpoint
155/// claim. A missing mapping is representable only at the numeric edge where
156/// the frontier itself supplies the sole safe checkpoint coordinate.
157pub fn scan_frontier_cursor_cp(
158    checkpoint: Option<u64>,
159    frontier: u64,
160    direction: ScanDirection,
161) -> Option<u64> {
162    checkpoint
163        .map(|cp| boundary_cursor_cp(cp, direction))
164        .or_else(|| {
165            ((direction.is_ascending() && frontier == 0)
166                || (!direction.is_ascending() && frontier == u64::MAX))
167                .then_some(frontier)
168        })
169}
170
171/// Terminal of a successful list scan that renders as the trailing
172/// payload-free `QueryEnd` frame. `ItemLimit` never reaches this type: the
173/// drive loops fuse it onto the final item frame and suppress the trailing
174/// frame. The wire reason and the watermark policy are projections of the
175/// same value, so they cannot disagree.
176#[derive(Clone, Copy, Debug, Eq, PartialEq)]
177pub enum NaturalRangeEnd {
178    LedgerTip,
179    CheckpointBound,
180}
181
182#[derive(Clone, Debug, PartialEq)]
183pub enum ScanTerminal {
184    /// Scan budget exhausted. Owns the mandatory authoritative frontier
185    /// watermark (its cursor is always set by the frontier constructors).
186    ScanLimit { watermark: Watermark },
187    /// The resolved interval is naturally exhausted at `position`.
188    NaturalRange {
189        end: NaturalRangeEnd,
190        position: Position,
191        /// The checkpoint this page's natural end reports as fully covered in
192        /// `Watermark.checkpoint`, or `None` when nothing remains to claim: the window has no
193        /// checkpoint, or an ascending scan ends at genesis. See
194        /// [`ResolvedScan::natural_end_claim`].
195        ///
196        /// [`ResolvedScan::natural_end_claim`]: crate::ledger_history::query_options::ResolvedScan::natural_end_claim
197        claimed_checkpoint: Option<u64>,
198    },
199    /// The caller-provided cursor bound truncates the interval at `position`.
200    CursorBound {
201        position: Position,
202        kind: CursorKind,
203    },
204    /// A descending scan with no lower bound of its own reached this server's serving floor at
205    /// `position`; everything below is pruned. It never renders as a `QueryEnd`: below the item
206    /// limit, the stream sends a payload-free frame with [`Self::into_watermark`] and no `end`,
207    /// then fails with the error [`Self::reason`] returns for it.
208    ServingFloor {
209        position: Position,
210        floor_checkpoint: u64,
211    },
212}
213
214impl ScanTerminal {
215    pub fn from_range_exhaustion(
216        exhaustion: RangeExhaustion,
217        position: Position,
218        claimed_checkpoint: Option<u64>,
219    ) -> Self {
220        match exhaustion {
221            RangeExhaustion::LedgerTip => Self::NaturalRange {
222                end: NaturalRangeEnd::LedgerTip,
223                position,
224                claimed_checkpoint,
225            },
226            RangeExhaustion::CheckpointBound => Self::NaturalRange {
227                end: NaturalRangeEnd::CheckpointBound,
228                position,
229                claimed_checkpoint,
230            },
231            RangeExhaustion::CursorBound { kind } => Self::CursorBound { position, kind },
232            RangeExhaustion::ServingFloor { floor_checkpoint } => Self::ServingFloor {
233                position,
234                floor_checkpoint,
235            },
236        }
237    }
238
239    /// The `QueryEnd` reason this terminal ends its page with. [`Self::ServingFloor`] ends its page
240    /// with no `QueryEnd`, so for it this is the `OutOfRange` error its stream fails with instead,
241    /// after the terminal watermark frame.
242    pub fn reason(&self) -> Result<QueryEndReason, RpcError> {
243        match self {
244            Self::ScanLimit { .. } => Ok(QueryEndReason::ScanLimit),
245            Self::NaturalRange {
246                end: NaturalRangeEnd::LedgerTip,
247                ..
248            } => Ok(QueryEndReason::LedgerTip),
249            Self::NaturalRange {
250                end: NaturalRangeEnd::CheckpointBound,
251                ..
252            } => Ok(QueryEndReason::CheckpointBound),
253            Self::CursorBound { .. } => Ok(QueryEndReason::CursorBound),
254            Self::ServingFloor {
255                floor_checkpoint, ..
256            } => Err(below_serving_floor(*floor_checkpoint)),
257        }
258    }
259
260    /// Render the trailing terminal frame's watermark. Natural completion
261    /// (LedgerTip/CheckpointBound) claims its `claimed_checkpoint` and ignores
262    /// `covered_checkpoint_bound` (the range claim is always at least as
263    /// strong). A cursor bound never claims its own checkpoint: its claim is
264    /// exactly the accumulated item coverage. A serving floor claims the floor
265    /// checkpoint behind a boundary cursor at the floor.
266    pub fn into_watermark(self, covered_checkpoint_bound: Option<u64>) -> Watermark {
267        match self {
268            Self::ScanLimit { watermark } => watermark,
269            Self::NaturalRange {
270                position,
271                claimed_checkpoint,
272                ..
273            } => boundary_watermark(position, claimed_checkpoint),
274            Self::CursorBound { position, kind } => {
275                cursor_watermark(position, covered_checkpoint_bound, kind)
276            }
277            Self::ServingFloor {
278                position,
279                floor_checkpoint,
280            } => boundary_watermark(position, Some(floor_checkpoint)),
281        }
282    }
283}
284
285#[cfg(test)]
286mod tests {
287    use super::*;
288    use sui_rpc_cursor::{CursorKind, Position};
289
290    fn options(ascending: bool) -> QueryOptions {
291        let mut request = sui_rpc::proto::sui::rpc::v2::QueryOptions::default();
292        request.ordering = Some(if ascending {
293            sui_rpc::proto::sui::rpc::v2::Ordering::Ascending as i32
294        } else {
295            sui_rpc::proto::sui::rpc::v2::Ordering::Descending as i32
296        });
297        QueryOptions::transactions_from_proto(Some(&request), 100, 100).unwrap()
298    }
299
300    #[test]
301    fn merge_covered_checkpoint_bound_keeps_most_advanced_in_direction() {
302        let asc = options(true);
303        assert_eq!(merge_covered_checkpoint_bound(None, 5, &asc), Some(5));
304        assert_eq!(merge_covered_checkpoint_bound(Some(5), 9, &asc), Some(9));
305        assert_eq!(merge_covered_checkpoint_bound(Some(9), 5, &asc), Some(9));
306
307        let desc = options(false);
308        assert_eq!(merge_covered_checkpoint_bound(None, 9, &desc), Some(9));
309        assert_eq!(merge_covered_checkpoint_bound(Some(9), 5, &desc), Some(5));
310        assert_eq!(merge_covered_checkpoint_bound(Some(5), 9, &desc), Some(5));
311    }
312
313    /// The per-checkpoint scanners exclude the item's own cp: `C - 1`
314    /// ascending, `C + 1` descending.
315    #[test]
316    fn advance_covered_bound_before_checkpoint_adjusts_by_one() {
317        let asc = options(true);
318        assert_eq!(
319            advance_covered_bound_before_checkpoint(None, 10, 5, &asc),
320            Some(9)
321        );
322        assert_eq!(
323            advance_covered_bound_before_checkpoint(Some(9), 12, 5, &asc),
324            Some(11)
325        );
326
327        let desc = options(false);
328        assert_eq!(
329            advance_covered_bound_before_checkpoint(None, 10, 15, &desc),
330            Some(11)
331        );
332        assert_eq!(
333            advance_covered_bound_before_checkpoint(Some(11), 8, 15, &desc),
334            Some(9)
335        );
336    }
337
338    /// A candidate before the entry checkpoint lies outside the window and is
339    /// discarded. The claim at exactly the entry checkpoint (candidate ==
340    /// entry) is the first legal one.
341    #[test]
342    fn advance_covered_bound_before_checkpoint_discards_candidates_outside_the_window() {
343        let asc = options(true);
344        assert_eq!(
345            advance_covered_bound_before_checkpoint(None, 10, 10, &asc),
346            None
347        );
348        assert_eq!(
349            advance_covered_bound_before_checkpoint(None, 11, 10, &asc),
350            Some(10)
351        );
352
353        let desc = options(false);
354        assert_eq!(
355            advance_covered_bound_before_checkpoint(None, 10, 10, &desc),
356            None
357        );
358        assert_eq!(
359            advance_covered_bound_before_checkpoint(None, 9, 10, &desc),
360            Some(10)
361        );
362    }
363
364    /// Overflow at the range edge (`cp 0` ascending, `u64::MAX` descending)
365    /// preserves the previously accumulated boundary instead of dropping it.
366    #[test]
367    fn advance_covered_bound_before_checkpoint_preserves_prev_on_overflow() {
368        let asc = options(true);
369        assert_eq!(
370            advance_covered_bound_before_checkpoint(Some(4), 0, 0, &asc),
371            Some(4)
372        );
373        assert_eq!(
374            advance_covered_bound_before_checkpoint(None, 0, 0, &asc),
375            None
376        );
377
378        let desc = options(false);
379        assert_eq!(
380            advance_covered_bound_before_checkpoint(Some(4), u64::MAX, u64::MAX, &desc),
381            Some(4)
382        );
383        assert_eq!(
384            advance_covered_bound_before_checkpoint(None, u64::MAX, u64::MAX, &desc),
385            None
386        );
387    }
388
389    #[test]
390    fn boundary_cursor_cp_bumps_descending_only() {
391        assert_eq!(boundary_cursor_cp(10, ScanDirection::Ascending), 10);
392        assert_eq!(boundary_cursor_cp(10, ScanDirection::Descending), 11);
393        assert_eq!(
394            boundary_cursor_cp(u64::MAX, ScanDirection::Descending),
395            u64::MAX
396        );
397    }
398
399    /// The direction-correct boundary is recorded in the single `checkpoint` field regardless of
400    /// ordering. A client reads the bound off the wire frame and interprets it per the request's
401    /// ordering.
402    #[test]
403    fn item_watermark_sets_direction_matching_bound() {
404        let pos = Position::Transactions {
405            checkpoint: 9,
406            tx_seq: 42,
407        };
408        let wm = item_watermark(pos, Some(8));
409        assert_eq!(wm.checkpoint, Some(8));
410        assert_eq!(wm.cursor.as_ref(), Some(&CursorToken::item(pos).encode()));
411
412        let wm = item_watermark(pos, None);
413        assert_eq!(wm.checkpoint, None);
414    }
415
416    #[test]
417    fn scan_terminal_converts_every_range_exhaustion() {
418        let position = Position::Transactions {
419            checkpoint: 9,
420            tx_seq: 4,
421        };
422        let cases = [
423            (
424                RangeExhaustion::LedgerTip,
425                Some(8),
426                ScanTerminal::NaturalRange {
427                    end: NaturalRangeEnd::LedgerTip,
428                    position,
429                    claimed_checkpoint: Some(8),
430                },
431            ),
432            (
433                RangeExhaustion::CheckpointBound,
434                None,
435                ScanTerminal::NaturalRange {
436                    end: NaturalRangeEnd::CheckpointBound,
437                    position,
438                    claimed_checkpoint: None,
439                },
440            ),
441            (
442                RangeExhaustion::CursorBound {
443                    kind: CursorKind::Item,
444                },
445                Some(8),
446                ScanTerminal::CursorBound {
447                    position,
448                    kind: CursorKind::Item,
449                },
450            ),
451            (
452                RangeExhaustion::CursorBound {
453                    kind: CursorKind::Item,
454                },
455                None,
456                ScanTerminal::CursorBound {
457                    position,
458                    kind: CursorKind::Item,
459                },
460            ),
461            (
462                RangeExhaustion::ServingFloor {
463                    floor_checkpoint: 7,
464                },
465                Some(7),
466                ScanTerminal::ServingFloor {
467                    position,
468                    floor_checkpoint: 7,
469                },
470            ),
471        ];
472
473        for (exhaustion, claimed_checkpoint, expected) in cases {
474            assert_eq!(
475                ScanTerminal::from_range_exhaustion(exhaustion, position, claimed_checkpoint),
476                expected
477            );
478        }
479    }
480
481    /// A serving-floor terminal renders a boundary cursor at the floor position claiming the floor
482    /// checkpoint, whatever coverage the items accumulated, and carries no `QueryEnd` reason: its
483    /// stream fails with `OutOfRange` instead.
484    #[test]
485    fn scan_terminal_serving_floor_claims_the_floor_and_fails_the_stream() {
486        let position = Position::Transactions {
487            checkpoint: 7,
488            tx_seq: 35,
489        };
490        let terminal = ScanTerminal::ServingFloor {
491            position,
492            floor_checkpoint: 7,
493        };
494
495        let status = terminal
496            .reason()
497            .expect_err("a serving floor ends with no QueryEnd")
498            .into_status_proto();
499        assert_eq!(status.code, tonic::Code::OutOfRange as i32);
500        assert_eq!(
501            status.message,
502            "requested data below earliest available; lowest available checkpoint is 7"
503        );
504
505        for covered_checkpoint_bound in [None, Some(9)] {
506            let watermark = terminal.clone().into_watermark(covered_checkpoint_bound);
507            assert_eq!(
508                watermark.cursor,
509                Some(CursorToken::boundary(position).encode())
510            );
511            assert_eq!(watermark.checkpoint, Some(7));
512        }
513    }
514
515    /// A natural end claims its `claimed_checkpoint` behind a boundary cursor at its position,
516    /// whatever coverage the items accumulated.
517    #[test]
518    fn scan_terminal_natural_range_claims_its_claimed_checkpoint() {
519        let position = Position::Transactions {
520            checkpoint: 9,
521            tx_seq: 4,
522        };
523
524        for claimed_checkpoint in [8, 9] {
525            let checkpoint_bound = ScanTerminal::NaturalRange {
526                end: NaturalRangeEnd::CheckpointBound,
527                position,
528                claimed_checkpoint: Some(claimed_checkpoint),
529            };
530            assert_eq!(
531                checkpoint_bound.reason().unwrap(),
532                QueryEndReason::CheckpointBound
533            );
534            let watermark = checkpoint_bound.into_watermark(None);
535            assert_eq!(
536                watermark.cursor,
537                Some(CursorToken::boundary(position).encode())
538            );
539            assert_eq!(watermark.checkpoint, Some(claimed_checkpoint));
540
541            let ledger_tip = ScanTerminal::NaturalRange {
542                end: NaturalRangeEnd::LedgerTip,
543                position,
544                claimed_checkpoint: Some(claimed_checkpoint),
545            };
546            assert_eq!(ledger_tip.reason().unwrap(), QueryEndReason::LedgerTip);
547            let watermark = ledger_tip.into_watermark(Some(6));
548            assert_eq!(
549                watermark.cursor,
550                Some(CursorToken::boundary(position).encode())
551            );
552            assert_eq!(watermark.checkpoint, Some(claimed_checkpoint));
553        }
554    }
555
556    /// A natural end with no claimed checkpoint (its window has no checkpoint this server
557    /// serves) keeps its terminal cursor and leaves the checkpoint claim unset, for both natural
558    /// reasons and whatever coverage the items accumulated.
559    #[test]
560    fn scan_terminal_unclaimed_natural_end_claims_nothing() {
561        let position = Position::Transactions {
562            checkpoint: 9,
563            tx_seq: 4,
564        };
565
566        for (end, reason) in [
567            (
568                NaturalRangeEnd::CheckpointBound,
569                QueryEndReason::CheckpointBound,
570            ),
571            (NaturalRangeEnd::LedgerTip, QueryEndReason::LedgerTip),
572        ] {
573            let terminal = ScanTerminal::NaturalRange {
574                end,
575                position,
576                claimed_checkpoint: None,
577            };
578            assert_eq!(terminal.reason().unwrap(), reason);
579            let watermark = terminal.clone().into_watermark(None);
580            assert_eq!(
581                watermark.cursor,
582                Some(CursorToken::boundary(position).encode())
583            );
584            assert_eq!(watermark.checkpoint, None);
585            let watermark = terminal.into_watermark(Some(6));
586            assert_eq!(
587                watermark.cursor,
588                Some(CursorToken::boundary(position).encode())
589            );
590            assert_eq!(watermark.checkpoint, None);
591        }
592    }
593
594    #[test]
595    fn scan_terminal_cursor_bound_preserves_coverage() {
596        let position = Position::Transactions {
597            checkpoint: 9,
598            tx_seq: 4,
599        };
600        let terminal = ScanTerminal::CursorBound {
601            position,
602            kind: CursorKind::Boundary,
603        };
604        assert_eq!(terminal.reason().unwrap(), QueryEndReason::CursorBound);
605
606        let watermark = terminal.clone().into_watermark(Some(6));
607        assert_eq!(
608            watermark.cursor,
609            Some(CursorToken::boundary(position).encode())
610        );
611        assert_eq!(watermark.checkpoint, Some(6));
612
613        let watermark = terminal.into_watermark(None);
614        assert_eq!(
615            watermark.cursor,
616            Some(CursorToken::boundary(position).encode())
617        );
618        assert_eq!(watermark.checkpoint, None);
619    }
620
621    #[test]
622    fn scan_terminal_event_cursor_bound_preserves_item_kind() {
623        let position = Position::Events {
624            checkpoint: 9,
625            tx_seq: 4,
626            event_index: 2,
627        };
628        let terminal = ScanTerminal::CursorBound {
629            position,
630            kind: CursorKind::Item,
631        };
632        assert_eq!(terminal.reason().unwrap(), QueryEndReason::CursorBound);
633
634        let watermark = terminal.into_watermark(Some(6));
635        assert_eq!(watermark.cursor, Some(CursorToken::item(position).encode()));
636        assert_eq!(watermark.checkpoint, Some(6));
637    }
638
639    #[test]
640    fn scan_terminal_scan_limit_returns_owned_watermark() {
641        let mut watermark = Watermark::default();
642        watermark.cursor = Some(b"scan-limit".to_vec().into());
643        watermark.checkpoint = Some(6);
644        let terminal = ScanTerminal::ScanLimit {
645            watermark: watermark.clone(),
646        };
647        assert_eq!(terminal.reason().unwrap(), QueryEndReason::ScanLimit);
648        assert_eq!(terminal.into_watermark(None), watermark);
649
650        let terminal = ScanTerminal::ScanLimit {
651            watermark: watermark.clone(),
652        };
653        assert_eq!(terminal.into_watermark(Some(99)), watermark);
654    }
655}
sui_rpc_api/ledger_history/watermark.rs

sui_rpc_api/ledger_history/
watermark.rs
