Create BackfillProcessorStatus Table and Write to it when backfill_alias is set

[dermanyang]

dependent on #8
cherry picked from PR in old repo structure

Purpose

Repoint backfill processors to write to the new backfill_processor_status table instead of the processor_status table so that backfill processors can run concurrently as head processors without having their checkpoints overwritten. Read more in design here.

Technical Overview

BackfillConfig

BackfillConfig.backfill_alias is introduced to IndexerProcessorConfig. This optional field introduced a new logic branch in LatestVersionTracker (saved as field backfill_mode) which writes to the new table when backfill_alias is Some (equivalently when backfill_mode is true).

The backfill_alias field is similarly used in the get_starting_version logic to determine where to read from.

ProcessorVersionSaverEnum

Created a ProcessorStatusSaverEnum for the DefaultProcessorStatusSaver and BackfillProcessorStatusSaver. This makes it easier to init the VersionTracker without exposing too much logic. See usage:

let version_tracker = VersionTrackerStep::new(
            get_processor_status_saver(self.db_pool.clone(), self.config.clone()),
            DEFAULT_UPDATE_PROCESSOR_STATUS_SECS,
        );

Where

pub fn get_processor_status_saver(
    conn_pool: ArcDbPool,
    config: IndexerProcessorConfig,
) -> ProcessorStatusSaverEnum {
    if let Some(backfill_config) = config.backfill_config {
        let txn_stream_cfg = config.transaction_stream_config;
        let backfill_start_version = txn_stream_cfg.starting_version;
        let backfill_end_version = txn_stream_cfg.request_ending_version;
        let backfill_alias = backfill_config.backfill_alias.clone();
        ProcessorStatusSaverEnum::Backfill {
            conn_pool,
            backfill_alias,
            backfill_start_version,
            backfill_end_version,
        }
    } else {
        let processor_name = config.processor_config.name().to_string();
        ProcessorStatusSaverEnum::Default {
            conn_pool,
            processor_name,
        }
    }
}

Testing

Ran a few manual tests validating existing and new behavior. Screenshots below:

Regular processor working as expected. No backfillconfig set.

Backfill processor starting from genesis and writing to new table. Backfillconfig.backfill_alias set to events_processor_backfill_1:

Backfill processor picking up where it left off (see bottom):

Backfill processor starting from starting_version

[rtso]

nit: we call this processor_name in other places, so let's rename it here too

[rtso]

Since we're already here, we should probably move the +1 to a higher level, this function seems like the wrong place for it.

[dermanyang]

good call. it disagrees with the name of the function. We now have:

get_starting_version(): 
  return get_latest_processed_version_from_db() + 1