Repository: stellar/go
Pull request #5132 (2023-11-29 15:14:15)

services/horizon/internal/ingest: Fix error transition in resume state

<!-- 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

If there is an error rebuilding trade aggregation buckets we should retry ingesting the ledger. The previous code terminated ingestion entirely when encountering a rebuild trade aggregations error.

### Why

Rebuilding trade aggregation buckets could fail because of transient db connection errors. In that scenario we should retry.

### Known limitations

[N/A]



Patch:
diff --git a/services/horizon/internal/ingest/fsm.go b/services/horizon/internal/ingest/fsm.go
index 4d764093..f5b4f944 100644
--- a/services/horizon/internal/ingest/fsm.go
+++ b/services/horizon/internal/ingest/fsm.go
@@ -513,7 +513,7 @@ func (r resumeState) run(s *system) (transition, error) {
 	rebuildStart := time.Now()
 	err = s.historyQ.RebuildTradeAggregationBuckets(s.ctx, ingestLedger, ingestLedger, s.config.RoundingSlippageFilter)
 	if err != nil {
-		return stop(), errors.Wrap(err, "error rebuilding trade aggregations")
+		return retryResume(r), errors.Wrap(err, "error rebuilding trade aggregations")
 	}
 	rebuildDuration := time.Since(rebuildStart).Seconds()
 	s.Metrics().LedgerIngestionTradeAggregationDuration.Observe(float64(rebuildDuration))


Test patch:
diff --git a/services/horizon/internal/ingest/resume_state_test.go b/services/horizon/internal/ingest/resume_state_test.go
index 90cbedb7..82a7869d 100644
--- a/services/horizon/internal/ingest/resume_state_test.go
+++ b/services/horizon/internal/ingest/resume_state_test.go
@@ -374,6 +374,36 @@ func (s *ResumeTestTestSuite) TestErrorSettingCursorIgnored() {
 	)
 }
 
+func (s *ResumeTestTestSuite) TestRebuildTradeAggregationBucketsError() {
+	s.historyQ.On("Begin", s.ctx).Return(nil).Once()
+	s.historyQ.On("GetLastLedgerIngest", s.ctx).Return(uint32(100), nil).Once()
+	s.historyQ.On("GetIngestVersion", s.ctx).Return(CurrentVersion, nil).Once()
+	s.historyQ.On("GetLatestHistoryLedger", s.ctx).Return(uint32(100), nil)
+
+	s.runner.On("RunAllProcessorsOnLedger", mock.AnythingOfType("xdr.LedgerCloseMeta")).
+		Run(func(args mock.Arguments) {
+			meta := args.Get(0).(xdr.LedgerCloseMeta)
+			s.Assert().Equal(uint32(101), meta.LedgerSequence())
+		}).
+		Return(
+			ledgerStats{},
+			nil,
+		).Once()
+
+	s.historyQ.On("RebuildTradeAggregationBuckets", s.ctx, uint32(101), uint32(101), 0).
+		Return(errors.New("transient error")).Once()
+
+	next, err := resumeState{latestSuccessfullyProcessedLedger: 100}.run(s.system)
+	s.Assert().EqualError(err, "error rebuilding trade aggregations: transient error")
+	s.Assert().Equal(
+		transition{
+			node:          resumeState{latestSuccessfullyProcessedLedger: 100},
+			sleepDuration: defaultSleep,
+		},
+		next,
+	)
+}
+
 func (s *ResumeTestTestSuite) TestReapingObjectsDisabled() {
 	s.historyQ.On("Begin", s.ctx).Return(nil).Once()
 	s.historyQ.On("GetLastLedgerIngest", s.ctx).Return(uint32(100), nil).Once()