Struct IndexerExecutor 

pub struct IndexerExecutor<P> { /* private fields */ }

The Executor of the main ingestion pipeline process.

This struct orchestrates the execution of multiple worker pools, handling checkpoint distribution, progress tracking, and shutdown. It utilizes ProgressStore for persisting checkpoint progress and provides metrics for monitoring the indexing process.

Example

use async_trait::async_trait;
use iota_data_ingestion_core::{
    DataIngestionMetrics, FileProgressStore, IndexerExecutor, IngestionError, ReaderOptions,
    Worker, WorkerPool,
};
use iota_types::full_checkpoint_content::CheckpointData;
use prometheus::Registry;
use tokio_util::sync::CancellationToken;
use std::{path::PathBuf, sync::Arc};

struct CustomWorker;

#[async_trait]
impl Worker for CustomWorker {
    type Message = ();
    type Error = IngestionError;

    async fn process_checkpoint(
        &self,
        checkpoint: Arc<CheckpointData>,
    ) -> Result<Self::Message, Self::Error> {
        // custom processing logic.
        println!(
            "Processing Local checkpoint: {}",
            checkpoint.checkpoint_summary.to_string()
        );
        Ok(())
    }
}

#[tokio::main]
async fn main() {
    let concurrency = 5;
    let progress_store = FileProgressStore::new("progress.json").await.unwrap();
    let mut executor = IndexerExecutor::new(
        progress_store,
        1, // number of registered WorkerPools.
        DataIngestionMetrics::new(&Registry::new()),
        CancellationToken::new(),
    );
    // register a worker pool with 5 workers to process checkpoints in parallel
    let worker_pool = WorkerPool::new(CustomWorker, "local_reader".to_string(), concurrency, Default::default());
    // register the worker pool to the executor.
    executor.register(worker_pool).await.unwrap();
    // run the ingestion pipeline.
    executor
        .run(
            PathBuf::from("./chk".to_string()), // path to a local directory where checkpoints are stored.
            None,
            vec![],                   // optional remote store access options.
            ReaderOptions::default(), // remote_read_batch_size.
        )
        .await
        .unwrap();
}

Implementations

impl<P: ProgressStore> IndexerExecutor<P>

pub fn new( progress_store: P, number_of_jobs: usize, metrics: DataIngestionMetrics, token: CancellationToken, ) -> Self

pub async fn register<W: Worker + 'static>( &mut self, pool: WorkerPool<W>, ) -> IngestionResult<()>

Registers new worker pool in executor.

pub fn shutdown_when<F>(&mut self, f: F)

where F: Fn(&CheckpointData) -> ShutdownAction + Send + 'static,

Registers a predicate callback that determines when the ingestion process should stop.

This function f will be called for every incoming checkpoint before it’s sent to the worker pool.

Based on the returned ShutdownAction the executor will evaluate whether to continue or stop the ingestion process by initiating the graceful shutdown process.

Once a shutdown action is triggered, the executor will stop sending new checkpoints and will wait for all previously sent checkpoints to be processed by workers before initiating graceful shutdown process.

Note:

Calling this method after with_ingestion_limit replaces the earlier predicate, and vice versa. They are not cumulative.

pub fn with_ingestion_limit(&mut self, limit: IngestionLimit)

Adds an upper‑limit policy that determines when the ingestion process should stop.

This is a convenience method, it internally uses shutdown_when by registering a predicate derived from the provided IngestionLimit.

Note:

Calling this method after shutdown_when replaces the earlier predicate, and vice versa. They are not cumulative.

pub async fn update_watermark( &mut self, task_name: String, watermark: CheckpointSequenceNumber, ) -> IngestionResult<()>

pub async fn read_watermark( &mut self, task_name: String, ) -> IngestionResult<CheckpointSequenceNumber>

pub async fn run( self, path: PathBuf, remote_store_url: Option<String>, remote_store_options: Vec<(String, String)>, reader_options: ReaderOptions, ) -> IngestionResult<HashMap<String, CheckpointSequenceNumber>>

Main executor loop.

Error

Returns an IngestionError::EmptyWorkerPool if no worker pool was registered.

pub async fn run_with_config( self, config: CheckpointReaderConfig, ) -> IngestionResult<HashMap<String, CheckpointSequenceNumber>>

Alternative main executor loop. Uses the new iteration of the CheckpointReader supporting syncing checkppints from hybrid historical store.

Error

Returns an IngestionError::EmptyWorkerPool if no worker pool was registered.