sui_indexer_alt_framework/pipeline/
mod.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::time::Duration;
5
6pub use crate::config::ConcurrencyConfig;
7use crate::store::CommitterWatermark;
8pub use processor::Processor;
9use rand::Rng;
10use serde::Deserialize;
11use serde::Serialize;
12
13mod batch_floor;
14pub mod concurrent;
15mod logging;
16mod processor;
17pub mod sequential;
18
19/// Issue a warning every time the number of pending watermarks exceeds this number. This can
20/// happen if the pipeline was started with its initial checkpoint overridden to be strictly
21/// greater than its current watermark -- in that case, the pipeline will never be able to update
22/// its watermarks.
23const WARN_PENDING_WATERMARKS: usize = 10000;
24
25#[derive(Serialize, Deserialize, Debug, Clone)]
26pub struct CommitterConfig {
27    /// Number of concurrent writers per pipeline.
28    pub write_concurrency: usize,
29
30    /// Minimum time between two watermark writes of a concurrent pipeline, in milliseconds. A
31    /// watermark that advances once this much time has passed since the previous write is written
32    /// immediately; one that advances sooner is written when the gap has passed.
33    pub watermark_interval_ms: u64,
34
35    /// Maximum random jitter to add to the watermark interval, in milliseconds.
36    pub watermark_interval_jitter_ms: u64,
37
38    /// How many rows the collector waits for before it cuts them into a batch. With the default
39    /// of 0, pending rows are cut into a batch as soon as the committer has room for one, so
40    /// batches only grow while the committer is busy. Stores that prefer fewer, larger writes can
41    /// raise this; rows then wait until this many are pending, or until the oldest has waited
42    /// `max_batch_delay_ms`. A sequential pipeline also cuts a smaller batch once its reorder
43    /// buffer holds `max_pending_rows` rows and the next checkpoint has arrived.
44    #[serde(default)]
45    pub min_batch_rows: usize,
46
47    /// Longest time, in milliseconds, that pending rows wait for `min_batch_rows` to be reached
48    /// before they are written anyway. Not used while `min_batch_rows` is 0.
49    #[serde(default = "default_max_batch_delay_ms")]
50    pub max_batch_delay_ms: u64,
51}
52
53/// Per-pipeline ingestion settings.
54#[derive(Serialize, Deserialize, Debug, Clone, Default)]
55pub struct IngestionConfig {
56    /// Capacity of this pipeline's bounded subscriber channel. If `None`, the built-in default
57    /// is used (see [`IngestionConfig::subscriber_channel_size`]).
58    pub subscriber_channel_size: Option<usize>,
59}
60
61impl IngestionConfig {
62    /// Resolves `subscriber_channel_size` to its final value, substituting the built-in default
63    /// if unset.
64    ///
65    /// The default is small on purpose: the adaptive controller does the real backpressure work,
66    /// and larger values just pin more decoded checkpoints in memory without throughput benefit.
67    /// Scales with CPU count for fetch parallelism headroom, with a floor of 4 so the
68    /// controller's dead band (0.6..0.85) has integer room to maneuver on small machines.
69    pub fn subscriber_channel_size(&self) -> usize {
70        self.subscriber_channel_size
71            .unwrap_or_else(|| (num_cpus::get() / 2).max(4))
72    }
73}
74
75/// Processed values associated with a single checkpoint. This is an internal type used to
76/// communicate between the processor and the collector parts of the pipeline.
77struct IndexedCheckpoint<P: Processor> {
78    /// Values to be inserted into the database from this checkpoint
79    values: Vec<P::Value>,
80    /// The watermark associated with this checkpoint
81    watermark: CommitterWatermark,
82}
83
84/// A representation of the proportion of a watermark.
85#[derive(Debug, Clone)]
86struct WatermarkPart {
87    /// The watermark itself
88    watermark: CommitterWatermark,
89    /// The number of rows from this watermark that are in this part
90    batch_rows: usize,
91    /// The total number of rows from this watermark
92    total_rows: usize,
93}
94
95impl CommitterConfig {
96    pub fn max_batch_delay(&self) -> Duration {
97        Duration::from_millis(self.max_batch_delay_ms)
98    }
99
100    pub fn watermark_interval(&self) -> Duration {
101        Duration::from_millis(self.watermark_interval_ms)
102    }
103
104    /// Returns the earliest instant at which the next watermark write may start: now plus the
105    /// watermark interval plus a random jitter between 0 and `watermark_interval_jitter_ms`.
106    pub fn watermark_interval_with_jitter(&self) -> tokio::time::Instant {
107        let jitter = if self.watermark_interval_jitter_ms == 0 {
108            0
109        } else {
110            rand::thread_rng().gen_range(0..=self.watermark_interval_jitter_ms)
111        };
112        tokio::time::Instant::now() + Duration::from_millis(self.watermark_interval_ms + jitter)
113    }
114}
115
116impl<P: Processor> IndexedCheckpoint<P> {
117    fn new(
118        epoch: u64,
119        cp_sequence_number: u64,
120        tx_hi: u64,
121        timestamp_ms: u64,
122        values: Vec<P::Value>,
123    ) -> Self {
124        Self {
125            watermark: CommitterWatermark {
126                epoch_hi_inclusive: epoch,
127                checkpoint_hi_inclusive: cp_sequence_number,
128                tx_hi,
129                timestamp_ms_hi_inclusive: timestamp_ms,
130            },
131            values,
132        }
133    }
134
135    /// Number of rows from this checkpoint
136    fn len(&self) -> usize {
137        self.values.len()
138    }
139
140    /// The checkpoint sequence number that this data is from
141    fn checkpoint(&self) -> u64 {
142        self.watermark.checkpoint_hi_inclusive
143    }
144}
145
146impl WatermarkPart {
147    fn checkpoint(&self) -> u64 {
148        self.watermark.checkpoint_hi_inclusive
149    }
150
151    fn timestamp_ms(&self) -> u64 {
152        self.watermark.timestamp_ms_hi_inclusive
153    }
154
155    /// Check if all the rows from this watermark are represented in this part.
156    fn is_complete(&self) -> bool {
157        self.batch_rows == self.total_rows
158    }
159
160    /// Add the rows from `other` to this part.
161    fn add(&mut self, other: WatermarkPart) {
162        assert_eq!(self.checkpoint(), other.checkpoint());
163        self.batch_rows += other.batch_rows;
164        assert!(
165            self.batch_rows <= self.total_rows,
166            "batch_rows ({}) exceeded total_rows ({})",
167            self.batch_rows,
168            self.total_rows,
169        );
170    }
171
172    /// Record that `rows` have been taken from this part.
173    fn take(&mut self, rows: usize) -> WatermarkPart {
174        assert!(
175            self.batch_rows >= rows,
176            "Can't take more rows than are available"
177        );
178
179        self.batch_rows -= rows;
180        WatermarkPart {
181            watermark: self.watermark,
182            batch_rows: rows,
183            total_rows: self.total_rows,
184        }
185    }
186}
187
188impl Default for CommitterConfig {
189    fn default() -> Self {
190        Self {
191            write_concurrency: 5,
192            watermark_interval_ms: 500,
193            watermark_interval_jitter_ms: 0,
194            min_batch_rows: 0,
195            max_batch_delay_ms: default_max_batch_delay_ms(),
196        }
197    }
198}
199
200fn default_max_batch_delay_ms() -> u64 {
201    500
202}
203
204#[cfg(test)]
205mod tests {
206    use super::*;
207    use async_trait::async_trait;
208    use std::sync::Arc;
209    use sui_types::full_checkpoint_content::Checkpoint;
210
211    // Test implementation of Processor
212    struct TestProcessor;
213    #[async_trait]
214    impl Processor for TestProcessor {
215        const NAME: &'static str = "test";
216        type Value = i32;
217
218        async fn process(&self, _checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Self::Value>> {
219            Ok(vec![1, 2, 3])
220        }
221    }
222
223    #[test]
224    fn test_watermark_part_getters() {
225        let watermark = CommitterWatermark {
226            epoch_hi_inclusive: 1,
227            checkpoint_hi_inclusive: 100,
228            tx_hi: 1000,
229            timestamp_ms_hi_inclusive: 1234567890,
230        };
231
232        let part = WatermarkPart {
233            watermark,
234            batch_rows: 50,
235            total_rows: 200,
236        };
237
238        assert_eq!(part.checkpoint(), 100);
239        assert_eq!(part.timestamp_ms(), 1234567890);
240    }
241
242    #[test]
243    fn test_watermark_part_is_complete() {
244        let part = WatermarkPart {
245            watermark: CommitterWatermark::default(),
246            batch_rows: 200,
247            total_rows: 200,
248        };
249
250        assert!(part.is_complete());
251    }
252
253    #[test]
254    fn test_watermark_part_is_not_complete() {
255        let part = WatermarkPart {
256            watermark: CommitterWatermark::default(),
257            batch_rows: 199,
258            total_rows: 200,
259        };
260
261        assert!(!part.is_complete());
262    }
263
264    #[test]
265    fn test_watermark_part_becomes_complete_after_adding_new_batch() {
266        let mut part = WatermarkPart {
267            watermark: CommitterWatermark::default(),
268            batch_rows: 199,
269            total_rows: 200,
270        };
271
272        // Add a batch that makes it complete
273        part.add(WatermarkPart {
274            watermark: CommitterWatermark::default(),
275            batch_rows: 1,
276            total_rows: 200,
277        });
278
279        assert!(part.is_complete());
280        assert_eq!(part.batch_rows, 200);
281    }
282
283    #[test]
284    fn test_watermark_part_becomes_incomplete_after_taking_away_batch() {
285        let mut part = WatermarkPart {
286            watermark: CommitterWatermark::default(),
287            batch_rows: 200,
288            total_rows: 200,
289        };
290        assert!(part.is_complete(), "Initial part should be complete");
291
292        // Take away a portion of the batch
293        let extracted_part = part.take(10);
294
295        // Verify state of extracted part
296        assert!(!extracted_part.is_complete());
297        assert_eq!(extracted_part.batch_rows, 10);
298        assert_eq!(extracted_part.total_rows, 200);
299    }
300
301    #[test]
302    fn test_indexed_checkpoint() {
303        let epoch = 1;
304        let cp_sequence_number = 100;
305        let tx_hi = 1000;
306        let timestamp_ms = 1234567890;
307        let values = vec![1, 2, 3];
308
309        let checkpoint = IndexedCheckpoint::<TestProcessor>::new(
310            epoch,
311            cp_sequence_number,
312            tx_hi,
313            timestamp_ms,
314            values,
315        );
316
317        assert_eq!(checkpoint.len(), 3);
318        assert_eq!(checkpoint.checkpoint(), 100);
319    }
320
321    #[test]
322    fn test_indexed_checkpoint_with_empty_values() {
323        let epoch = 1;
324        let cp_sequence_number = 100;
325        let tx_hi = 1000;
326        let timestamp_ms = 1234567890;
327        let values: Vec<<TestProcessor as Processor>::Value> = vec![];
328
329        let checkpoint = IndexedCheckpoint::<TestProcessor>::new(
330            epoch,
331            cp_sequence_number,
332            tx_hi,
333            timestamp_ms,
334            values,
335        );
336
337        assert_eq!(checkpoint.len(), 0);
338        assert_eq!(checkpoint.checkpoint(), 100);
339    }
340}
sui_indexer_alt_framework/pipeline/mod.rs

sui_indexer_alt_framework/pipeline/
mod.rs
