remove skip_watermark, which allows us to standardize how to update the starting ingestion checkpoint, and each pipeline's checkpoint to start processing from. rework tests

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);
 
@@ -333,7 +326,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,13 +30,12 @@ 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>(
     config: CommitterConfig,
-    skip_watermark: bool,
     rx: mpsc::Receiver<BatchedRows<H>>,
     tx: mpsc::Sender<Vec<WatermarkPart>>,
     db: H::Store,
@@ -187,7 +186,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);
                         }
@@ -318,8 +317,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();
@@ -329,16 +327,7 @@ mod tests {
 
         let store_clone = store.clone();
         let committer_handle = tokio::spawn(async move {
-            let _ = committer(
-                config,
-                skip_watermark,
-                batch_rx,
-                watermark_tx,
-                store_clone,
-                metrics,
-                cancel,
-            )
-            .await;
+            let _ = committer(config, batch_rx, watermark_tx, store_clone, metrics, cancel).await;
         });
 
         TestSetup {
@@ -351,7 +340,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 {
@@ -434,7 +423,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 {
@@ -503,7 +492,7 @@ mod tests {
             })),
             ..Default::default()
         };
-        let mut setup = setup_test(store, false).await;
+        let mut setup = setup_test(store).await;
 
         let batch = BatchedRows {
             values: vec![StoredData {
@@ -560,7 +549,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 {
             values: vec![], // Empty values
@@ -599,54 +588,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 {
-            values: vec![StoredData {
-                cp_sequence_number: 1,
-                tx_sequence_numbers: vec![1, 2, 3],
-                ..Default::default()
-            }],
-            watermark: 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 {
             values: vec![StoredData {

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

@@ -177,8 +177,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
@@ -188,7 +187,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<CheckpointData>>,
     metrics: Arc<IndexerMetrics>,
@@ -236,7 +234,6 @@ pub(crate) fn pipeline<H: Handler + Send + Sync + 'static>(
 
     let committer = committer::<H>(
         committer_config.clone(),
-        skip_watermark,
         committer_rx,
         committer_tx,
         store.clone(),
@@ -247,7 +244,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(),
@@ -391,12 +387,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

@@ -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

@@ -6,11 +6,11 @@ keywords: [ sui indexer performance, indexer optimization, tokio console debuggi
 
 Proper configuration and resource monitoring delivers the most performant custom indexer possible. For example:
 
-- Runtime configuration options for ingestion, database connections, and pipeline selection, as well as purposeful use of debugging tools like `tokio_console` help dial in your indexer performance. 
+- Runtime configuration options for ingestion, database connections, and pipeline selection, as well as purposeful use of debugging tools like `tokio_console` help dial in your indexer performance.
 
-- A sensible strategy targeting efficient data pruning for your tables keeps them performant over time. 
+- A sensible strategy targeting efficient data pruning for your tables keeps them performant over time.
 
-- Following best practices for exposing and extending Prometheus metrics helps you keep track of indexer performance. 
+- Following best practices for exposing and extending Prometheus metrics helps you keep track of indexer performance.
 
 Together, these techniques help you run indexers that are fast, resource-efficient, and easier to monitor in both development and production.
 
@@ -26,10 +26,10 @@ Control how checkpoint data is fetched and distributed:
 let ingestion_config = IngestionConfig {
     // Buffer size across all downstream workers (default: 5000)
     checkpoint_buffer_size: 10000,
-    
+
     // Concurrent checkpoint fetches (default: 200)
     ingest_concurrency: 500,
-    
+
     // Retry interval for missing checkpoints in ms (default: 200)
     retry_interval_ms: 100,
 };
@@ -47,10 +47,10 @@ let ingestion_config = IngestionConfig {
 let db_args = DbArgs {
     // Connection pool size (default: 100)
     db_connection_pool_size: 200,
-    
+
     // Connection timeout in ms (default: 60,000)
     db_connection_timeout_ms: 30000,
-    
+
     // Statement timeout in ms (default: None)
     db_statement_timeout_ms: Some(120000),
 };
@@ -64,7 +64,7 @@ let db_args = DbArgs {
 
 ### Command-line arguments
 
-Include the following command-line arguments to help focus processing. These values are for demonstration. Use values that make sense to your environment and goals. 
+Include the following command-line arguments to help focus processing. These values are for demonstration. Use values that make sense to your environment and goals.
 
 ```sh
 # Checkpoint range control
@@ -74,9 +74,6 @@ Include the following command-line arguments to help focus processing. These val
 # Pipeline selection
 --pipeline "tx_counts"          # Run specific pipeline only
 --pipeline "events"             # Can specify multiple pipelines
-
-# Watermark behavior
---skip-watermark  
 ```
 
 **Use cases:**
@@ -256,7 +253,7 @@ Pipelines with more complex pruning rules can still benefit. For example, in con
 
 ### Implementation
 
-You can use `pg_partman` to simplify partition management. Configure a partitioned table with `create_parent`, followed by a cron job to periodically `run_maintenance`. You might need to iterate to determine the correct frequency of `run_maintenance`. 
+You can use `pg_partman` to simplify partition management. Configure a partitioned table with `create_parent`, followed by a cron job to periodically `run_maintenance`. You might need to iterate to determine the correct frequency of `run_maintenance`.
 
 :::info
 
@@ -319,7 +316,7 @@ BEGIN
         -- Table is already managed
         result_message := 'EXISTS: Table ' || full_table_name || ' is already managed by pg_partman';
     END IF;
-    
+
     RETURN result_message;
 END;
 $$ LANGUAGE plpgsql;
@@ -347,7 +344,7 @@ FROM part_config
 ORDER BY parent_table;
 
 -- Check maintenance job is scheduled
-SELECT jobid, schedule, command, nodename, database, username 
+SELECT jobid, schedule, command, nodename, database, username
 FROM cron.job
 WHERE command = 'SELECT run_maintenance()';
 ```