[indexer-alt-framework] remove skip_watermark and standardize first_checkpoint behavior (#24192)

## Description 

This PR standardizes behavior around `--first-checkpoint` and its
effects on
A. The checkpoint to start ingesting from
B. The checkpoint for a pipeline to resume processing from

- When an indexer starts, it will always start ingesting from the next
checkpoint after the smallest committed checkpoint across all pipelines
added to it.
- Pipelines without a watermark are by default to be backfilled from
genesis.
- Providing `--first-checkpoint` overrides this behavior, and tells the
indexer that these pipelines are to process checkpoints from
`--first-checkpoint` instead.

## Test plan 

New tests

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] gRPC:
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:

Author: wlmyng

crates/sui-indexer-alt-framework/src/lib.rs

@@ -41,23 +41,16 @@ use super::Handler;
 /// [LOUD_WATERMARK_UPDATE_INTERVAL]-many checkpoints.
 ///
 /// The task will shutdown if the `cancel` token is signalled, or if the `rx` channel closes and
-/// the watermark cannot be progressed. If `skip_watermark` is set, the task will shutdown
-/// immediately.
+/// the watermark cannot be progressed.
 pub(super) fn commit_watermark<H: Handler + 'static>(
     mut next_checkpoint: u64,
     config: CommitterConfig,
