Repository: MystenLabs/sui
File: crates/sui-indexer-alt-framework/src/lib.rs (Rust)

Code before:
        &mut self,
        init_watermark: bool,
    ) -> Result<Option<u64>> {
        ensure!(
            self.added_pipelines.insert(P::NAME),
            "Pipeline {:?} already added",
            P::NAME,
        );

        if let Some(enabled_pipelines) = &mut self.enabled_pipelines
            && !enabled_pipelines.remove(P::NAME)
        {
            info!(pipeline = P::NAME, "Skipping");
            return Ok(None);
        }

        let mut conn = self
            .store
            .connect()
            .await
            .context("Failed to establish connection to store")?;

        let pipeline_task =
            pipeline_task::<S>(P::NAME, self.task.as_ref().map(|t| t.task.as_str()))?;

        let watermark = if init_watermark {
            let init_watermark = InitWatermark {
                checkpoint_hi_inclusive: self.default_next_checkpoint as i64 - 1,
                reader_lo: self.default_next_checkpoint,
                pruner_hi: self.default_next_checkpoint,
            };
            conn.init_watermark(&pipeline_task, init_watermark)
                .await
                .with_context(|| format!("Failed to init watermark for {pipeline_task}"))?
        } else {
            conn.committer_watermark(&pipeline_task)
                .await
                .with_context(|| format!("Failed to get watermark for {pipeline_task}"))?
        };

        let next_checkpoint = watermark
            .as_ref()
            .map(|w| w.checkpoint_hi_inclusive + 1)
            .unwrap_or(self.default_next_checkpoint);

        self.first_ingestion_checkpoint = next_checkpoint.min(self.first_ingestion_checkpoint);

        Ok(Some(next_checkpoint))
    }
}


Review comment (suggestion):
> i think it would be nice if we could avoid store implementors from having to deal with `checkpoint_hi_inclusive < 0`, which barring negative numbers, means someone configured `--first-checkpoint 0`. In that case, it should be the same as not providing `--first-checkpoint` at all. We can then leverage the existing logic

- If `default_next_checkpoint` is 0 
  - we don't have to do anything in `lib`
- If `default_next_checkpoint` is `>0`, then seed the watermark if no entry exists

So something like
```
if self.task.is_some() || default_next_checkpoint = 0, conn.committer_watermark()
else conn.init_watermark()
```



Code after:

        let pipeline_task =
            pipeline_task::<S>(P::NAME, self.task.as_ref().map(|t| t.task.as_str()))?;

        let checkpoint_hi_inclusive = conn
            .init_watermark(&pipeline_task, self.default_next_checkpoint)
            .await
            .with_context(|| format!("Failed to init watermark for {pipeline_task}"))?;

        let next_checkpoint =
            checkpoint_hi_inclusive.map_or(self.default_next_checkpoint, |c| c + 1);

        self.first_ingestion_checkpoint = next_checkpoint.min(self.first_ingestion_checkpoint);

        Ok(Some(next_checkpoint))
    }
}

impl<T: TransactionalStore> Indexer<T> {
    /// Adds a new pipeline to this indexer and starts it up. Although their tasks have started,
    /// they will be idle until the ingestion service starts, and serves it checkpoint data.
    ///
    /// Sequential pipelines commit checkpoint data in-order which sacrifices throughput, but may be
    /// required to handle pipelines that modify data in-place (where each update is not an insert,
    /// but could be a modification of an existing row, where ordering between updates is
    /// important).
    ///
    /// The pipeline can optionally be configured to lag behind the ingestion service by a fixed
    /// number of checkpoints (configured by `checkpoint_lag`).
    pub async fn sequential_pipeline<H>(
        &mut self,
        handler: H,
        config: SequentialConfig,
    ) -> Result<()>
    where
        H: Handler<Store = T> + Send + Sync + 'static,
    {
        let Some(next_checkpoint) = self.add_pipeline::<H>().await? else {
            return Ok(());
        };

        if self.task.is_some() {
            bail!(
                "Sequential pipelines do not support pipeline tasks. \
                These pipelines guarantee that each checkpoint is committed exactly once and in order. \
                Running the same pipeline under a different task would violate these guarantees."
            );
        }

        // Track the minimum next_checkpoint across all sequential pipelines
        self.next_sequential_checkpoint = Some(