Repository: stellar/go
Pull request #4860 (2023-05-02 17:11:57)

services/horizon: Suppress Core timeout error when ingestion state ma…

<!-- 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

In the ingest package, I have added a state variable to store the current state of the ingestion state machine. I query this state from the `app` and suppress the error message only when the state is in the `build` state.


### Why

Fixes https://github.com/stellar/go/issues/4446


### Known limitations

I'm unsure if this is the best approach. It seems like overkill to reveal the current state of the ingestion state machine to suppress an error message. 


Patch:
diff --git a/services/horizon/internal/app.go b/services/horizon/internal/app.go
index 6b9376b7..119e8757 100644
--- a/services/horizon/internal/app.go
+++ b/services/horizon/internal/app.go
@@ -218,6 +218,14 @@ func (a *App) Paths() paths.Finder {
 func (a *App) UpdateCoreLedgerState(ctx context.Context) {
 	var next ledger.CoreStatus
 
+	// #4446 If the ingestion state machine is in the build state, the query can time out
+	// because the captive-core buffer may be full. In this case, skip the check.
+	if a.config.CaptiveCoreToml != nil &&
+		a.config.StellarCoreURL == fmt.Sprintf("http://localhost:%d", a.config.CaptiveCoreToml.HTTPPort) &&
+		a.ingester != nil && a.ingester.GetCurrentState() == ingest.Build {
+		return
+	}
+
 	logErr := func(err error, msg string) {
 		log.WithStack(err).WithField("err", err.Error()).Error(msg)
 	}
diff --git a/services/horizon/internal/ingest/fsm.go b/services/horizon/internal/ingest/fsm.go
index a8c5f64d..38e3fe9e 100644
--- a/services/horizon/internal/ingest/fsm.go
+++ b/services/horizon/internal/ingest/fsm.go
@@ -30,9 +30,25 @@ func (e ErrReingestRangeConflict) Error() string {
 	return fmt.Sprintf("reingest range overlaps with horizon ingestion, supplied range shouldn't contain ledger %d", e.maximumLedgerSequence)
 }
 
+type State int
+
+const (
+	None State = iota
+	Start
+	Stop
+	Build
+	Resume
+	WaitForCheckpoint
+	StressTest
+	VerifyRange
+	HistoryRange
+	ReingestHistoryRange
+)
+
 type stateMachineNode interface {
 	run(*system) (transition, error)
 	String() string
+	GetState() State
 }
 
 type transition struct {
@@ -105,6 +121,10 @@ func (stopState) String() string {
 	return "stop"
 }
 
+func (stopState) GetState() State {
+	return Stop
+}
+
 func (stopState) run(s *system) (transition, error) {
 	return stop(), errors.New("Cannot run terminal state")
 }
@@ -117,6 +137,10 @@ func (startState) String() string {
 	return "start"
 }
 
+func (startState) GetState() State {
+	return Start
+}
+
 func (state startState) run(s *system) (transition, error) {
 	if err := s.historyQ.Begin(); err != nil {
 		return start(), errors.Wrap(err, "Error starting a transaction")
@@ -234,6 +258,10 @@ func (b buildState) String() string {
 	return fmt.Sprintf("buildFromCheckpoint(checkpointLedger=%d, skipChecks=%t)", b.checkpointLedger, b.skipChecks)
 }
 
+func (buildState) GetState() State {
+	return Build
+}
+
 func (b buildState) run(s *system) (transition, error) {
 	var nextFailState = start()
 	if b.stop {
@@ -377,6 +405,10 @@ func (r resumeState) String() string {
 	return fmt.Sprintf("resume(latestSuccessfullyProcessedLedger=%d)", r.latestSuccessfullyProcessedLedger)
 }
 
+func (resumeState) GetState() State {
+	return Resume
+}
+
 func (r resumeState) run(s *system) (transition, error) {
 	if r.latestSuccessfullyProcessedLedger == 0 {
 		return start(), errors.New("unexpected latestSuccessfullyProcessedLedger value")
@@ -566,6 +598,10 @@ func (h historyRangeState) String() string {
 	)
 }
 
+func (historyRangeState) GetState() State {
+	return HistoryRange
+}
+
 // historyRangeState is used when catching up history data
 func (h historyRangeState) run(s *system) (transition, error) {
 	if h.fromLedger == 0 || h.toLedger == 0 ||
@@ -679,6 +715,10 @@ func (h reingestHistoryRangeState) String() string {
 	)
 }
 
+func (reingestHistoryRangeState) GetState() State {
+	return ReingestHistoryRange
+}
+
 func (h reingestHistoryRangeState) ingestRange(s *system, fromLedger, toLedger uint32) error {
 	if s.historyQ.GetTx() == nil {
 		return errors.New("expected transaction to be present")
@@ -833,6 +873,10 @@ func (waitForCheckpointState) String() string {
 	return "waitForCheckpoint"
 }
 
+func (waitForCheckpointState) GetState() State {
+	return WaitForCheckpoint
+}
+
 func (waitForCheckpointState) run(*system) (transition, error) {
 	log.Info("Waiting for the next checkpoint...")
 	time.Sleep(10 * time.Second)
@@ -854,6 +898,10 @@ func (v verifyRangeState) String() string {
 	)
 }
 
+func (verifyRangeState) GetState() State {
+	return VerifyRange
+}
+
 func (v verifyRangeState) run(s *system) (transition, error) {
 	if v.fromLedger == 0 || v.toLedger == 0 ||
 		v.fromLedger > v.toLedger {
@@ -985,6 +1033,10 @@ func (stressTestState) String() string {
 	return "stressTest"
 }
 
+func (stressTestState) GetState() State {
+	return StressTest
+}
+
 func (stressTestState) run(s *system) (transition, error) {
 	if err := s.historyQ.Begin(); err != nil {
 		err = errors.Wrap(err, "Error starting a transaction")
diff --git a/services/horizon/internal/ingest/main.go b/services/horizon/internal/ingest/main.go
index 735bfc6d..ab418420 100644
--- a/services/horizon/internal/ingest/main.go
+++ b/services/horizon/internal/ingest/main.go
@@ -196,6 +196,7 @@ type System interface {
 	ReingestRange(ledgerRanges []history.LedgerRange, force bool) error
 	BuildGenesisState() error
 	Shutdown()
+	GetCurrentState() State
 }
 
 type system struct {
@@ -228,6 +229,8 @@ type system struct {
 	runStateVerificationOnLedger func(uint32) bool
 
 	reapOffsets map[string]int64
+
+	currentState State
 }
 
 func NewSystem(config Config) (System, error) {
@@ -292,6 +295,7 @@ func NewSystem(config Config) (System, error) {
 		cancel:                      cancel,
 		config:                      config,
 		ctx:                         ctx,
+		currentState:                None,
 		disableStateVerification:    config.DisableStateVerification,
 		historyAdapter:              historyAdapter,
 		historyQ:                    historyQ,
@@ -479,6 +483,10 @@ func (s *system) initMetrics() {
 	)
 }
 
+func (s *system) GetCurrentState() State {
+	return s.currentState
+}
+
 func (s *system) Metrics() Metrics {
 	return s.metrics
 }
@@ -644,6 +652,7 @@ func (s *system) runStateMachine(cur stateMachineNode) error {
 			panic("unexpected transaction")
 		}
 
+		s.currentState = cur.GetState()
 		next, err := cur.run(s)
 		if err != nil {
 			logger := log.WithFields(logpkg.F{


Test patch:
diff --git a/services/horizon/internal/ingest/main_test.go b/services/horizon/internal/ingest/main_test.go
index a9d9bb37..65c9f47e 100644
--- a/services/horizon/internal/ingest/main_test.go
+++ b/services/horizon/internal/ingest/main_test.go
@@ -556,6 +556,11 @@ func (m *mockSystem) BuildGenesisState() error {
 	return args.Error(0)
 }
 
+func (m *mockSystem) GetCurrentState() State {
+	args := m.Called()
+	return args.Get(0).(State)
+}
+
 func (m *mockSystem) Shutdown() {
 	m.Called()
 }