Repository: stellar/go
Pull request #4687 (2022-11-07 23:59:30)

services/horizon: Propogate *all* history archives through the ingestion interface.
<!-- 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

* [x] This PR has reasonably narrow scope (if not, break it down into smaller PRs).
* [x] This PR avoids mixing refactoring changes with feature changes (split into two PRs
  otherwise).
* [x] 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.
* [x] 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.
* [x] 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
Use the entire list of provided history archives when ingesting.

### Why
Though #3402 allows Captive Core to rotate through a list of history archives, the ingestion interface continued to use the first one in the list. This creates undue strain on certain HA hosts.

### Known limitations
n/a


Patch:
diff --git a/services/horizon/CHANGELOG.md b/services/horizon/CHANGELOG.md
index 0091ee66..566fa104 100644
--- a/services/horizon/CHANGELOG.md
+++ b/services/horizon/CHANGELOG.md
@@ -3,6 +3,14 @@
 All notable changes to this project will be documented in this
 file. This project adheres to [Semantic Versioning](http://semver.org/).
 
+
+## Unreleased
+
+### Fixes
+
+* The ingestion subsystem will now properly use a pool of history archives if more than one is provided ([#4687](https://github.com/stellar/go/pull/4687)).
+
+
 ## 2.22.1
 
 **Upgrading to this version from <= v2.8.3 will trigger a state rebuild. During this process (which will take at least 10 minutes), Horizon will not ingest new ledgers.**
diff --git a/services/horizon/cmd/db.go b/services/horizon/cmd/db.go
index b29535c4..37a16653 100644
--- a/services/horizon/cmd/db.go
+++ b/services/horizon/cmd/db.go
@@ -401,7 +401,7 @@ func runDBReingestRange(ledgerRanges []history.LedgerRange, reingestForce bool,
 
 	ingestConfig := ingest.Config{
 		NetworkPassphrase:           config.NetworkPassphrase,
-		HistoryArchiveURL:           config.HistoryArchiveURLs[0],
+		HistoryArchiveURLs:          config.HistoryArchiveURLs,
 		CheckpointFrequency:         config.CheckpointFrequency,
 		ReingestEnabled:             true,
 		MaxReingestRetries:          int(retries),
diff --git a/services/horizon/cmd/ingest.go b/services/horizon/cmd/ingest.go
index 2cae1fea..c0fdaf75 100644
--- a/services/horizon/cmd/ingest.go
+++ b/services/horizon/cmd/ingest.go
@@ -127,7 +127,7 @@ var ingestVerifyRangeCmd = &cobra.Command{
 		ingestConfig := ingest.Config{
 			NetworkPassphrase:        config.NetworkPassphrase,
 			HistorySession:           horizonSession,
-			HistoryArchiveURL:        config.HistoryArchiveURLs[0],
+			HistoryArchiveURLs:       config.HistoryArchiveURLs,
 			EnableCaptiveCore:        config.EnableCaptiveCoreIngestion,
 			CaptiveCoreBinaryPath:    config.CaptiveCoreBinaryPath,
 			CaptiveCoreConfigUseDB:   config.CaptiveCoreConfigUseDB,
@@ -223,7 +223,7 @@ var ingestStressTestCmd = &cobra.Command{
 		ingestConfig := ingest.Config{
 			NetworkPassphrase:      config.NetworkPassphrase,
 			HistorySession:         horizonSession,
-			HistoryArchiveURL:      config.HistoryArchiveURLs[0],
+			HistoryArchiveURLs:     config.HistoryArchiveURLs,
 			EnableCaptiveCore:      config.EnableCaptiveCoreIngestion,
 			RoundingSlippageFilter: config.RoundingSlippageFilter,
 		}
@@ -314,7 +314,7 @@ var ingestInitGenesisStateCmd = &cobra.Command{
 		ingestConfig := ingest.Config{
 			NetworkPassphrase:        config.NetworkPassphrase,
 			HistorySession:           horizonSession,
-			HistoryArchiveURL:        config.HistoryArchiveURLs[0],
+			HistoryArchiveURLs:       config.HistoryArchiveURLs,
 			EnableCaptiveCore:        config.EnableCaptiveCoreIngestion,
 			CheckpointFrequency:      config.CheckpointFrequency,
 			RoundingSlippageFilter:   config.RoundingSlippageFilter,
@@ -391,7 +391,7 @@ var ingestBuildStateCmd = &cobra.Command{
 		ingestConfig := ingest.Config{
 			NetworkPassphrase:        config.NetworkPassphrase,
 			HistorySession:           horizonSession,
-			HistoryArchiveURL:        config.HistoryArchiveURLs[0],
+			HistoryArchiveURLs:       config.HistoryArchiveURLs,
 			EnableCaptiveCore:        config.EnableCaptiveCoreIngestion,
 			CaptiveCoreBinaryPath:    config.CaptiveCoreBinaryPath,
 			CaptiveCoreConfigUseDB:   config.CaptiveCoreConfigUseDB,
diff --git a/services/horizon/internal/ingest/main.go b/services/horizon/internal/ingest/main.go
index 4f1de4a2..465965de 100644
--- a/services/horizon/internal/ingest/main.go
+++ b/services/horizon/internal/ingest/main.go
@@ -81,8 +81,8 @@ type Config struct {
 	RemoteCaptiveCoreURL   string
 	NetworkPassphrase      string
 
-	HistorySession    db.SessionInterface
-	HistoryArchiveURL string
+	HistorySession     db.SessionInterface
+	HistoryArchiveURLs []string
 
 	DisableStateVerification     bool
 	EnableReapLookupTables       bool
@@ -214,8 +214,8 @@ type system struct {
 func NewSystem(config Config) (System, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 
-	archive, err := historyarchive.Connect(
-		config.HistoryArchiveURL,
+	archive, err := historyarchive.NewArchivePool(
+		config.HistoryArchiveURLs,
 		historyarchive.ConnectOptions{
 			Context:             ctx,
 			NetworkPassphrase:   config.NetworkPassphrase,
@@ -245,7 +245,7 @@ func NewSystem(config Config) (System, error) {
 					UseDB:               config.CaptiveCoreConfigUseDB,
 					Toml:                config.CaptiveCoreToml,
 					NetworkPassphrase:   config.NetworkPassphrase,
-					HistoryArchiveURLs:  []string{config.HistoryArchiveURL},
+					HistoryArchiveURLs:  config.HistoryArchiveURLs,
 					CheckpointFrequency: config.CheckpointFrequency,
 					LedgerHashStore:     ledgerbackend.NewHorizonDBLedgerHashStore(config.HistorySession),
 					Log:                 logger,
diff --git a/services/horizon/internal/init.go b/services/horizon/internal/init.go
index 4150a1e2..33cee8ee 100644
--- a/services/horizon/internal/init.go
+++ b/services/horizon/internal/init.go
@@ -102,11 +102,8 @@ func initIngester(app *App) {
 		HistorySession: mustNewDBSession(
 			db.IngestSubservice, app.config.DatabaseURL, ingest.MaxDBConnections, ingest.MaxDBConnections, app.prometheusRegistry,
 		),
-		NetworkPassphrase: app.config.NetworkPassphrase,
-		// TODO:
-		// Use the first archive for now. We don't have a mechanism to
-		// use multiple archives at the same time currently.
-		HistoryArchiveURL:            app.config.HistoryArchiveURLs[0],
+		NetworkPassphrase:            app.config.NetworkPassphrase,
+		HistoryArchiveURLs:           app.config.HistoryArchiveURLs,
 		CheckpointFrequency:          app.config.CheckpointFrequency,
 		StellarCoreURL:               app.config.StellarCoreURL,
 		StellarCoreCursor:            app.config.CursorName,


Test patch:
diff --git a/services/horizon/internal/ingest/db_integration_test.go b/services/horizon/internal/ingest/db_integration_test.go
index 2857a277..86576db1 100644
--- a/services/horizon/internal/ingest/db_integration_test.go
+++ b/services/horizon/internal/ingest/db_integration_test.go
@@ -83,7 +83,7 @@ func (s *DBTestSuite) SetupTest() {
 	sIface, err := NewSystem(Config{
 		CoreSession:              s.tt.CoreSession(),
 		HistorySession:           s.tt.HorizonSession(),
-		HistoryArchiveURL:        "http://ignore.test",
+		HistoryArchiveURLs:       []string{"http://ignore.test"},
 		DisableStateVerification: false,
 		CheckpointFrequency:      64,
 	})
diff --git a/services/horizon/internal/ingest/main_test.go b/services/horizon/internal/ingest/main_test.go
index 0ee2a80b..c8e72022 100644
--- a/services/horizon/internal/ingest/main_test.go
+++ b/services/horizon/internal/ingest/main_test.go
@@ -84,7 +84,7 @@ func TestNewSystem(t *testing.T) {
 		CoreSession:              &db.Session{DB: &sqlx.DB{}},
 		HistorySession:           &db.Session{DB: &sqlx.DB{}},
 		DisableStateVerification: true,
-		HistoryArchiveURL:        "https://history.stellar.org/prd/core-live/core_live_001",
+		HistoryArchiveURLs:       []string{"https://history.stellar.org/prd/core-live/core_live_001"},
 		CheckpointFrequency:      64,
 	}