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

Code before:
            .await
            .context("Failed to start ingestion service")?;

        self.handles.push(broadcaster_handle);

        Ok(tokio::spawn(async move {
            // Wait for the ingestion service and all its related tasks to wind down gracefully:
            // If ingestion has been configured to only handle a specific range of checkpoints, we
            // want to make sure that tasks are allowed to run to completion before shutting them
            // down.
            future::join_all(self.handles).await;
            info!("Indexing pipeline gracefully shut down");
        }))
    }

    /// Determine the checkpoint for the pipeline to resume processing from. This is either the
    /// checkpoint after its watermark, or if that doesn't exist, then the provided
    /// [Self::first_checkpoint], and if that is not set, then 0 (genesis).
    ///
    /// Update the starting ingestion checkpoint as the minimum across all the next checkpoints
    /// calculated above.
    ///
    /// Returns `Ok(None)` if the pipeline is disabled.
    async fn add_pipeline<P: Processor + 'static>(
        &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 {

Review comment (suggestion):
I suppose alternatively, we don't need to add this function parameter, and just check `self.task.is_none()` in `add_pipeline` right? If it's a tasked indexer, we don't need to seed un-watermarked pipelines, because they're tasked pipelines and don't have the ability to prune

Code after:
    ///
    /// Update the starting ingestion checkpoint as the minimum across all the next checkpoints
    /// calculated above.
    ///
    /// Returns `Ok(None)` if the pipeline is disabled.
    async fn add_pipeline<P: Processor + 'static>(&mut self) -> 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 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).
    ///