Trait Handler

pub trait Handler: Processor<Value: FieldCount> {
    type Store: Store;

    const MIN_EAGER_ROWS: usize = 50usize;
    const MAX_PENDING_ROWS: usize = 5_000usize;
    const PRUNING_REQUIRES_PROCESSED_VALUES: bool = false;

    // Required method
    fn commit<'a, 'life0, 'life1, 'async_trait>(
        values: &'life0 [Self::Value],
        conn: &'life1 mut <Self::Store as Store>::Connection<'a>,
    ) -> Pin<Box<dyn Future<Output = Result<usize>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'a: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;

    // Provided method
    fn prune<'a, 'life0, 'life1, 'async_trait>(
        &'life0 self,
        _from: u64,
        _to_exclusive: u64,
        _conn: &'life1 mut <Self::Store as Store>::Connection<'a>,
    ) -> Pin<Box<dyn Future<Output = Result<usize>> + Send + 'async_trait>>
       where Self: Sync + 'async_trait,
             'a: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
}

Handlers implement the logic for a given indexing pipeline: How to process checkpoint data (by implementing Processor) into rows for their table, and how to write those rows to the database.

The handler is also responsible for tuning the various parameters of the pipeline (provided as associated values). Reasonable defaults have been chosen to balance concurrency with memory usage, but each handle may choose to override these defaults, e.g.

• Handlers that produce many small rows may wish to increase their batch/chunk/max-pending sizes).
• Handlers that do more work during processing may wish to increase their fanout so more of it can be done concurrently, to preserve throughput.

Concurrent handlers can only be used in concurrent pipelines, where checkpoint data is processed and committed out-of-order and a watermark table is kept up-to-date with the latest checkpoint below which all data has been committed.

Back-pressure is handled through the MAX_PENDING_SIZE constant – if more than this many rows build up, the collector will stop accepting new checkpoints, which will eventually propagate back to the ingestion service.

Provided Associated Constants

const MIN_EAGER_ROWS: usize = 50usize

If at least this many rows are pending, the committer will commit them eagerly.

const MAX_PENDING_ROWS: usize = 5_000usize

If there are more than this many rows pending, the committer applies backpressure.

const PRUNING_REQUIRES_PROCESSED_VALUES: bool = false

Whether the pruner requires processed values in order to prune. This will determine the first checkpoint to process when we start the pipeline. If this is true, when the pipeline starts, it will process all checkpoints from the pruner watermark, so that the pruner have access to the processed values for any unpruned checkpoints. If this is false, when the pipeline starts, it will process all checkpoints from the committer watermark.

Required Associated Types

type Store: Store

Required Methods

fn commit<'a, 'life0, 'life1, 'async_trait>( values: &'life0 [Self::Value], conn: &'life1 mut <Self::Store as Store>::Connection<'a>, ) -> Pin<Box<dyn Future<Output = Result<usize>> + Send + 'async_trait>>

where Self: 'async_trait, 'a: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Take a chunk of values and commit them to the database, returning the number of rows affected.

Provided Methods

fn prune<'a, 'life0, 'life1, 'async_trait>( &'life0 self, _from: u64, _to_exclusive: u64, _conn: &'life1 mut <Self::Store as Store>::Connection<'a>, ) -> Pin<Box<dyn Future<Output = Result<usize>> + Send + 'async_trait>>

where Self: Sync + 'async_trait, 'a: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Clean up data between checkpoints _from and _to_exclusive (exclusive) in the database, returning the number of rows affected. This function is optional, and defaults to not pruning at all.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors