[indexer-alt-framework] watermark tasks (#23737)

## Description 

Building on #24192, we now have
new fields `--task` and `--reader-interval-ms`, which when set creates
an indexer whose pipelines commit checkpoint data as long as it is not
less than each of its main pipeline's `pruner_hi`.

Startup behavior remains the same as defined in the previous PR: all
pipelines, tasked or not, will resume committing from next after its
committer watermark, or expect 0, unless overridden by
`--first-checkpoint`.

The indexer determines ingestion as the smallest next checkpoint across
pipelines (as calculated above.)

To backfill, operators either:
- run a new indexer with `-task` and `-first-checkpoint` (preferred), or
- wipe watermark entries for main pipelines (without task) and restart
main indexer with `-first-checkpoint.`

Multiple tasked indexers can run in parallel with different ranges to
accelerate backfill.


## Test plan 

New tests in `lib.rs` and `collector.rs`

Additional tests in `collector`

---

## 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:
- [x] JSON-RPC: 
This PR, in tandem
with [https://github.com/MystenLabs/sui/pull/24192](https://github.com/MystenLabs/sui/pull/24192),
unifies how indexers determine their ingestion starting point and
introduce watermark-gated backfill tasks.

1. `--skip-watermark` is removed, and the previous ability to bypass
watermark safety checks for concurrent pipelines is no longer supported.
2. `--first-checkpoint` no longer forces the indexer to start ingesting
from the configured checkpoint. The indexer now always determines its
starting ingestion point as the minimum next checkpoint across all
pipelines to resume processing from. From this release,
`--first-checkpoint` now only applies to pipelines that do not yet have
a committer watermark. These pipelines will resume processing from the
configured value. Pipelines with existing watermarks will always resume
processing from their own next checkpoint.
3. A new mechanism, watermark tasks, allows operators to run the same
pipelines on multiple indexer instances for historical backfilling. Two
new flags, `--task` and `--reader-interval-ms`, enable this mechanism.
These flags create a tasked indexer whose pipelines commit checkpoint
data as long as the checkpoint is not below the `reader_lo` watermark of
their corresponding main pipelines. The indexer controls how frequently
these tasked pipelines poll the main pipelines' watermarks
per `--reader-interval-ms`.

Migration guidance:  
1. If you use `--first-checkpoint` only for _initial_ ingestion of a
fresh pipeline, no further action is needed.
2. If you previously used `--first-checkpoint` and
optionally `--skip-watermark` to backfill existing tables, you can
achieve the same workflow by starting a new indexer instance with a
configured `--task`, `--reader-interval-ms`, and `--first-checkpoint`.
3. Like `--skip-watermark`, `--task` cannot be used to run sequential
pipelines.
- [x] GraphQL: 
See above
- [ ] CLI: 
- [ ] Rust SDK:

Author: wlmyng

crates/sui-checkpoint-blob-indexer/src/lib.rs

@@ -18,7 +18,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_checkpoint_blob_handler_uncompressed() {
-        let store = ObjectStore::new(Arc::new(InMemory::new()), None);
+        let store = ObjectStore::new(Arc::new(InMemory::new()));
         let mut conn = store.connect().await.unwrap();
 
         let blob = CheckpointBlob {
@@ -40,7 +40,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_checkpoint_blob_handler_compressed() {
-        let store = ObjectStore::new(Arc::new(InMemory::new()), None);
+        let store = ObjectStore::new(Arc::new(InMemory::new()));
         let mut conn = store.connect().await.unwrap();
 
         let test_data = vec![0u8; 1000];
@@ -66,7 +66,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_epochs_handler() {
-        let store = ObjectStore::new(Arc::new(InMemory::new()), None);
+        let store = ObjectStore::new(Arc::new(InMemory::new()));
         let path = ObjectPath::from("epochs.json");
 
         // Test 1: Create new file with first epoch

crates/sui-checkpoint-blob-indexer/src/main.rs

@@ -126,7 +126,7 @@ async fn main() -> anyhow::Result<()> {
         unreachable!("clap ensures exactly one storage backend is provided");
     };
 
-    let store = ObjectStore::new(object_store, args.watermark_task);
+    let store = ObjectStore::new(object_store);
 
     let cancel = tokio_util::sync::CancellationToken::new();
 

crates/sui-indexer-alt-consistent-store/src/store/mod.rs

@@ -38,7 +38,7 @@ pub(crate) struct Store<S>(Arc<Inner<S>>);
 pub(crate) struct Connection<'s, S> {
     pub store: &'s Store<S>,
     pub batch: rocksdb::WriteBatch,
-    watermark: Option<(&'static str, Watermark)>,
+    watermark: Option<(String, Watermark)>,
 }
 
 /// The contents of the store.
@@ -147,7 +147,7 @@ impl<S: Send + Sync + 'static> store::TransactionalStore for Store<S> {
             .queue
             .get()
             .context("Synchronizer not running for store")?
-            .get(pipeline)
+            .get(pipeline.as_str())
             .with_context(|| format!("No {pipeline:?} synchronizer queue"))?
             .send((watermark, conn.batch))
             .await
@@ -161,17 +161,22 @@ impl<S: Send + Sync + 'static> store::TransactionalStore for Store<S> {
 impl<S: Send + Sync> store::Connection for Connection<'_, S> {
     async fn committer_watermark(
         &mut self,
-        pipeline: &'static str,
+        pipeline_task: &str,
     ) -> anyhow::Result<Option<CommitterWatermark>> {
-        Ok(self.store.0.db.commit_watermark(pipeline)?.map(Into::into))
+        Ok(self
+            .store
+            .0
+            .db
+            .commit_watermark(pipeline_task)?
+            .map(Into::into))
     }
 
     async fn set_committer_watermark(
         &mut self,
-        pipeline: &'static str,
+        pipeline_task: &str,
         watermark: CommitterWatermark,
     ) -> anyhow::Result<bool> {
-        self.watermark = Some((pipeline, watermark.into()));
+        self.watermark = Some((pipeline_task.to_string(), watermark.into()));
         Ok(true)
     }
 

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

@@ -1,6 +1,7 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+use anyhow::Result;
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use scoped_futures::ScopedBoxFuture;
@@ -10,11 +11,13 @@ use std::time::Duration;
 /// operations, agnostic of the underlying store implementation.
 #[async_trait]
 pub trait Connection: Send {
-    /// Given a pipeline, return the committer watermark from the `Store`. This is used by the
-    /// indexer on startup to determine which checkpoint to resume processing from.
+    /// Given a `pipeline_task` representing either a pipeline name or a pipeline with an associated
+    /// task (formatted as `{pipeline}{Store::DELIMITER}{task}`), return the committer watermark
+    /// from the `Store`. The indexer fetches this value for each pipeline added to determine which
+    /// checkpoint to resume processing from.
     async fn committer_watermark(
         &mut self,
-        pipeline: &'static str,
+        pipeline_task: &str,
     ) -> anyhow::Result<Option<CommitterWatermark>>;
 
     /// Given a pipeline, return the reader watermark from the database. This is used by the indexer
@@ -36,11 +39,13 @@ pub trait Connection: Send {
         delay: Duration,
     ) -> anyhow::Result<Option<PrunerWatermark>>;
 
-    /// Upsert the high watermark as long as it raises the watermark stored in the database. Returns
-    /// a boolean indicating whether the watermark was actually updated or not.
+    /// Upsert the high watermark for the `pipeline_task` - representing either a pipeline name or a
+    /// pipeline with an associated task (formatted as `{pipeline}{Store::DELIMITER}{task}`) - as
+    /// long as it raises the watermark stored in the database. Returns a boolean indicating whether
+    /// the watermark was actually updated or not.
     async fn set_committer_watermark(
         &mut self,
-        pipeline: &'static str,
+        pipeline_task: &str,
         watermark: CommitterWatermark,
     ) -> anyhow::Result<bool>;
 
@@ -63,7 +68,7 @@ pub trait Connection: Send {
         reader_lo: u64,
     ) -> anyhow::Result<bool>;
 
-    /// Update the pruner watermark, returns true if the watermark was actually updated
+    /// Update the pruner watermark, returns true if the watermark was actually updated.
     async fn set_pruner_watermark(
         &mut self,
         pipeline: &'static str,
@@ -81,6 +86,10 @@ pub trait Store: Send + Sync + 'static + Clone {
     where
         Self: 'c;
 
+    /// Delimiter used to separate pipeline names from task identifiers when reading or writing the
+    /// committer watermark.
+    const DELIMITER: &'static str = "@";
+
     async fn connect<'c>(&'c self) -> Result<Self::Connection<'c>, anyhow::Error>;
 }
 
@@ -179,6 +188,27 @@ impl PrunerWatermark {
     }
 }
 
+/// Check that the pipeline name does not contain the store's delimiter, and construct the string
+/// used for tracking a pipeline's watermarks in the store. This is either the pipeline name itself,
+/// or `{pipeline}{Store::DELIMITER}{task}` if a task name is provided.
+pub fn pipeline_task<S: Store>(
+    pipeline_name: &'static str,
+    task_name: Option<&str>,
+) -> Result<String> {
+    if pipeline_name.contains(S::DELIMITER) {
+        anyhow::bail!(
+            "Pipeline name '{}' contains invalid delimiter '{}'",
+            pipeline_name,
+            S::DELIMITER
+        );
+    }
+
+    Ok(match task_name {
+        Some(task_name) => format!("{}{}{}", pipeline_name, S::DELIMITER, task_name),
+        None => pipeline_name.to_string(),
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/sui-indexer-alt-framework/src/ingestion/broadcaster.rs

@@ -217,7 +217,7 @@ async fn ingest_and_broadcast_range(
                 // we treat an error returned here as cancellation too.
                 if tokio::select! {
                     result = ingest_hi_rx.wait_for(|hi| hi.is_none_or(|h| cp < h)) => result.is_err(),
-                    _ = cancel.cancelled() => true,
+                    _ = cancel.cancelled() => true
                 } {
                     return Err(Error::Cancelled);
                 }
@@ -238,8 +238,8 @@ async fn ingest_and_broadcast_range(
                             debug!(checkpoint = cp, "Broadcasted checkpoint");
                             Ok(())
                         } else {
-                            // An error is returned meaning some subscriber channel has closed, which we consider
-                            // a cancellation signal for the entire ingestion.
+                            // An error is returned meaning some subscriber channel has closed,
+                            // which we consider a cancellation signal for the entire ingestion.
                             cancel.cancel();
                             Err(Error::Cancelled)
                         }