Ingest batches of ledgers in-memory before flushing to DB

[tamirms]

In #4909 we have updated all the transaction processors to use the FastBatchInsertBuilder to insert rows into the history tables with the postgres COPY command. We also refactored the transaction processor interface to allow a single processor to accumulate records across multiple ledgers:

type horizonTransactionProcessor interface {
	ProcessTransaction(xdr.LedgerCloseMeta, ingest.LedgerTransaction) error
	Flush(ctx context.Context, session db.SessionInterface) error
}

Now we can build on this work to improve the performance of reingestion. #4909 improved the performance of ingesting a single ledger. But we can extract more performance gains by ingesting multiple ledgers within a single processor lifetime.

In the spike branch the dataflow for ingesting batches of ledgers within a single processor lifetime is:

	accountLoader := history.NewAccountLoader()
	cbLoader := history.NewClaimableBalanceLoader()
	lpLoader := history.NewLiquidityPoolLoader()
	assetLoader := history.NewAssetLoader()
	processors := buildTransactionProcessors(
		s.historyQ,
		accountLoader,
		cbLoader,
		lpLoader,
		assetLoader,
	)

       // apply all the ledgers in the batch on the processors
       for _, ledger := range ledgers {
		if err = s.runner.ApplyProcessorsOnLedger(processors, ledgerCloseMeta); err != nil {
			return err
		}
       }

       // use the loaders to lookup all the accounts, assets, claimable balances, and liquidity pools registered
       // by the processors
       err = func() error {
		if err := s.historyQ.Begin(); err != nil {
			return errors.Wrap(err, "Error starting a transaction")
		}
		defer s.historyQ.Rollback()

		if err := accountLoader.Exec(s.ctx, s.historyQ); err != nil {
			return err
		}
		if err := cbLoader.Exec(s.ctx, s.historyQ); err != nil {
			return err
		}
		if err := lpLoader.Exec(s.ctx, s.historyQ); err != nil {
			return err
		}
		if err := assetLoader.Exec(s.ctx, s.historyQ); err != nil {
			return err
		}
		if err := s.historyQ.Commit(); err != nil {
			return errors.Wrap(err, commitErrMsg)
		}
		return nil
	}()

        // flush the rows to the db, the processors will be able to obtain the integer ids from the loaders
	if err := s.historyQ.Begin(); err != nil {
		return errors.Wrap(err, "Error starting a transaction")
	}
	defer s.historyQ.Rollback()
	if err := processors.Commit(s.ctx, s.historyQ); err != nil {
		return err
	}
	if err := s.historyQ.Commit(); err != nil {
		return errors.Wrap(err, commitErrMsg)
	}

We will need to implement this new dataflow on the following states in the ingestion state machine:

• reingestHistoryRangeState (we should make sure to test both single threaded reingestion and parallel reingestion)
• historyRangeState

Note the resume state only ingests a single ledger so it is already covered by #4909 .

Also, the following bugs will need to be addressed when implementing this issue:

• RebuildTradeAggregationBuckets() cannot be invoked concurrently during parallel ingestion because there will be duplicate key constraint errors when two workers invoke the function on adjacent buckets. That is because the buckets occur on minute boundaries and two adjacent ledger ranges will share the same trade aggregations bucket. We can fix this by modifying parallel reingestion so that the trade aggregation buckets are built once all the workers have completed their ingestion jobs. - services/horizon/ingest: RebuildTradeAggregationBuckets cannot be invoked concurrently during parallel ingestion #5127

• We should remove the force flag because it is incompatible with the new data-flow of ingestion where we batch multiple ledgers in a single transaction. - services/horizon/ingest: remove the force flag on reingestion cmds #5128

[sreuland]

reduced scope slightly, removed - verifyRangeState from list of states that need ranged ledger enablement, due to that state invoked change processors also, which we want to retain those processors to having a single ledger scope, we want to limit ledger ranged scope to only the tx processors.