services/horizon: Add DISABLE_SOROBAN_INGEST flag to skip soroban ingestion processing

services/horizon/cmd/db.go

@@ -419,6 +419,7 @@ func runDBReingestRange(ledgerRanges []history.LedgerRange, reingestForce bool,
 		RoundingSlippageFilter:      config.RoundingSlippageFilter,
 		EnableIngestionFiltering:    config.EnableIngestionFiltering,
 		MaxLedgerPerFlush:           maxLedgersPerFlush,
+		SkipSorobanIngestion:        config.SkipSorobanIngestion,
 	}
 
 	if ingestConfig.HistorySession, err = db.Open("postgres", config.DatabaseURL); err != nil {

services/horizon/internal/config.go

@@ -108,4 +108,6 @@ type Config struct {
 	Network string
 	// DisableTxSub disables transaction submission functionality for Horizon.
 	DisableTxSub bool
+	// SkipSorobanIngestion skips Soroban related ingestion processing.
+	SkipSorobanIngestion bool
 }

services/horizon/internal/flags.go

@@ -57,6 +57,8 @@ const (
 	EnableIngestionFilteringFlagName = "exp-enable-ingestion-filtering"
 	// DisableTxSubFlagName is the command line flag for disabling transaction submission feature of Horizon
 	DisableTxSubFlagName = "disable-tx-sub"
+	// SkipSorobanIngestionFlagName is the command line flag for disabling Soroban related ingestion processing
+	SkipSorobanIngestionFlagName = "disable-soroban-ingest-processors"
 
 	// StellarPubnet is a constant representing the Stellar public network
 	StellarPubnet = "pubnet"
@@ -730,6 +732,15 @@ func Flags() (*Config, support.ConfigOptions) {
 				HistoryArchiveURLsFlagName, CaptiveCoreConfigPathName),
 			UsedInCommands: IngestionCommands,
 		},
+		&support.ConfigOption{
+			Name:           SkipSorobanIngestionFlagName,
+			ConfigKey:      &config.SkipSorobanIngestion,
+			OptType:        types.Bool,
+			FlagDefault:    false,
+			Required:       false,
+			Usage:          "excludes Soroban data during ingestion processing",
+			UsedInCommands: IngestionCommands,
+		},
 	}
 
 	return config, flags

services/horizon/internal/ingest/group_processors.go

@@ -21,16 +21,25 @@ func (d processorsRunDurations) AddRunDuration(name string, startTime time.Time)
 type groupChangeProcessors struct {
 	processors []horizonChangeProcessor
 	processorsRunDurations
+	skipEntryType []xdr.LedgerEntryType
 }
 
