Add flag to skip protocol version check in Horizon (stellar#5834)

Author: leighmcculloch

services/horizon/internal/config.go

@@ -104,6 +104,9 @@ type Config struct {
 	// IngestEnableExtendedLogLedgerStats enables extended ledger stats in
 	// logging.
 	IngestEnableExtendedLogLedgerStats bool
+	// IngestSkipProtocolVersionCheck skips protocol version verification during
+	// ingestion.
+	IngestSkipProtocolVersionCheck bool
 	// ApplyMigrations will apply pending migrations to the horizon database
 	// before starting the horizon service
 	ApplyMigrations bool

services/horizon/internal/flags.go

@@ -729,6 +729,14 @@ func Flags() (*Config, support.ConfigOptions) {
 			Usage:          "enables extended ledger stats in the log (ledger entry changes and operations stats)",
 			UsedInCommands: IngestionCommands,
 		},
+		&support.ConfigOption{
+			Name:           "ingest-skip-protocol-version-check",
+			ConfigKey:      &config.IngestSkipProtocolVersionCheck,
+			OptType:        types.Bool,
+			FlagDefault:    false,
+			Usage:          "skip protocol version verification during ingestion",
+			UsedInCommands: IngestionCommands,
+		},
 		&support.ConfigOption{
 			Name:           "apply-migrations",
 			ConfigKey:      &config.ApplyMigrations,

services/horizon/internal/ingest/main.go

@@ -134,6 +134,7 @@ type Config struct {
 	DisableStateVerification     bool
 	ReapLookupTables             bool
 	EnableExtendedLogLedgerStats bool
+	SkipProtocolVersionCheck     bool
 
 	MaxReingestRetries          int
 	ReingestRetryBackoffSeconds int

services/horizon/internal/ingest/processor_runner.go

@@ -184,6 +184,9 @@ func (s *ProcessorRunner) buildFilteredOutProcessor() *groupTransactionProcessor
 // checkIfProtocolVersionSupported checks if this Horizon version supports the
 // protocol version of a ledger with the given sequence number.
 func (s *ProcessorRunner) checkIfProtocolVersionSupported(ledgerProtocolVersion uint32) error {
+	if s.config.SkipProtocolVersionCheck {
+		return nil
+	}
 	if ledgerProtocolVersion > MaxSupportedProtocolVersion {
 		return fmt.Errorf(
 			"This Horizon version does not support protocol version %d. "+

services/horizon/internal/ingest/processor_runner_test.go

@@ -401,7 +401,8 @@ func TestProcessorRunnerRunAllProcessorsOnLedgerProtocolVersionNotSupported(t *t
 	maxBatchSize := 100000
 
 	config := Config{
-		NetworkPassphrase: network.PublicNetworkPassphrase,
+		NetworkPassphrase:        network.PublicNetworkPassphrase,
+		SkipProtocolVersionCheck: false,
 	}
 
 	q := &mockDBQ{}
@@ -411,7 +412,7 @@ func TestProcessorRunnerRunAllProcessorsOnLedgerProtocolVersionNotSupported(t *t
 		V0: &xdr.LedgerCloseMetaV0{
 			LedgerHeader: xdr.LedgerHeaderHistoryEntry{
 				Header: xdr.LedgerHeader{
-					LedgerVersion: 200,
+					LedgerVersion: xdr.Uint32(MaxSupportedProtocolVersion + 1),
 				},
 			},
 		},
@@ -444,12 +445,85 @@ func TestProcessorRunnerRunAllProcessorsOnLedgerProtocolVersionNotSupported(t *t
 	_, err := runner.RunAllProcessorsOnLedger(ledger)
 	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.",
+			"Error while checking for supported protocol version: This Horizon version does not support protocol version %d. The latest supported protocol version is %d. Please upgrade to the latest Horizon version.",
+			MaxSupportedProtocolVersion+1,
 			MaxSupportedProtocolVersion,
 		),
 	)
 }
 
+func TestProcessorRunnerRunAllProcessorsOnLedgerProtocolVersionNotSupportedButAllowed(t *testing.T) {
+	ctx := context.Background()
+
+	config := Config{
+		NetworkPassphrase:        network.PublicNetworkPassphrase,
+		SkipProtocolVersionCheck: true,
+	}
+
+	mockSession := &db.MockSession{}
+	q := &mockDBQ{}
+	defer mock.AssertExpectationsForObjects(t, q)
+
+	ledger := xdr.LedgerCloseMeta{
+		V0: &xdr.LedgerCloseMetaV0{
+			LedgerHeader: xdr.LedgerHeaderHistoryEntry{
+				Header: xdr.LedgerHeader{
+					LedgerVersion:  xdr.Uint32(MaxSupportedProtocolVersion + 1),
+					BucketListHash: xdr.Hash([32]byte{0, 1, 2}),
+					LedgerSeq:      23,
+				},
+			},
+		},
+	}
+
+	// Batches
+	defer mock.AssertExpectationsForObjects(t, mockTxProcessorBatchBuilders(q, mockSession, ctx)...)
+	defer mock.AssertExpectationsForObjects(t, mockChangeProcessorBatchBuilders(q, ctx, true)...)
+	defer mock.AssertExpectationsForObjects(t, mockFilteredOutProcessorsForNoRules(q, mockSession, ctx)...)
+
+	mockBatchInsertBuilder := &history.MockLedgersBatchInsertBuilder{}
+	q.MockQLedgers.On("NewLedgerBatchInsertBuilder").Return(mockBatchInsertBuilder)
+	mockBatchInsertBuilder.On(
+		"Add",
+		ledger.V0.LedgerHeader, 0, 0, 0, 0, CurrentVersion).Return(nil).Once()
+	mockBatchInsertBuilder.On(
+		"Exec",
+		ctx,
+		mockSession,
+	).Return(nil).Once()
+
+	defer mock.AssertExpectationsForObjects(t, mockBatchInsertBuilder)
+
+	q.MockQAssetStats.On("InsertAssetContracts", ctx, []history.AssetContract(nil)).
+		Return(nil).Once()
+	q.MockQAssetStats.On("UpdateAssetContractExpirations", ctx, []xdr.Hash{}, []uint32{}).
+		Return(nil).Once()
+	q.MockQAssetStats.On("DeleteAssetContractsExpiringAt", ctx, uint32(22)).
+		Return(int64(0), nil).Once()
+
+	q.MockQAssetStats.On("RemoveContractAssetBalances", ctx, []xdr.Hash(nil)).
+		Return(nil).Once()
+	q.MockQAssetStats.On("UpdateContractAssetBalanceAmounts", ctx, []xdr.Hash{}, []string{}).
+		Return(nil).Once()
+	q.MockQAssetStats.On("InsertContractAssetBalances", ctx, []history.ContractAssetBalance(nil)).
+		Return(nil).Once()
+	q.MockQAssetStats.On("UpdateContractAssetBalanceExpirations", ctx, []xdr.Hash{}, []uint32{}).
+		Return(nil).Once()
+	q.MockQAssetStats.On("DeleteContractAssetBalancesExpiringAt", ctx, uint32(22)).
+		Return([]history.ContractAssetBalance{}, nil).Once()
+
+	runner := ProcessorRunner{
+		ctx:      ctx,
+		config:   config,
+		historyQ: q,
+		session:  mockSession,
+		filters:  &MockFilters{},
+	}
+
+	_, err := runner.RunAllProcessorsOnLedger(ledger)
+	assert.NoError(t, err)
+}
+
 func mockTxProcessorBatchBuilders(q *mockDBQ, mockSession *db.MockSession, ctx context.Context) []interface{} {
 	// no mocking of builder Add methods needed, the fake ledgers used in tests don't have any operations
 	// that would trigger the respective processors to invoke Add, each test locally decides to use

services/horizon/internal/init.go

@@ -100,6 +100,7 @@ func initIngester(app *App) {
 		StateVerificationTimeout:             app.config.IngestStateVerificationTimeout,
 		ReapLookupTables:                     app.config.ReapLookupTables && app.config.HistoryRetentionCount > 0,
 		EnableExtendedLogLedgerStats:         app.config.IngestEnableExtendedLogLedgerStats,
+		SkipProtocolVersionCheck:             app.config.IngestSkipProtocolVersionCheck,
 		RoundingSlippageFilter:               app.config.RoundingSlippageFilter,
 		SkipTxmeta:                           app.config.SkipTxmeta,
 		ReapConfig: ingest.ReapConfig{