services/horizon: Propogate *all* history archives through the ingest…

…ion interface. (#4687)
stellar · Nov 17, 2022 · 0666818 · 0666818
1 parent 6eaca4f
commit 0666818
Show file tree

Hide file tree

Showing 7 changed files with 22 additions and 17 deletions.
diff --git 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
@@ -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
@@ -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/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.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/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,
 	}
 

diff --git 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,