services/horizon/internal/ingest: Add retries in historyRangeState when ledger is not available

services/horizon/CHANGELOG.md

@@ -3,6 +3,10 @@
 All notable changes to this project will be documented in this
 file. This project adheres to [Semantic Versioning](http://semver.org/).
 
+## v2.1.1
+
+* When ingesting a backlog of ledgers, Horizon sometimes consumes ledgers faster than the rate which Captive Core emits them. Previously this scenario caused failures in the ingestion system. That is now fixed in ([3531](https://github.com/stellar/go/pull/3531)).
+
 ## v2.1.0
 
 ### DB State Migration

services/horizon/internal/ingest/fsm.go

@@ -534,6 +534,11 @@ func (h historyRangeState) String() string {
 	)
 }
 
+var (
+	maxledgerNotFoundRetries   = 4
+	ledgerNotFoundRetryBackOff = 3 * time.Second
+)
+
 // historyRangeState is used when catching up history data
 func (h historyRangeState) run(s *system) (transition, error) {
 	if h.fromLedger == 0 || h.toLedger == 0 ||
@@ -570,8 +575,22 @@ func (h historyRangeState) run(s *system) (transition, error) {
 	}
 
 	for cur := h.fromLedger; cur <= h.toLedger; cur++ {
-		if err = runTransactionProcessorsOnLedger(s, cur); err != nil {
-			return start(), err
+		for attempt := 0; attempt < maxledgerNotFoundRetries; attempt++ {
+			if err = runTransactionProcessorsOnLedger(s, cur); err == nil {
+				break
+			} else if errors.Cause(err) == ingest.ErrNotFound {
+				if attempt == maxledgerNotFoundRetries-1 {
+					return start(), errors.Wrapf(
+						err,
+						"Could not obtain ledger %v after %v attempts",
+						cur,
+						maxledgerNotFoundRetries,
+					)
+				}
+				time.Sleep(ledgerNotFoundRetryBackOff)
+			} else {
+				return start(), err
+			}
 		}
 	}
 

services/horizon/internal/ingest/ingest_history_range_state_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/jmoiron/sqlx"
 	"github.com/stretchr/testify/suite"
 
+	"github.com/stellar/go/ingest"
 	"github.com/stellar/go/ingest/ledgerbackend"
 	"github.com/stellar/go/services/horizon/internal/ingest/processors"
 	"github.com/stellar/go/services/horizon/internal/toid"
@@ -42,6 +43,9 @@ func (s *IngestHistoryRangeStateTestSuite) SetupTest() {
 	}
 	s.system.initMetrics()
 
+	// configure back off to 0 so we can speed up tests
+	ledgerNotFoundRetryBackOff = 0
+
 	s.historyQ.On("Rollback").Return(nil).Once()
 }
 
@@ -140,6 +144,51 @@ func (s *IngestHistoryRangeStateTestSuite) TestRunTransactionProcessorsOnLedgerR
 	s.Assert().Equal(transition{node: startState{}, sleepDuration: defaultSleep}, next)
 }
 
+func (s *IngestHistoryRangeStateTestSuite) TestRunTransactionProcessorsRetrySucceeds() {
+	s.historyQ.On("Begin").Return(nil).Once()
+	s.historyQ.On("GetLastLedgerIngest").Return(uint32(0), nil).Once()
+	s.historyQ.On("GetLatestLedger").Return(uint32(99), nil).Once()
+
+	s.ledgerBackend.On("IsPrepared", ledgerbackend.UnboundedRange(100)).Return(true, nil).Once()
+	s.system.maxReingestRetries = 2
+	s.runner.On("RunTransactionProcessorsOnLedger", uint32(100)).Return(
+		processors.StatsLedgerTransactionProcessorResults{},
+		processorsRunDurations{},
+		errors.Wrap(errors.Wrap(ingest.ErrNotFound, "layer1"), "layer2"),
+	).Once()
+
+	s.runner.On("RunTransactionProcessorsOnLedger", uint32(100)).Return(
+		processors.StatsLedgerTransactionProcessorResults{},
+		processorsRunDurations{},
+		nil,
+	).Once()
+
+	s.historyQ.On("Commit").Return(nil).Once()
+
+	next, err := historyRangeState{fromLedger: 100, toLedger: 100}.run(s.system)
+	s.Assert().NoError(err)
+	s.Assert().Equal(transition{node: startState{}, sleepDuration: defaultSleep}, next)
+}
+
+func (s *IngestHistoryRangeStateTestSuite) TestRunTransactionProcessorsRetryFails() {
+	s.historyQ.On("Begin").Return(nil).Once()
+	s.historyQ.On("GetLastLedgerIngest").Return(uint32(0), nil).Once()
+	s.historyQ.On("GetLatestLedger").Return(uint32(99), nil).Once()
+
+	s.ledgerBackend.On("IsPrepared", ledgerbackend.UnboundedRange(100)).Return(true, nil).Once()
+	s.system.maxReingestRetries = 2
+	s.runner.On("RunTransactionProcessorsOnLedger", uint32(100)).Return(
+		processors.StatsLedgerTransactionProcessorResults{},
+		processorsRunDurations{},
+		errors.Wrap(errors.Wrap(ingest.ErrNotFound, "layer1"), "layer2"),
+	).Times(4)
+
+	next, err := historyRangeState{fromLedger: 100, toLedger: 200}.run(s.system)
+	s.Assert().Error(err)
+	s.Assert().EqualError(err, "Could not obtain ledger 100 after 4 attempts: error processing ledger sequence=100: layer2: layer1: ledger not found")
+	s.Assert().Equal(transition{node: startState{}, sleepDuration: defaultSleep}, next)
+}
+
 func (s *IngestHistoryRangeStateTestSuite) TestRangeNotPreparedFailPrepare() {
 	s.historyQ.On("Begin").Return(nil).Once()
 	s.historyQ.On("GetLastLedgerIngest").Return(uint32(0), nil).Once()