[sui-indexer-alt-framework] Address first_checkpoint + pipeline with no watermark row commit stall (#23209) (#23754)

## Description

Currently, if `first_checkpoint` is configured and a pipeline with no
watermark row is added to the indexer, it'll essentially stall since the
pipeline's committer task creates a default watermark with
checkpoint_hi_inclusive = 0 and waits for this checkpoint. This prevents
someone from using `first_checkpoint` to arbitrarily seed the starting
point of pipelines for the first time. Consequently, indexer operators
have to manually update the watermarks table to set a starting point
that isn't 0.

In this PR, the indexer is now responsible for determining the
`next_checkpoint` to commit and make watermark updates for each
pipeline. In practice, for a valid `first_checkpoint`
(`next_checkpoint`), sequential pipelines will wait to commit the next
processed checkpoint after its watermark, while concurrent pipelines
will start from the same `first_checkpoint`.

Meanwhile, the pipeline tasks initialize a default watermark variable
for the logger, but in practice will always be overwritten by the next
processed checkpoint for metrics and other reporting.

## Test plan 

- Unit tests for the behavior above
- commit_watermark and sequential committer 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:

## Description 

Describe the changes or additions included in this PR.

## Test plan 

How did you test the new or updated feature?

---

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

@@ -1,4 +1,4 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-pub mod mock_store;
+pub mod store;

…indexer-alt-framework/src/testing/mod.rs …i-indexer-alt-framework/src/mocks/mod.rscrates/sui-indexer-alt-framework/src/testing/mod.rs renamed to crates/sui-indexer-alt-framework/src/mocks/mod.rs crates/sui-indexer-alt-framework/src/testing/mod.rs renamed to crates/sui-indexer-alt-framework/src/mocks/mod.rs

@@ -52,12 +52,14 @@ pub struct Failures {
     pub attempts: AtomicUsize,
 }
 
-/// A mock store for testing. It maintains a map of checkpoint sequence numbers to transaction
-/// sequence numbers, and a watermark that can be used to test the watermark task.
+/// A mock store for testing. Represents an indexer with a single pipeline. It maintains a map of
+/// checkpoint sequence numbers to transaction sequence numbers, and a watermark that can be used to
+/// test the watermark task.
 #[derive(Clone, Default)]
 pub struct MockStore {
-    /// Tracks various watermark states (committer, reader, pruner)
-    pub watermarks: Arc<Mutex<MockWatermark>>,
+    /// Tracks various watermark states (committer, reader, pruner). This value can be optional
+    /// until a checkpoint is committed.
+    pub watermark: Arc<Mutex<Option<MockWatermark>>>,
     /// Stores the actual data, mapping checkpoint sequence numbers to transaction sequence numbers
     pub data: Arc<Mutex<HashMap<u64, Vec<u64>>>>,
     /// Tracks the order of checkpoint processing for testing sequential processing
@@ -88,23 +90,23 @@ impl Connection for MockConnection<'_> {
         &mut self,
         _pipeline: &'static str,
     ) -> Result<Option<CommitterWatermark>, anyhow::Error> {
-        let watermarks = self.0.watermarks.lock().unwrap();
-        Ok(Some(CommitterWatermark {
-            epoch_hi_inclusive: watermarks.epoch_hi_inclusive,
-            checkpoint_hi_inclusive: watermarks.checkpoint_hi_inclusive,
-            tx_hi: watermarks.tx_hi,
-            timestamp_ms_hi_inclusive: watermarks.timestamp_ms_hi_inclusive,
+        let watermark = self.0.watermark();
+        Ok(watermark.map(|w| CommitterWatermark {
+            epoch_hi_inclusive: w.epoch_hi_inclusive,
+            checkpoint_hi_inclusive: w.checkpoint_hi_inclusive,
+            tx_hi: w.tx_hi,
+            timestamp_ms_hi_inclusive: w.timestamp_ms_hi_inclusive,
         }))
     }
 
     async fn reader_watermark(
         &mut self,
         _pipeline: &'static str,
     ) -> Result<Option<ReaderWatermark>, anyhow::Error> {
-        let watermarks = self.0.watermarks.lock().unwrap();
-        Ok(Some(ReaderWatermark {
-            checkpoint_hi_inclusive: watermarks.checkpoint_hi_inclusive,
-            reader_lo: watermarks.reader_lo,
+        let watermark = self.0.watermark();
+        Ok(watermark.map(|w| ReaderWatermark {
+            checkpoint_hi_inclusive: w.checkpoint_hi_inclusive,
+            reader_lo: w.reader_lo,
         }))
     }
 
@@ -113,17 +115,19 @@ impl Connection for MockConnection<'_> {
         _pipeline: &'static str,
         delay: Duration,
     ) -> Result<Option<PrunerWatermark>, anyhow::Error> {
-        let watermarks = self.0.watermarks.lock().unwrap();
-        let elapsed_ms = watermarks.pruner_timestamp as i64
-            - SystemTime::now()
-                .duration_since(UNIX_EPOCH)
-                .unwrap()
-                .as_millis() as i64;
-        let wait_for_ms = delay.as_millis() as i64 + elapsed_ms;
-        Ok(Some(PrunerWatermark {
-            pruner_hi: watermarks.pruner_hi,
-            reader_lo: watermarks.reader_lo,
-            wait_for_ms,
+        let watermark = self.0.watermark();
+        Ok(watermark.map(|w| {
+            let elapsed_ms = w.pruner_timestamp as i64
+                - SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .unwrap()
+                    .as_millis() as i64;
+            let wait_for_ms = delay.as_millis() as i64 + elapsed_ms;
+            PrunerWatermark {
+                pruner_hi: w.pruner_hi,
+                reader_lo: w.reader_lo,
+                wait_for_ms,
+            }
         }))
     }
 
@@ -132,11 +136,16 @@ impl Connection for MockConnection<'_> {
         _pipeline: &'static str,
         watermark: CommitterWatermark,
     ) -> anyhow::Result<bool> {
-        let mut watermarks = self.0.watermarks.lock().unwrap();
-        watermarks.epoch_hi_inclusive = watermark.epoch_hi_inclusive;
-        watermarks.checkpoint_hi_inclusive = watermark.checkpoint_hi_inclusive;
-        watermarks.tx_hi = watermark.tx_hi;
-        watermarks.timestamp_ms_hi_inclusive = watermark.timestamp_ms_hi_inclusive;
+        let mut curr = self.0.watermark.lock().unwrap();
+        *curr = Some(MockWatermark {
+            epoch_hi_inclusive: watermark.epoch_hi_inclusive,
+            checkpoint_hi_inclusive: watermark.checkpoint_hi_inclusive,
+            tx_hi: watermark.tx_hi,
+            timestamp_ms_hi_inclusive: watermark.timestamp_ms_hi_inclusive,
+            reader_lo: curr.as_ref().map(|w| w.reader_lo).unwrap_or(0),
+            pruner_timestamp: curr.as_ref().map(|w| w.pruner_timestamp).unwrap_or(0),
+            pruner_hi: curr.as_ref().map(|w| w.pruner_hi).unwrap_or(0),
+        });
         Ok(true)
     }
 
@@ -160,12 +169,15 @@ impl Connection for MockConnection<'_> {
             return Err(anyhow::anyhow!("set_reader_watermark failed"));
         }
 
-        let mut watermarks = self.0.watermarks.lock().unwrap();
-        watermarks.reader_lo = reader_lo;
-        watermarks.pruner_timestamp = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .unwrap()
-            .as_millis() as u64;
+        let mut curr = self.0.watermark.lock().unwrap();
+        *curr = Some(MockWatermark {
+            reader_lo,
+            pruner_timestamp: SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_millis() as u64,
+            ..curr.as_ref().unwrap().clone()
+        });
         Ok(true)
     }
 
@@ -174,8 +186,11 @@ impl Connection for MockConnection<'_> {
         _pipeline: &'static str,
         pruner_hi: u64,
     ) -> anyhow::Result<bool> {
-        let mut watermarks = self.0.watermarks.lock().unwrap();
-        watermarks.pruner_hi = pruner_hi;
+        let mut curr = self.0.watermark.lock().unwrap();
+        *curr = Some(MockWatermark {
+            pruner_hi,
+            ..curr.as_ref().unwrap().clone()
+        });
         Ok(true)
     }
 }
@@ -350,9 +365,9 @@ impl MockStore {
         self.sequential_checkpoint_data.lock().unwrap().clone()
     }
 
-    /// Helper to get the current watermark state for testing
-    pub fn get_watermark(&self) -> MockWatermark {
-        self.watermarks.lock().unwrap().clone()
+    /// Helper to get the current watermark state for testing.
+    pub fn watermark(&self) -> Option<MockWatermark> {
+        self.watermark.lock().unwrap().clone()
     }
 
     /// Helper to get the number of connection attempts for testing
@@ -440,9 +455,10 @@ impl MockStore {
     ) -> MockWatermark {
         let start = std::time::Instant::now();
         while start.elapsed() < timeout_duration {
-            let watermark = self.get_watermark();
-            if watermark.checkpoint_hi_inclusive >= checkpoint {
-                return watermark;
+            if let Some(watermark) = self.watermark() {
+                if watermark.checkpoint_hi_inclusive >= checkpoint {
+                    return watermark;
+                }
             }
             tokio::time::sleep(Duration::from_millis(50)).await;
         }

…-alt-framework/src/testing/mock_store.rs …indexer-alt-framework/src/mocks/store.rscrates/sui-indexer-alt-framework/src/testing/mock_store.rs renamed to crates/sui-indexer-alt-framework/src/mocks/store.rs crates/sui-indexer-alt-framework/src/testing/mock_store.rs renamed to crates/sui-indexer-alt-framework/src/mocks/store.rs

@@ -44,7 +44,7 @@ use super::Handler;
 /// the watermark cannot be progressed. If `skip_watermark` is set, the task will shutdown
 /// immediately.
 pub(super) fn commit_watermark<H: Handler + 'static>(
-    initial_watermark: Option<CommitterWatermark>,
+    mut next_checkpoint: u64,
     config: CommitterConfig,
     skip_watermark: bool,
     mut rx: mpsc::Receiver<Vec<WatermarkPart>>,
@@ -67,12 +67,9 @@ pub(super) fn commit_watermark<H: Handler + 'static>(
         // watermark as much as possible without going over any holes in the sequence of
         // checkpoints (entirely missing watermarks, or incomplete watermarks).
         let mut precommitted: BTreeMap<u64, WatermarkPart> = BTreeMap::new();
-        let (mut watermark, mut next_checkpoint) = if let Some(watermark) = initial_watermark {
-            let next = watermark.checkpoint_hi_inclusive + 1;
-            (watermark, next)
-        } else {
-            (CommitterWatermark::default(), 0)
-        };
+        // Initially, this watermark is synthetic, and will be overwritten by a processed
+        // checkpoint.
+        let mut watermark = CommitterWatermark::default();
 
         // The watermark task will periodically output a log message at a higher log level to
         // demonstrate that the pipeline is making progress.
@@ -84,7 +81,10 @@ pub(super) fn commit_watermark<H: Handler + 'static>(
             &metrics.watermark_checkpoint_in_db,
         );
 
-        info!(pipeline = H::NAME, ?watermark, "Starting commit watermark");
+        info!(
+            pipeline = H::NAME,
+            next_checkpoint, "Starting commit watermark task"
+        );
 
         loop {
             tokio::select! {
@@ -275,9 +275,9 @@ mod tests {
 
     use crate::{
         metrics::IndexerMetrics,
+        mocks::store::*,
         pipeline::{CommitterConfig, Processor, WatermarkPart},
         store::CommitterWatermark,
-        testing::mock_store::*,
         FieldCount,
     };
 
@@ -318,7 +318,7 @@ mod tests {
 
     fn setup_test<H: Handler<Store = MockStore> + 'static>(
         config: CommitterConfig,
-        initial_watermark: Option<CommitterWatermark>,
+        next_checkpoint: u64,
         store: MockStore,
     ) -> TestSetup {
         let (watermark_tx, watermark_rx) = mpsc::channel(100);
@@ -329,7 +329,7 @@ mod tests {
         let cancel_clone = cancel.clone();
 
         let commit_watermark_handle = commit_watermark::<H>(
-            initial_watermark,
+            next_checkpoint,
             config,
             false,
             watermark_rx,
@@ -360,11 +360,7 @@ mod tests {
     #[tokio::test]
     async fn test_basic_watermark_progression() {
         let config = CommitterConfig::default();
-        let initial_watermark = Some(CommitterWatermark {
-            checkpoint_hi_inclusive: 0,
-            ..Default::default()
-        });
-        let setup = setup_test::<DataPipeline>(config, initial_watermark, MockStore::default());
+        let setup = setup_test::<DataPipeline>(config, 1, MockStore::default());
 
         // Send watermark parts in order
         for cp in 1..4 {
@@ -376,7 +372,7 @@ mod tests {
         tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
 
         // Verify watermark progression
-        let watermark = setup.store.get_watermark();
+        let watermark = setup.store.watermark().unwrap();
         assert_eq!(watermark.checkpoint_hi_inclusive, 3);
 
         // Clean up
@@ -387,11 +383,7 @@ mod tests {
     #[tokio::test]
     async fn test_out_of_order_watermarks() {
         let config = CommitterConfig::default();
-        let initial_watermark = Some(CommitterWatermark {
-            checkpoint_hi_inclusive: 0,
-            ..Default::default()
-        });
-        let setup = setup_test::<DataPipeline>(config, initial_watermark, MockStore::default());
+        let setup = setup_test::<DataPipeline>(config, 1, MockStore::default());
 
         // Send watermark parts out of order
         let parts = vec![
@@ -405,7 +397,7 @@ mod tests {
         tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
 
         // Verify watermark hasn't progressed past 2
-        let watermark = setup.store.get_watermark();
+        let watermark = setup.store.watermark().unwrap();
         assert_eq!(watermark.checkpoint_hi_inclusive, 2);
 
         // Send checkpoint 3 to fill the gap
@@ -419,7 +411,7 @@ mod tests {
         tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
 
         // Verify watermark has progressed to 4
-        let watermark = setup.store.get_watermark();
+        let watermark = setup.store.watermark().unwrap();
         assert_eq!(watermark.checkpoint_hi_inclusive, 4);
 
         // Clean up
@@ -433,12 +425,8 @@ mod tests {
             watermark_interval_ms: 1_000, // Long polling interval to test connection retry
             ..Default::default()
         };
-        let initial_watermark = Some(CommitterWatermark {
-            checkpoint_hi_inclusive: 0,
-            ..Default::default()
-        });
         let store = MockStore::default().with_connection_failures(1);
-        let setup = setup_test::<DataPipeline>(config, initial_watermark, store);
+        let setup = setup_test::<DataPipeline>(config, 1, store);
 
         // Send watermark part
         let part = create_watermark_part_for_checkpoint(1);
@@ -448,14 +436,14 @@ mod tests {
         tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
 
         // Verify watermark hasn't progressed
-        let watermark = setup.store.get_watermark();
-        assert_eq!(watermark.checkpoint_hi_inclusive, 0);
+        let watermark = setup.store.watermark();
+        assert!(watermark.is_none());
 
         // Wait for next polling and processing
         tokio::time::sleep(tokio::time::Duration::from_millis(1_200)).await;
 
         // Verify watermark has progressed
-        let watermark = setup.store.get_watermark();
+        let watermark = setup.store.watermark().unwrap();
         assert_eq!(watermark.checkpoint_hi_inclusive, 1);
 
         // Clean up
@@ -469,11 +457,7 @@ mod tests {
             watermark_interval_ms: 1_000, // Long polling interval to test adding complete part
             ..Default::default()
         };
-        let initial_watermark = Some(CommitterWatermark {
-            checkpoint_hi_inclusive: 0,
-            ..Default::default()
-        });
-        let setup = setup_test::<DataPipeline>(config, initial_watermark, MockStore::default());
+        let setup = setup_test::<DataPipeline>(config, 1, MockStore::default());
 
         // Send the first incomplete watermark part
         let part = WatermarkPart {
@@ -490,8 +474,8 @@ mod tests {
         tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
 
         // Verify watermark hasn't progressed
-        let watermark = setup.store.get_watermark();
-        assert_eq!(watermark.checkpoint_hi_inclusive, 0);
+        let watermark = setup.store.watermark();
+        assert!(watermark.is_none());
 
         // Send the other two parts to complete the watermark
         setup
@@ -504,7 +488,7 @@ mod tests {
         tokio::time::sleep(tokio::time::Duration::from_millis(1_200)).await;
 
         // Verify watermark has progressed
-        let watermark = setup.store.get_watermark();
+        let watermark = setup.store.watermark().unwrap();
         assert_eq!(watermark.checkpoint_hi_inclusive, 1);
 
         // Clean up
@@ -515,8 +499,7 @@ mod tests {
     #[tokio::test]
     async fn test_no_initial_watermark() {
         let config = CommitterConfig::default();
-        let initial_watermark = None;
-        let setup = setup_test::<DataPipeline>(config, initial_watermark, MockStore::default());
+        let setup = setup_test::<DataPipeline>(config, 0, MockStore::default());
 
         // Send the checkpoint 1 watermark
         setup
@@ -529,8 +512,8 @@ mod tests {
         tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
 
         // Verify watermark hasn't progressed
-        let watermark = setup.store.get_watermark();
-        assert_eq!(watermark.checkpoint_hi_inclusive, 0);
+        let watermark = setup.store.watermark();
+        assert!(watermark.is_none());
 
         // Send the checkpoint 0 watermark to fill the gap.
         setup
@@ -543,7 +526,7 @@ mod tests {
         tokio::time::sleep(tokio::time::Duration::from_millis(1200)).await;
 
         // Verify watermark has progressed
-        let watermark = setup.store.get_watermark();
+        let watermark = setup.store.watermark().unwrap();
         assert_eq!(watermark.checkpoint_hi_inclusive, 1);
 
         // Clean up