Repository: stellar/go
Pull request #4636 (2022-10-05 11:20:15)

services/horizon: Add `ingest build-state` command
<!-- If you're making a doc PR or something tiny where the below is irrelevant, delete this
template and use a short description, but in your description aim to include both what the
change is, and why it is being made, with enough context for anyone to understand. -->

<details>
  <summary>PR Checklist</summary>
  
### PR Structure

* [ ] This PR has reasonably narrow scope (if not, break it down into smaller PRs).
* [ ] This PR avoids mixing refactoring changes with feature changes (split into two PRs
  otherwise).
* [ ] This PR's title starts with name of package that is most changed in the PR, ex.
  `services/friendbot`, or `all` or `doc` if the changes are broad or impact many
  packages.

### Thoroughness

* [ ] This PR adds tests for the most critical parts of the new functionality or fixes.
* [ ] I've updated any docs ([developer docs](https://developers.stellar.org/api/), `.md`
  files, etc... affected by this change). Take a look in the `docs` folder for a given service,
  like [this one](https://github.com/stellar/go/tree/master/services/horizon/internal/docs).

### Release planning

* [ ] I've updated the relevant CHANGELOG ([here](services/horizon/CHANGELOG.md) for Horizon) if
  needed with deprecations, added features, breaking changes, and DB schema changes.
* [ ] I've decided if this PR requires a new major/minor version according to
  [semver](https://semver.org/), or if it's mainly a patch change. The PR is targeted at the next
  release branch if it's not a patch change.
</details>

### What

Add `ingest build-state` command which builds state at any given sequence number and exists. It's also possible to skip bucket list hash checks which significantly speeds up the process (useful for debugging purposes).

### Why

It's ofter useful to debug state at any ledger and do it fast without bucket list hash checks.

### Known limitations

[TODO or N/A]



Patch:
diff --git a/services/horizon/cmd/ingest.go b/services/horizon/cmd/ingest.go
index fa7d7121..2cae1fea 100644
--- a/services/horizon/cmd/ingest.go
+++ b/services/horizon/cmd/ingest.go
@@ -23,9 +23,30 @@ var ingestCmd = &cobra.Command{
 	Short: "ingestion related commands",
 }
 
+var ingestBuildStateSequence uint32
+var ingestBuildStateSkipChecks bool
 var ingestVerifyFrom, ingestVerifyTo, ingestVerifyDebugServerPort uint32
 var ingestVerifyState bool
 
+var ingestBuildStateCmdOpts = []*support.ConfigOption{
+	{
+		Name:        "sequence",
+		ConfigKey:   &ingestBuildStateSequence,
+		OptType:     types.Uint32,
+		Required:    true,
+		FlagDefault: uint32(0),
+		Usage:       "checkpoint ledger sequence",
+	},
+	{
+		Name:        "skip-checks",
+		ConfigKey:   &ingestBuildStateSkipChecks,
+		OptType:     types.Bool,
+		Required:    false,
+		FlagDefault: false,
+		Usage:       "[optional] set to skip protocol version and bucket list hash verification, can speed up the process because does not require a running Stellar-Core",
+	},
+}
+
 var ingestVerifyRangeCmdOpts = []*support.ConfigOption{
 	{
 		Name:        "from",
@@ -330,6 +351,90 @@ var ingestInitGenesisStateCmd = &cobra.Command{
 	},
 }
 
+var ingestBuildStateCmd = &cobra.Command{
+	Use:   "build-state",
+	Short: "builds state at a given checkpoint. warning! requires clean DB.",
+	Long:  "useful for debugging or starting Horizon at specific checkpoint.",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		for _, co := range ingestBuildStateCmdOpts {
+			if err := co.RequireE(); err != nil {
+				return err
+			}
+			co.SetValue()
+		}
+
+		if err := horizon.ApplyFlags(config, flags, horizon.ApplyOptions{RequireCaptiveCoreConfig: false, AlwaysIngest: true}); err != nil {
+			return err
+		}
+
+		horizonSession, err := db.Open("postgres", config.DatabaseURL)
+		if err != nil {
+			return fmt.Errorf("cannot open Horizon DB: %v", err)
+		}
+
+		historyQ := &history.Q{horizonSession}
+
+		lastIngestedLedger, err := historyQ.GetLastLedgerIngestNonBlocking(context.Background())
+		if err != nil {
+			return fmt.Errorf("cannot get last ledger value: %v", err)
+		}
+
+		if lastIngestedLedger != 0 {
+			return fmt.Errorf("cannot run on non-empty DB")
+		}
+
+		mngr := historyarchive.NewCheckpointManager(config.CheckpointFrequency)
+		if !mngr.IsCheckpoint(ingestBuildStateSequence) && ingestBuildStateSequence != 1 {
+			return fmt.Errorf("`--sequence` must be a checkpoint ledger")
+		}
+
+		ingestConfig := ingest.Config{
+			NetworkPassphrase:        config.NetworkPassphrase,
+			HistorySession:           horizonSession,
+			HistoryArchiveURL:        config.HistoryArchiveURLs[0],
+			EnableCaptiveCore:        config.EnableCaptiveCoreIngestion,
+			CaptiveCoreBinaryPath:    config.CaptiveCoreBinaryPath,
+			CaptiveCoreConfigUseDB:   config.CaptiveCoreConfigUseDB,
+			RemoteCaptiveCoreURL:     config.RemoteCaptiveCoreURL,
+			CheckpointFrequency:      config.CheckpointFrequency,
+			CaptiveCoreToml:          config.CaptiveCoreToml,
+			CaptiveCoreStoragePath:   config.CaptiveCoreStoragePath,
+			RoundingSlippageFilter:   config.RoundingSlippageFilter,
+			EnableIngestionFiltering: config.EnableIngestionFiltering,
+		}
+
+		if !ingestBuildStateSkipChecks {
+			if !ingestConfig.EnableCaptiveCore {
+				if config.StellarCoreDatabaseURL == "" {
+					return fmt.Errorf("flag --%s cannot be empty", horizon.StellarCoreDBURLFlagName)
+				}
+
+				coreSession, dbErr := db.Open("postgres", config.StellarCoreDatabaseURL)
+				if dbErr != nil {
+					return fmt.Errorf("cannot open Core DB: %v", dbErr)
+				}
+				ingestConfig.CoreSession = coreSession
+			}
+		}
+
+		system, err := ingest.NewSystem(ingestConfig)
+		if err != nil {
+			return err
+		}
+
+		err = system.BuildState(
+			ingestBuildStateSequence,
+			ingestBuildStateSkipChecks,
+		)
+		if err != nil {
+			return err
+		}
+
+		log.Info("State built successfully!")
+		return nil
+	},
+}
+
 func init() {
 	for _, co := range ingestVerifyRangeCmdOpts {
 		err := co.Init(ingestVerifyRangeCmd)
@@ -345,6 +450,13 @@ func init() {
 		}
 	}
 
+	for _, co := range ingestBuildStateCmdOpts {
+		err := co.Init(ingestBuildStateCmd)
+		if err != nil {
+			log.Fatal(err.Error())
+		}
+	}
+
 	viper.BindPFlags(ingestVerifyRangeCmd.PersistentFlags())
 
 	RootCmd.AddCommand(ingestCmd)
@@ -353,5 +465,6 @@ func init() {
 		ingestStressTestCmd,
 		ingestTriggerStateRebuildCmd,
 		ingestInitGenesisStateCmd,
+		ingestBuildStateCmd,
 	)
 }
diff --git a/services/horizon/internal/ingest/fsm.go b/services/horizon/internal/ingest/fsm.go
index 831642b5..2f6ef199 100644
--- a/services/horizon/internal/ingest/fsm.go
+++ b/services/horizon/internal/ingest/fsm.go
@@ -226,11 +226,12 @@ func (state startState) run(s *system) (transition, error) {
 
 type buildState struct {
 	checkpointLedger uint32
+	skipChecks       bool
 	stop             bool
 }
 
 func (b buildState) String() string {
-	return fmt.Sprintf("buildFromCheckpoint(checkpointLedger=%d)", b.checkpointLedger)
+	return fmt.Sprintf("buildFromCheckpoint(checkpointLedger=%d, skipChecks=%t)", b.checkpointLedger, b.skipChecks)
 }
 
 func (b buildState) run(s *system) (transition, error) {
@@ -248,8 +249,11 @@ func (b buildState) run(s *system) (transition, error) {
 	// In the long term we should probably create artificial xdr.LedgerCloseMeta
 	// for ledger #1 instead of using `ingest.GenesisChange` reader in
 	// ProcessorRunner.RunHistoryArchiveIngestion().
-	var ledgerCloseMeta xdr.LedgerCloseMeta
-	if b.checkpointLedger != 1 {
+	// We can also skip preparing range if `skipChecks` is `true` because we
+	// won't need bucket list hash and protocol version.
+	var protocolVersion uint32
+	var bucketListHash xdr.Hash
+	if b.checkpointLedger != 1 && !b.skipChecks {
 		err := s.maybePrepareRange(s.ctx, b.checkpointLedger)
 		if err != nil {
 			return nextFailState, err
@@ -257,7 +261,7 @@ func (b buildState) run(s *system) (transition, error) {
 
 		log.WithField("sequence", b.checkpointLedger).Info("Waiting for ledger to be available in the backend...")
 		startTime := time.Now()
-		ledgerCloseMeta, err = s.ledgerBackend.GetLedger(s.ctx, b.checkpointLedger)
+		ledgerCloseMeta, err := s.ledgerBackend.GetLedger(s.ctx, b.checkpointLedger)
 		if err != nil {
 			return nextFailState, errors.Wrap(err, "error getting ledger blocking")
 		}
@@ -265,6 +269,9 @@ func (b buildState) run(s *system) (transition, error) {
 			"sequence": b.checkpointLedger,
 			"duration": time.Since(startTime).Seconds(),
 		}).Info("Ledger returned from the backend")
+
+		protocolVersion = ledgerCloseMeta.ProtocolVersion()
+		bucketListHash = ledgerCloseMeta.BucketListHash()
 	}
 
 	if err := s.historyQ.Begin(); err != nil {
@@ -328,9 +335,10 @@ func (b buildState) run(s *system) (transition, error) {
 		stats, err = s.runner.RunGenesisStateIngestion()
 	} else {
 		stats, err = s.runner.RunHistoryArchiveIngestion(
-			ledgerCloseMeta.LedgerSequence(),
-			ledgerCloseMeta.ProtocolVersion(),
-			ledgerCloseMeta.BucketListHash(),
+			b.checkpointLedger,
+			b.skipChecks,
+			protocolVersion,
+			bucketListHash,
 		)
 	}
 
@@ -895,6 +903,7 @@ func (v verifyRangeState) run(s *system) (transition, error) {
 
 	stats, err := s.runner.RunHistoryArchiveIngestion(
 		ledgerCloseMeta.LedgerSequence(),
+		false,
 		ledgerCloseMeta.ProtocolVersion(),
 		ledgerCloseMeta.BucketListHash(),
 	)
diff --git a/services/horizon/internal/ingest/main.go b/services/horizon/internal/ingest/main.go
index cd033fc0..4f1de4a2 100644
--- a/services/horizon/internal/ingest/main.go
+++ b/services/horizon/internal/ingest/main.go
@@ -173,6 +173,7 @@ type System interface {
 	Metrics() Metrics
 	StressTest(numTransactions, changesPerTransaction int) error
 	VerifyRange(fromLedger, toLedger uint32, verifyState bool) error
+	BuildState(sequence uint32, skipChecks bool) error
 	ReingestRange(ledgerRanges []history.LedgerRange, force bool) error
 	BuildGenesisState() error
 	Shutdown()
@@ -529,6 +530,16 @@ func (s *system) VerifyRange(fromLedger, toLedger uint32, verifyState bool) erro
 	})
 }
 
+// BuildState runs the state ingestion on selected checkpoint ledger then exits.
+// When skipChecks is true it skips bucket list hash verification and protocol version check.
+func (s *system) BuildState(sequence uint32, skipChecks bool) error {
+	return s.runStateMachine(buildState{
+		checkpointLedger: sequence,
+		skipChecks:       skipChecks,
+		stop:             true,
+	})
+}
+
 func validateRanges(ledgerRanges []history.LedgerRange) error {
 	for i, cur := range ledgerRanges {
 		if cur.StartSequence > cur.EndSequence {
diff --git a/services/horizon/internal/ingest/processor_runner.go b/services/horizon/internal/ingest/processor_runner.go
index f9ddce53..29635dce 100644
--- a/services/horizon/internal/ingest/processor_runner.go
+++ b/services/horizon/internal/ingest/processor_runner.go
@@ -65,6 +65,7 @@ type ProcessorRunnerInterface interface {
 	RunGenesisStateIngestion() (ingest.StatsChangeProcessorResults, error)
 	RunHistoryArchiveIngestion(
 		checkpointLedger uint32,
+		skipChecks bool,
 		ledgerProtocolVersion uint32,
 		bucketListHash xdr.Hash,
 	) (ingest.StatsChangeProcessorResults, error)
@@ -212,11 +213,12 @@ func (s *ProcessorRunner) validateBucketList(ledgerSequence uint32, ledgerBucket
 }
 
 func (s *ProcessorRunner) RunGenesisStateIngestion() (ingest.StatsChangeProcessorResults, error) {
-	return s.RunHistoryArchiveIngestion(1, 0, xdr.Hash{})
+	return s.RunHistoryArchiveIngestion(1, false, 0, xdr.Hash{})
 }
 
 func (s *ProcessorRunner) RunHistoryArchiveIngestion(
 	checkpointLedger uint32,
+	skipChecks bool,
 	ledgerProtocolVersion uint32,
 	bucketListHash xdr.Hash,
 ) (ingest.StatsChangeProcessorResults, error) {
@@ -228,12 +230,14 @@ func (s *ProcessorRunner) RunHistoryArchiveIngestion(
 			return changeStats.GetResults(), errors.Wrap(err, "Error ingesting genesis ledger")
 		}
 	} else {
-		if err := s.checkIfProtocolVersionSupported(ledgerProtocolVersion); err != nil {
-			return changeStats.GetResults(), errors.Wrap(err, "Error while checking for supported protocol version")
-		}
-
-		if err := s.validateBucketList(checkpointLedger, bucketListHash); err != nil {
-			return changeStats.GetResults(), errors.Wrap(err, "Error validating bucket list from HAS")
+		if !skipChecks {
+			if err := s.checkIfProtocolVersionSupported(ledgerProtocolVersion); err != nil {
+				return changeStats.GetResults(), errors.Wrap(err, "Error while checking for supported protocol version")
+			}
+
+			if err := s.validateBucketList(checkpointLedger, bucketListHash); err != nil {
+				return changeStats.GetResults(), errors.Wrap(err, "Error validating bucket list from HAS")
+			}
 		}
 
 		changeReader, err := s.historyAdapter.GetState(s.ctx, checkpointLedger)


Test patch:
diff --git a/services/horizon/internal/ingest/build_state_test.go b/services/horizon/internal/ingest/build_state_test.go
index f7779f6f..c4d8f478 100644
--- a/services/horizon/internal/ingest/build_state_test.go
+++ b/services/horizon/internal/ingest/build_state_test.go
@@ -232,7 +232,7 @@ func (s *BuildStateTestSuite) TestTruncateIngestStateTablesReturnsError() {
 func (s *BuildStateTestSuite) TestRunHistoryArchiveIngestionReturnsError() {
 	s.mockCommonHistoryQ()
 	s.runner.
-		On("RunHistoryArchiveIngestion", s.checkpointLedger, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
+		On("RunHistoryArchiveIngestion", s.checkpointLedger, false, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
 		Return(ingest.StatsChangeProcessorResults{}, errors.New("my error")).
 		Once()
 	next, err := buildState{checkpointLedger: s.checkpointLedger}.run(s.system)
@@ -272,7 +272,7 @@ func (s *BuildStateTestSuite) TestRunHistoryArchiveIngestionGenesisReturnsError(
 func (s *BuildStateTestSuite) TestUpdateLastLedgerIngestAfterIngestReturnsError() {
 	s.mockCommonHistoryQ()
 	s.runner.
-		On("RunHistoryArchiveIngestion", s.checkpointLedger, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
+		On("RunHistoryArchiveIngestion", s.checkpointLedger, false, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
 		Return(ingest.StatsChangeProcessorResults{}, nil).
 		Once()
 	s.historyQ.On("UpdateIngestVersion", s.ctx, CurrentVersion).
@@ -291,7 +291,7 @@ func (s *BuildStateTestSuite) TestUpdateLastLedgerIngestAfterIngestReturnsError(
 func (s *BuildStateTestSuite) TestUpdateIngestVersionIngestReturnsError() {
 	s.mockCommonHistoryQ()
 	s.runner.
-		On("RunHistoryArchiveIngestion", s.checkpointLedger, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
+		On("RunHistoryArchiveIngestion", s.checkpointLedger, false, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
 		Return(ingest.StatsChangeProcessorResults{}, nil).
 		Once()
 	s.historyQ.On("UpdateIngestVersion", s.ctx, CurrentVersion).
@@ -307,7 +307,7 @@ func (s *BuildStateTestSuite) TestUpdateIngestVersionIngestReturnsError() {
 func (s *BuildStateTestSuite) TestUpdateCommitReturnsError() {
 	s.mockCommonHistoryQ()
 	s.runner.
-		On("RunHistoryArchiveIngestion", s.checkpointLedger, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
+		On("RunHistoryArchiveIngestion", s.checkpointLedger, false, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
 		Return(ingest.StatsChangeProcessorResults{}, nil).
 		Once()
 	s.historyQ.On("UpdateLastLedgerIngest", s.ctx, s.checkpointLedger).
@@ -329,7 +329,7 @@ func (s *BuildStateTestSuite) TestUpdateCommitReturnsError() {
 func (s *BuildStateTestSuite) TestBuildStateSucceeds() {
 	s.mockCommonHistoryQ()
 	s.runner.
-		On("RunHistoryArchiveIngestion", s.checkpointLedger, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
+		On("RunHistoryArchiveIngestion", s.checkpointLedger, false, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
 		Return(ingest.StatsChangeProcessorResults{}, nil).
 		Once()
 	s.historyQ.On("UpdateLastLedgerIngest", s.ctx, s.checkpointLedger).
@@ -357,7 +357,7 @@ func (s *BuildStateTestSuite) TestBuildStateSucceeds() {
 func (s *BuildStateTestSuite) TestUpdateCommitReturnsErrorStop() {
 	s.mockCommonHistoryQ()
 	s.runner.
-		On("RunHistoryArchiveIngestion", s.checkpointLedger, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
+		On("RunHistoryArchiveIngestion", s.checkpointLedger, false, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
 		Return(ingest.StatsChangeProcessorResults{}, nil).
 		Once()
 	s.historyQ.On("UpdateLastLedgerIngest", s.ctx, s.checkpointLedger).
@@ -379,7 +379,7 @@ func (s *BuildStateTestSuite) TestUpdateCommitReturnsErrorStop() {
 func (s *BuildStateTestSuite) TestBuildStateSucceedStop() {
 	s.mockCommonHistoryQ()
 	s.runner.
-		On("RunHistoryArchiveIngestion", s.checkpointLedger, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
+		On("RunHistoryArchiveIngestion", s.checkpointLedger, false, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).
 		Return(ingest.StatsChangeProcessorResults{}, nil).
 		Once()
 	s.historyQ.On("UpdateLastLedgerIngest", s.ctx, s.checkpointLedger).
diff --git a/services/horizon/internal/ingest/main_test.go b/services/horizon/internal/ingest/main_test.go
index 7b930e80..0ee2a80b 100644
--- a/services/horizon/internal/ingest/main_test.go
+++ b/services/horizon/internal/ingest/main_test.go
@@ -458,10 +458,11 @@ func (m *mockProcessorsRunner) RunGenesisStateIngestion() (ingest.StatsChangePro
 
 func (m *mockProcessorsRunner) RunHistoryArchiveIngestion(
 	checkpointLedger uint32,
+	skipChecks bool,
 	ledgerProtocolVersion uint32,
 	bucketListHash xdr.Hash,
 ) (ingest.StatsChangeProcessorResults, error) {
-	args := m.Called(checkpointLedger, ledgerProtocolVersion, bucketListHash)
+	args := m.Called(checkpointLedger, skipChecks, ledgerProtocolVersion, bucketListHash)
 	return args.Get(0).(ingest.StatsChangeProcessorResults), args.Error(1)
 }
 
@@ -527,6 +528,11 @@ func (m *mockSystem) VerifyRange(fromLedger, toLedger uint32, verifyState bool)
 	return args.Error(0)
 }
 
+func (m *mockSystem) BuildState(sequence uint32, skipChecks bool) error {
+	args := m.Called(sequence, skipChecks)
+	return args.Error(0)
+}
+
 func (m *mockSystem) ReingestRange(ledgerRanges []history.LedgerRange, force bool) error {
 	args := m.Called(ledgerRanges, force)
 	return args.Error(0)
diff --git a/services/horizon/internal/ingest/processor_runner_test.go b/services/horizon/internal/ingest/processor_runner_test.go
index 5cae018e..ea2a99ef 100644
--- a/services/horizon/internal/ingest/processor_runner_test.go
+++ b/services/horizon/internal/ingest/processor_runner_test.go
@@ -124,7 +124,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionHistoryArchive(t *testing.T) {
 		filters:        &MockFilters{},
 	}
 
-	_, err := runner.RunHistoryArchiveIngestion(63, MaxSupportedProtocolVersion, bucketListHash)
+	_, err := runner.RunHistoryArchiveIngestion(63, false, MaxSupportedProtocolVersion, bucketListHash)
 	assert.NoError(t, err)
 }
 
@@ -159,7 +159,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionProtocolVersionNotSupported(t
 		filters:        &MockFilters{},
 	}
 
-	_, err := runner.RunHistoryArchiveIngestion(100, 200, xdr.Hash{})
+	_, err := runner.RunHistoryArchiveIngestion(100, false, 200, xdr.Hash{})
 	assert.EqualError(t, err,
 		fmt.Sprintf(
 			"Error while checking for supported protocol version: This Horizon version does not support protocol version 200. The latest supported protocol version is %d. Please upgrade to the latest Horizon version.",
diff --git a/services/horizon/internal/ingest/verify_range_state_test.go b/services/horizon/internal/ingest/verify_range_state_test.go
index 49a4e417..75298db9 100644
--- a/services/horizon/internal/ingest/verify_range_state_test.go
+++ b/services/horizon/internal/ingest/verify_range_state_test.go
@@ -160,7 +160,7 @@ func (s *VerifyRangeStateTestSuite) TestRunHistoryArchiveIngestionReturnsError()
 		},
 	}
 	s.ledgerBackend.On("GetLedger", s.ctx, uint32(100)).Return(meta, nil).Once()
-	s.runner.On("RunHistoryArchiveIngestion", uint32(100), MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).Return(ingest.StatsChangeProcessorResults{}, errors.New("my error")).Once()
+	s.runner.On("RunHistoryArchiveIngestion", uint32(100), false, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).Return(ingest.StatsChangeProcessorResults{}, errors.New("my error")).Once()
 
 	next, err := verifyRangeState{fromLedger: 100, toLedger: 200}.run(s.system)
 	s.Assert().Error(err)
@@ -188,7 +188,7 @@ func (s *VerifyRangeStateTestSuite) TestSuccess() {
 		},
 	}
 	s.ledgerBackend.On("GetLedger", s.ctx, uint32(100)).Return(meta, nil).Once()
-	s.runner.On("RunHistoryArchiveIngestion", uint32(100), MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).Return(ingest.StatsChangeProcessorResults{}, nil).Once()
+	s.runner.On("RunHistoryArchiveIngestion", uint32(100), false, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).Return(ingest.StatsChangeProcessorResults{}, nil).Once()
 
 	s.historyQ.On("UpdateLastLedgerIngest", s.ctx, uint32(100)).Return(nil).Once()
 	s.historyQ.On("Commit").Return(nil).Once()
@@ -244,7 +244,7 @@ func (s *VerifyRangeStateTestSuite) TestSuccessWithVerify() {
 		},
 	}
 	s.ledgerBackend.On("GetLedger", s.ctx, uint32(100)).Return(meta, nil).Once()
-	s.runner.On("RunHistoryArchiveIngestion", uint32(100), MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).Return(ingest.StatsChangeProcessorResults{}, nil).Once()
+	s.runner.On("RunHistoryArchiveIngestion", uint32(100), false, MaxSupportedProtocolVersion, xdr.Hash{1, 2, 3}).Return(ingest.StatsChangeProcessorResults{}, nil).Once()
 
 	s.historyQ.On("UpdateLastLedgerIngest", s.ctx, uint32(100)).Return(nil).Once()
 	s.historyQ.On("Commit").Return(nil).Once()