Struct Indexer

pub struct Indexer { /* private fields */ }

Implementations

impl Indexer

pub async fn new( database_url: Url, db_args: DbArgs, indexer_args: IndexerArgs, client_args: ClientArgs, ingestion_config: IngestionConfig, migrations: Option<&'static EmbeddedMigrations>, registry: &Registry, cancel: CancellationToken, ) -> Result<Self>

Create a new instance of the indexer framework. database_url, db_args, indexer_args,, client_args, and ingestion_config contain configurations for the following, respectively:

• Connecting to the database,
• What is indexed (which checkpoints, which pipelines, whether to update the watermarks table) and where to serve metrics from,
• Where to download checkpoints from,
• Concurrency and buffering parameters for downloading checkpoints.

Optional migrations contains the SQL to run in order to bring the database schema up-to-date for the specific instance of the indexer, generated using diesel’s embed_migrations! macro. These migrations will be run as part of initializing the indexer if provided.

After initialization, at least one pipeline must be added using Self::concurrent_pipeline or Self::sequential_pipeline, before the indexer is started using Self::run.

pub async fn new_for_testing( migrations: &'static EmbeddedMigrations, ) -> (Self, TempDb)

pub fn db(&self) -> &Db

The database connection pool used by the indexer.

pub fn ingestion_client(&self) -> &IngestionClient

The ingestion client used by the indexer to fetch checkpoints.

pub fn metrics(&self) -> &Arc<IndexerMetrics>

The indexer’s metrics.

pub fn pipelines(&self) -> impl Iterator<Item = &'static str> + '_

The pipelines that this indexer will run.

pub async fn concurrent_pipeline<H>( &mut self, handler: H, config: ConcurrentConfig, ) -> Result<()>

where H: Handler<Store = Db> + Send + Sync + 'static,

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.

Concurrent pipelines commit checkpoint data out-of-order to maximise throughput, and they keep the watermark table up-to-date with the highest point they can guarantee all data exists for, for their pipeline.

pub async fn sequential_pipeline<H>( &mut self, handler: H, config: SequentialConfig, ) -> Result<()>

where H: Handler<Store = Db> + Send + Sync + 'static,

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 run(self) -> Result<JoinHandle<()>>

Start ingesting checkpoints. Ingestion either starts from the configured first_checkpoint, or it is calculated based on the watermarks of all active pipelines. Ingestion will stop after consuming the configured last_checkpoint, if one is provided, or will continue until it tracks the tip of the network.

pub fn migrations( migrations: Option<&'static EmbeddedMigrations>, ) -> impl MigrationSource<Pg> + Send + Sync + 'static

Combine the provided migrations with the migrations necessary to set up the indexer framework. The returned migration source can be passed to Db::run_migrations to ensure the database’s schema is up-to-date for both the indexer framework and the specific indexer.