-func newGroupChangeProcessors(processors []horizonChangeProcessor) *groupChangeProcessors {
+func newGroupChangeProcessors(processors []horizonChangeProcessor, skipEntryType []xdr.LedgerEntryType) *groupChangeProcessors {
 	return &groupChangeProcessors{
 		processors:             processors,
 		processorsRunDurations: make(map[string]time.Duration),
+		skipEntryType:          skipEntryType,
 	}
 }
 
 func (g groupChangeProcessors) ProcessChange(ctx context.Context, change ingest.Change) error {
+
+	for _, op := range g.skipEntryType {
+		if op == change.Type {
+			return nil
+		}
+	}
+
 	for _, p := range g.processors {
 		startTime := time.Now()
 		if err := p.ProcessChange(ctx, change); err != nil {

services/horizon/internal/ingest/group_processors_test.go

@@ -68,7 +68,7 @@ func (s *GroupChangeProcessorsTestSuiteLedger) SetupTest() {
 	s.processors = newGroupChangeProcessors([]horizonChangeProcessor{
 		s.processorA,
 		s.processorB,
-	})
+	}, []xdr.LedgerEntryType{})
 }
 
 func (s *GroupChangeProcessorsTestSuiteLedger) TearDownTest() {

services/horizon/internal/ingest/main.go

@@ -107,6 +107,8 @@ type Config struct {
 
 	EnableIngestionFiltering bool
 	MaxLedgerPerFlush        uint32
+
+	SkipSorobanIngestion bool
 }
 
 const (

services/horizon/internal/ingest/processor_runner.go

@@ -111,11 +111,23 @@ func buildChangeProcessor(
 	source ingestionSource,
 	ledgerSequence uint32,
 	networkPassphrase string,
+	skipSorobanIngestion bool,
 ) *groupChangeProcessors {
 	statsChangeProcessor := &statsChangeProcessor{
 		StatsChangeProcessor: changeStats,
 	}
 
+	var skipEntryType []xdr.LedgerEntryType
+
+	if skipSorobanIngestion {
+		skipEntryType = []xdr.LedgerEntryType{
+			xdr.LedgerEntryTypeContractData,
+			xdr.LedgerEntryTypeContractCode,
+			xdr.LedgerEntryTypeTtl,
+			xdr.LedgerEntryTypeConfigSetting,
+		}
+	}
+
 	useLedgerCache := source == ledgerSource
 	return newGroupChangeProcessors([]horizonChangeProcessor{
 		statsChangeProcessor,
@@ -127,7 +139,7 @@ func buildChangeProcessor(
 		processors.NewTrustLinesProcessor(historyQ),
 		processors.NewClaimableBalancesChangeProcessor(historyQ),
 		processors.NewLiquidityPoolsChangeProcessor(historyQ, ledgerSequence),
-	})
+	}, skipEntryType)
 }
 
 func (s *ProcessorRunner) buildTransactionProcessor(ledgersProcessor *processors.LedgersProcessor) *groupTransactionProcessors {
@@ -142,11 +154,22 @@ func (s *ProcessorRunner) buildTransactionProcessor(ledgersProcessor *processors
 	tradeProcessor := processors.NewTradeProcessor(accountLoader,
 		lpLoader, assetLoader, s.historyQ.NewTradeBatchInsertBuilder())
 
+	var skipOperationType []xdr.OperationType
+
+	if s.config.SkipSorobanIngestion {
+		skipOperationType = []xdr.OperationType{
+			xdr.OperationTypeInvokeHostFunction,
+			xdr.OperationTypeExtendFootprintTtl,
+			xdr.OperationTypeRestoreFootprint,
+			xdr.OperationTypeExtendFootprintTtl,
+		}
+	}
+
 	processors := []horizonTransactionProcessor{
 		statsLedgerTransactionProcessor,
-		processors.NewEffectProcessor(accountLoader, s.historyQ.NewEffectBatchInsertBuilder(), s.config.NetworkPassphrase),
+		processors.NewEffectProcessor(accountLoader, s.historyQ.NewEffectBatchInsertBuilder(), s.config.NetworkPassphrase, skipOperationType),
 		ledgersProcessor,
-		processors.NewOperationProcessor(s.historyQ.NewOperationBatchInsertBuilder(), s.config.NetworkPassphrase),
+		processors.NewOperationProcessor(s.historyQ.NewOperationBatchInsertBuilder(), s.config.NetworkPassphrase, skipOperationType),
 		tradeProcessor,
 		processors.NewParticipantsProcessor(accountLoader,
 			s.historyQ.NewTransactionParticipantsBatchInsertBuilder(), s.historyQ.NewOperationParticipantBatchInsertBuilder()),
@@ -235,6 +258,7 @@ func (s *ProcessorRunner) RunHistoryArchiveIngestion(
 		historyArchiveSource,
 		checkpointLedger,
 		s.config.NetworkPassphrase,
+		s.config.SkipSorobanIngestion,
 	)
 
 	if checkpointLedger == 1 {
@@ -493,6 +517,7 @@ func (s *ProcessorRunner) RunAllProcessorsOnLedger(ledger xdr.LedgerCloseMeta) (
 		ledgerSource,
 		ledger.LedgerSequence(),
 		s.config.NetworkPassphrase,
+		s.config.SkipSorobanIngestion,
 	)
 	err = s.runChangeProcessorOnLedger(groupChangeProcessors, ledger)
 	if err != nil {

services/horizon/internal/ingest/processor_runner_test.go

@@ -180,7 +180,7 @@ func TestProcessorRunnerBuildChangeProcessor(t *testing.T) {
 	}
 
 	stats := &ingest.StatsChangeProcessor{}
-	processor := buildChangeProcessor(runner.historyQ, stats, ledgerSource, 123, "")
+	processor := buildChangeProcessor(runner.historyQ, stats, ledgerSource, 123, "", false)
 	assert.IsType(t, &groupChangeProcessors{}, processor)
 
 	assert.IsType(t, &statsChangeProcessor{}, processor.processors[0])
@@ -201,7 +201,7 @@ func TestProcessorRunnerBuildChangeProcessor(t *testing.T) {
 		filters:  &MockFilters{},
 	}
 
-	processor = buildChangeProcessor(runner.historyQ, stats, historyArchiveSource, 456, "")
+	processor = buildChangeProcessor(runner.historyQ, stats, historyArchiveSource, 456, "", false)
 	assert.IsType(t, &groupChangeProcessors{}, processor)
 
 	assert.IsType(t, &statsChangeProcessor{}, processor.processors[0])

services/horizon/internal/ingest/processors/effects_processor.go

@@ -25,20 +25,23 @@ import (
 
 // EffectProcessor process effects
 type EffectProcessor struct {
-	accountLoader *history.AccountLoader
-	batch         history.EffectBatchInsertBuilder
-	network       string
+	accountLoader     *history.AccountLoader
+	batch             history.EffectBatchInsertBuilder
+	network           string
+	skipOperationType []xdr.OperationType
 }
 
 func NewEffectProcessor(
 	accountLoader *history.AccountLoader,
 	batch history.EffectBatchInsertBuilder,
 	network string,
+	skipOperationType []xdr.OperationType,
 ) *EffectProcessor {
 	return &EffectProcessor{
-		accountLoader: accountLoader,
-		batch:         batch,
-		network:       network,
+		accountLoader:     accountLoader,
+		batch:             batch,
+		network:           network,
+		skipOperationType: skipOperationType,
 	}
 }
 
@@ -50,6 +53,7 @@ func (p *EffectProcessor) ProcessTransaction(
 		return nil
 	}
 
+OUTER:
 	for opi, op := range transaction.Envelope.Operations() {
 		operation := transactionOperationWrapper{
 			index:          uint32(opi),
@@ -58,6 +62,13 @@ func (p *EffectProcessor) ProcessTransaction(
 			ledgerSequence: uint32(lcm.LedgerSequence()),
 			network:        p.network,
 		}
+
+		for _, op := range p.skipOperationType {
+			if op == operation.OperationType() {
+				continue OUTER
+			}
+		}
+
 		if err := operation.ingestEffects(p.accountLoader, p.batch); err != nil {
 			return errors.Wrapf(err, "reading operation %v effects", operation.ID())
 		}

services/horizon/internal/ingest/processors/effects_processor_test.go

@@ -143,6 +143,7 @@ func (s *EffectsProcessorTestSuiteLedger) SetupTest() {
 		s.accountLoader,
 		s.mockBatchInsertBuilder,
 		networkPassphrase,
+		[]xdr.OperationType{},
 	)
 
 	s.txs = []ingest.LedgerTransaction{

services/horizon/internal/ingest/processors/operations_processor.go

@@ -21,19 +21,23 @@
 
 // OperationProcessor operations processor
 type OperationProcessor struct {
-	batch   history.OperationBatchInsertBuilder
-	network string
+	batch             history.OperationBatchInsertBuilder
+	network           string
+	skipOperationType []xdr.OperationType
 }
 
-func NewOperationProcessor(batch history.OperationBatchInsertBuilder, network string) *OperationProcessor {
+func NewOperationProcessor(batch history.OperationBatchInsertBuilder, network string, skipOperationType []xdr.OperationType) *OperationProcessor {
 	return &OperationProcessor{
-		batch:   batch,
-		network: network,
+		batch:             batch,
+		network:           network,
+		skipOperationType: skipOperationType,
 	}
 }
 
 // ProcessTransaction process the given transaction
 func (p *OperationProcessor) ProcessTransaction(lcm xdr.LedgerCloseMeta, transaction ingest.LedgerTransaction) error {
+
+OUTER:
 	for i, op := range transaction.Envelope.Operations() {
 		operation := transactionOperationWrapper{
 			index:          uint32(i),
@@ -42,6 +46,12 @@
 			ledgerSequence: lcm.LedgerSequence(),
 			network:        p.network,
 		}
+		for _, op := range p.skipOperationType {
+			if op == operation.OperationType() {
+				continue OUTER
+			}
+		}
+
 		details, err := operation.Details()
 		if err != nil {
 			return errors.Wrapf(err, "Error obtaining details for operation %v", operation.ID())

services/horizon/internal/ingest/processors/operations_processor_test.go

@@ -42,6 +42,7 @@ func (s *OperationsProcessorTestSuiteLedger) SetupTest() {
 	s.processor = NewOperationProcessor(
 		s.mockBatchInsertBuilder,
 		"test network",
+		[]xdr.OperationType{},
 	)
 }
 

services/horizon/internal/ingest/verify_test.go

@@ -292,7 +292,7 @@ func TestStateVerifierLockBusy(t *testing.T) {
 	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
 
 	checkpointLedger := uint32(63)
-	changeProcessor := buildChangeProcessor(q, &ingest.StatsChangeProcessor{}, ledgerSource, checkpointLedger, "")
+	changeProcessor := buildChangeProcessor(q, &ingest.StatsChangeProcessor{}, ledgerSource, checkpointLedger, "", false)
 
 	gen := randxdr.NewGenerator()
 	var changes []xdr.LedgerEntryChange
@@ -350,7 +350,7 @@ func TestStateVerifier(t *testing.T) {
 
 	ledger := rand.Int31()
 	checkpointLedger := uint32(ledger - (ledger % 64) - 1)
-	changeProcessor := buildChangeProcessor(q, &ingest.StatsChangeProcessor{}, ledgerSource, checkpointLedger, "")
+	changeProcessor := buildChangeProcessor(q, &ingest.StatsChangeProcessor{}, ledgerSource, checkpointLedger, "", false)
 	mockChangeReader := &ingest.MockChangeReader{}
 
 	gen := randxdr.NewGenerator()

services/horizon/internal/init.go

@@ -110,6 +110,7 @@ func initIngester(app *App) {
 		EnableExtendedLogLedgerStats:         app.config.IngestEnableExtendedLogLedgerStats,
 		RoundingSlippageFilter:               app.config.RoundingSlippageFilter,
 		EnableIngestionFiltering:             app.config.EnableIngestionFiltering,
+		SkipSorobanIngestion:                 app.config.SkipSorobanIngestion,
 	})
 
 	if err != nil {