-    skip_watermark: bool,
     mut rx: mpsc::Receiver<Vec<WatermarkPart>>,
     store: H::Store,
     metrics: Arc<IndexerMetrics>,
     cancel: CancellationToken,
 ) -> JoinHandle<()> {
     tokio::spawn(async move {
-        if skip_watermark {
-            info!(pipeline = H::NAME, "Skipping commit watermark task");
-            return;
-        }
-
         let mut poll = interval(config.watermark_interval());
         poll.set_missed_tick_behavior(MissedTickBehavior::Delay);
 
@@ -341,7 +334,6 @@ mod tests {
         let commit_watermark_handle = commit_watermark::<H>(
             next_checkpoint,
             config,
-            false,
             watermark_rx,
             store_clone,
             metrics,

crates/sui-indexer-alt-framework/src/pipeline/concurrent/commit_watermark.rs

@@ -30,14 +30,13 @@ const MAX_RETRY_INTERVAL: Duration = Duration::from_secs(1);
 ///
 /// The writing of each batch will be repeatedly retried on an exponential back-off until it
 /// succeeds. Once the write succeeds, the [WatermarkPart]s for that batch are sent on `tx` to the
-/// watermark task, as long as `skip_watermark` is not true.
+/// watermark task.
 ///
 /// This task will shutdown via its `cancel`lation token, or if its receiver or sender channels are
 /// closed.
 pub(super) fn committer<H: Handler + 'static>(
     handler: Arc<H>,
     config: CommitterConfig,
-    skip_watermark: bool,
     rx: mpsc::Receiver<BatchedRows<H>>,
     tx: mpsc::Sender<Vec<WatermarkPart>>,
     db: H::Store,
@@ -194,7 +193,7 @@ pub(super) fn committer<H: Handler + 'static>(
                             }
                         };
 
-                        if !skip_watermark && tx.send(watermark).await.is_err() {
+                        if tx.send(watermark).await.is_err() {
                             info!(pipeline = H::NAME, "Watermark closed channel");
                             return Err(Break::Cancel);
                         }
@@ -333,8 +332,7 @@ mod tests {
     ///
     /// # Arguments
     /// * `store` - The mock store to use for testing
-    /// * `skip_watermark` - Whether to skip sending watermarks to the watermark channel
-    async fn setup_test(store: MockStore, skip_watermark: bool) -> TestSetup {
+    async fn setup_test(store: MockStore) -> TestSetup {
         let config = CommitterConfig::default();
         let metrics = IndexerMetrics::new(None, &Default::default());
         let cancel = CancellationToken::new();
@@ -348,7 +346,6 @@ mod tests {
             let _ = committer(
                 handler,
                 config,
-                skip_watermark,
                 batch_rx,
                 watermark_tx,
                 store_clone,
@@ -368,7 +365,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_concurrent_batch_processing() {
-        let mut setup = setup_test(MockStore::default(), false).await;
+        let mut setup = setup_test(MockStore::default()).await;
 
         // Send batches
         let batch1 = BatchedRows::from_vec(
@@ -451,7 +448,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_commit_with_retries_for_commit_failure() {
-        let mut setup = setup_test(MockStore::default(), false).await;
+        let mut setup = setup_test(MockStore::default()).await;
 
         // Create a batch with a single item that will fail once before succeeding
         let batch = BatchedRows::from_vec(
@@ -520,7 +517,7 @@ mod tests {
             })),
             ..Default::default()
         };
-        let mut setup = setup_test(store, false).await;
+        let mut setup = setup_test(store).await;
 
         let batch = BatchedRows::from_vec(
             vec![StoredData {
@@ -577,7 +574,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_empty_batch_handling() {
-        let mut setup = setup_test(MockStore::default(), false).await;
+        let mut setup = setup_test(MockStore::default()).await;
 
         let empty_batch = BatchedRows::from_vec(
             vec![], // Empty batch
@@ -616,54 +613,9 @@ mod tests {
         let _ = setup.committer_handle.await;
     }
 
-    #[tokio::test]
-    async fn test_skip_watermark_mode() {
-        let mut setup = setup_test(MockStore::default(), true).await;
-
-        let batch = BatchedRows::from_vec(
-            vec![StoredData {
-                cp_sequence_number: 1,
-                tx_sequence_numbers: vec![1, 2, 3],
-                ..Default::default()
-            }],
-            vec![WatermarkPart {
-                watermark: CommitterWatermark {
-                    epoch_hi_inclusive: 0,
-                    checkpoint_hi_inclusive: 1,
-                    tx_hi: 3,
-                    timestamp_ms_hi_inclusive: 1000,
-                },
-                batch_rows: 1,
-                total_rows: 1,
-            }],
-        );
-
-        // Send the batch
-        setup.batch_tx.send(batch).await.unwrap();
-
-        // Wait for processing
-        tokio::time::sleep(Duration::from_millis(200)).await;
-
-        // Verify data was committed
-        {
-            let data = setup.store.data.get(DataPipeline::NAME).unwrap();
-            assert_eq!(data.get(&1).unwrap().value(), &vec![1, 2, 3]);
-        }
-
-        // Verify no watermark was sent (skip_watermark mode)
-        assert!(
-            setup.watermark_rx.try_recv().is_err(),
-            "No watermark should be sent in skip_watermark mode"
-        );
-
-        // Clean up
-        drop(setup.batch_tx);
-        let _ = setup.committer_handle.await;
-    }
-
     #[tokio::test]
     async fn test_watermark_channel_closed() {
-        let setup = setup_test(MockStore::default(), false).await;
+        let setup = setup_test(MockStore::default()).await;
 
         let batch = BatchedRows::from_vec(
             vec![StoredData {

crates/sui-indexer-alt-framework/src/pipeline/concurrent/committer.rs

@@ -194,8 +194,7 @@ impl Default for PrunerConfig {
 /// time.
 ///
 /// The pipeline also maintains a row in the `watermarks` table for the pipeline which tracks the
-/// watermark below which all data has been committed (modulo pruning), as long as `skip_watermark`
-/// is not true.
+/// watermark below which all data has been committed (modulo pruning).
 ///
 /// Checkpoint data is fed into the pipeline through the `checkpoint_rx` channel, and internal
 /// channels are created to communicate between its various components. The pipeline can be
@@ -205,7 +204,6 @@ pub(crate) fn pipeline<H: Handler + Send + Sync + 'static>(
     handler: H,
     next_checkpoint: u64,
     config: ConcurrentConfig,
-    skip_watermark: bool,
     store: H::Store,
     checkpoint_rx: mpsc::Receiver<Arc<Checkpoint>>,
     metrics: Arc<IndexerMetrics>,
@@ -255,7 +253,6 @@ pub(crate) fn pipeline<H: Handler + Send + Sync + 'static>(
     let committer = committer::<H>(
         handler.clone(),
         committer_config.clone(),
-        skip_watermark,
         committer_rx,
         committer_tx,
         store.clone(),
@@ -266,7 +263,6 @@ pub(crate) fn pipeline<H: Handler + Send + Sync + 'static>(
     let commit_watermark = commit_watermark::<H>(
         next_checkpoint,
         committer_config,
-        skip_watermark,
         watermark_rx,
         store.clone(),
         metrics.clone(),
@@ -412,12 +408,10 @@ mod tests {
             let metrics = IndexerMetrics::new(None, &Registry::default());
             let cancel = CancellationToken::new();
 
-            let skip_watermark = false;
             let pipeline_handle = pipeline(
                 DataPipeline,
                 next_checkpoint,
                 config,
-                skip_watermark,
                 store.clone(),
                 checkpoint_rx,
                 metrics,

crates/sui-indexer-alt-framework/src/pipeline/concurrent/mod.rs

@@ -21,9 +21,6 @@ const PIPELINE_BUFFER: usize = 5;
 /// happen if the pipeline was started with its initial checkpoint overridden to be strictly
 /// greater than its current watermark -- in that case, the pipeline will never be able to update
 /// its watermarks.
-///
-/// This may be a legitimate thing to do when backfilling a table, but in that case
-/// `--skip-watermarks` should be used.
 const WARN_PENDING_WATERMARKS: usize = 10000;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]

crates/sui-indexer-alt-framework/src/pipeline/mod.rs

@@ -160,7 +160,7 @@ pub mod tests {
             .concurrent_pipeline(ConcurrentPipeline1, ConcurrentConfig::default())
             .await
             .unwrap();
-        assert_eq!(indexer.first_checkpoint_from_watermark, 0);
+        assert_eq!(indexer.first_ingestion_checkpoint, 0);
     }
 
     #[tokio::test]
@@ -179,7 +179,7 @@ pub mod tests {
             .concurrent_pipeline(ConcurrentPipeline1, ConcurrentConfig::default())
             .await
             .unwrap();
-        assert_eq!(indexer.first_checkpoint_from_watermark, 11);
+        assert_eq!(indexer.first_ingestion_checkpoint, 11);
     }
 
     #[tokio::test]
@@ -205,11 +205,11 @@ pub mod tests {
             .concurrent_pipeline(ConcurrentPipeline2, ConcurrentConfig::default())
             .await
             .unwrap();
-        assert_eq!(indexer.first_checkpoint_from_watermark, 21);
+        assert_eq!(indexer.first_ingestion_checkpoint, 21);
         indexer
             .concurrent_pipeline(ConcurrentPipeline1, ConcurrentConfig::default())
             .await
             .unwrap();
-        assert_eq!(indexer.first_checkpoint_from_watermark, 11);
+        assert_eq!(indexer.first_ingestion_checkpoint, 11);
     }
 }

crates/sui-indexer-alt-framework/src/postgres/mod.rs

@@ -51,7 +51,6 @@ pub async fn run_benchmark(
         first_checkpoint: Some(first_checkpoint),
         last_checkpoint: Some(last_checkpoint),
         pipeline,
-        ..Default::default()
     };
 
     let client_args = ClientArgs {

crates/sui-indexer-alt/src/benchmark.rs

@@ -75,10 +75,6 @@ Include the following command-line arguments to help focus processing. These val
 --pipeline "tx_counts"          # Run specific pipeline only
 --pipeline "events"             # Can specify multiple pipelines
 
-# Watermark behavior
---skip-watermark  
-```
-
 **Use cases:**
 
 - **Checkpoint range:** Essential for backfills and historical data processing.