Repository: stellar/go
Pull request #5321 (2024-05-17 09:19:32)

exp/services/ledgerexporter/internal: Call PrepareRange() in exporter manager goroutine
<!-- 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

Call PrepareRange() in exporter manager goroutine instead of during app initialization.

### Why

PrepareRange() can block up to 15 minutes because it will wait until the first ledger has been emitted over the captive core pipe. So if we call it during the app initialization we will have no metrics during this period because we first initialize the ledger backend before initializing the metrics server.

### Known limitations

[N/A]



Patch:
diff --git a/exp/services/ledgerexporter/internal/app.go b/exp/services/ledgerexporter/internal/app.go
index b90c7a33..fe84022d 100644
--- a/exp/services/ledgerexporter/internal/app.go
+++ b/exp/services/ledgerexporter/internal/app.go
@@ -112,7 +112,7 @@ func (a *App) init(ctx context.Context) error {
 
 	logger.Infof("Final computed ledger range for backend retrieval and export, start=%d, end=%d", a.config.StartLedger, a.config.EndLedger)
 
-	if a.ledgerBackend, err = newLedgerBackend(ctx, a.config, registry); err != nil {
+	if a.ledgerBackend, err = newLedgerBackend(a.config, registry); err != nil {
 		return err
 	}
 
@@ -247,7 +247,7 @@ func (a *App) Run() {
 
 // newLedgerBackend Creates and initializes captive core ledger backend
 // Currently, only supports captive-core as ledger backend
-func newLedgerBackend(ctx context.Context, config *Config, prometheusRegistry *prometheus.Registry) (ledgerbackend.LedgerBackend, error) {
+func newLedgerBackend(config *Config, prometheusRegistry *prometheus.Registry) (ledgerbackend.LedgerBackend, error) {
 	captiveConfig, err := config.GenerateCaptiveCoreConfig()
 	if err != nil {
 		return nil, err
@@ -261,15 +261,5 @@ func newLedgerBackend(ctx context.Context, config *Config, prometheusRegistry *p
 	}
 	backend = ledgerbackend.WithMetrics(backend, prometheusRegistry, "ledger_exporter")
 
-	var ledgerRange ledgerbackend.Range
-	if config.EndLedger == 0 {
-		ledgerRange = ledgerbackend.UnboundedRange(config.StartLedger)
-	} else {
-		ledgerRange = ledgerbackend.BoundedRange(config.StartLedger, config.EndLedger)
-	}
-
-	if err = backend.PrepareRange(ctx, ledgerRange); err != nil {
-		return nil, errors.Wrap(err, "Could not prepare captive core ledger backend")
-	}
 	return backend, nil
 }
diff --git a/exp/services/ledgerexporter/internal/exportmanager.go b/exp/services/ledgerexporter/internal/exportmanager.go
index 5e6c87bb..55f85b9c 100644
--- a/exp/services/ledgerexporter/internal/exportmanager.go
+++ b/exp/services/ledgerexporter/internal/exportmanager.go
@@ -89,21 +89,25 @@ func (e *ExportManager) Run(ctx context.Context, startLedger, endLedger uint32)
 		"end_ledger":   strconv.FormatUint(uint64(endLedger), 10),
 	}
 
+	var ledgerRange ledgerbackend.Range
+	if endLedger < 1 {
+		ledgerRange = ledgerbackend.UnboundedRange(startLedger)
+	} else {
+		ledgerRange = ledgerbackend.BoundedRange(startLedger, endLedger)
+	}
+	if err := e.ledgerBackend.PrepareRange(ctx, ledgerRange); err != nil {
+		return errors.Wrap(err, "Could not prepare captive core ledger backend")
+	}
+
 	for nextLedger := startLedger; endLedger < 1 || nextLedger <= endLedger; nextLedger++ {
-		select {
-		case <-ctx.Done():
-			logger.Info("Stopping ExportManager due to context cancellation")
-			return ctx.Err()
-		default:
-			ledgerCloseMeta, err := e.ledgerBackend.GetLedger(ctx, nextLedger)
-			if err != nil {
-				return errors.Wrapf(err, "failed to retrieve ledger %d from the ledger backend", nextLedger)
-			}
-			e.latestLedgerMetric.With(labels).Set(float64(nextLedger))
-			err = e.AddLedgerCloseMeta(ctx, ledgerCloseMeta)
-			if err != nil {
-				return errors.Wrapf(err, "failed to add ledgerCloseMeta for ledger %d", nextLedger)
-			}
+		ledgerCloseMeta, err := e.ledgerBackend.GetLedger(ctx, nextLedger)
+		if err != nil {
+			return errors.Wrapf(err, "failed to retrieve ledger %d from the ledger backend", nextLedger)
+		}
+		e.latestLedgerMetric.With(labels).Set(float64(nextLedger))
+		err = e.AddLedgerCloseMeta(ctx, ledgerCloseMeta)
+		if err != nil {
+			return errors.Wrapf(err, "failed to add ledgerCloseMeta for ledger %d", nextLedger)
 		}
 	}
 	logger.Infof("ExportManager successfully exported ledgers from %d to %d", startLedger, endLedger)


Test patch:
diff --git a/exp/services/ledgerexporter/internal/exportmanager_test.go b/exp/services/ledgerexporter/internal/exportmanager_test.go
index a84e20fb..d99f88cd 100644
--- a/exp/services/ledgerexporter/internal/exportmanager_test.go
+++ b/exp/services/ledgerexporter/internal/exportmanager_test.go
@@ -9,7 +9,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/mock"
 
-	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
 	"github.com/stellar/go/ingest/ledgerbackend"
@@ -61,7 +60,7 @@ func (s *ExportManagerSuite) TestInvalidExportConfig() {
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	_, err := NewExportManager(config, &s.mockBackend, queue, registry)
-	require.Error(s.T(), err)
+	s.Require().Error(err)
 }
 
 func (s *ExportManagerSuite) TestRun() {
@@ -69,11 +68,12 @@ func (s *ExportManagerSuite) TestRun() {
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
-	require.NoError(s.T(), err)
+	s.Require().NoError(err)
 
 	start := uint32(0)
 	end := uint32(255)
 	expectedKeys := set.NewSet[string](10)
+	s.mockBackend.On("PrepareRange", s.ctx, ledgerbackend.BoundedRange(start, end)).Return(nil)
 	for i := start; i <= end; i++ {
 		s.mockBackend.On("GetLedger", s.ctx, i).
 			Return(createLedgerCloseMeta(i), nil)
@@ -97,13 +97,12 @@ func (s *ExportManagerSuite) TestRun() {
 	}()
 
 	err = exporter.Run(s.ctx, start, end)
-	require.NoError(s.T(), err)
+	s.Require().NoError(err)
 
 	wg.Wait()
 
-	require.Equal(s.T(), expectedKeys, actualKeys)
-	require.Equal(
-		s.T(),
+	s.Require().Equal(expectedKeys, actualKeys)
+	s.Require().Equal(
 		float64(255),
 		getMetricValue(exporter.latestLedgerMetric.With(
 			prometheus.Labels{
@@ -119,9 +118,10 @@ func (s *ExportManagerSuite) TestRunContextCancel() {
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
-	require.NoError(s.T(), err)
+	s.Require().NoError(err)
 	ctx, cancel := context.WithCancel(context.Background())
 
+	s.mockBackend.On("PrepareRange", ctx, ledgerbackend.BoundedRange(0, 255)).Return(nil)
 	s.mockBackend.On("GetLedger", mock.Anything, mock.Anything).
 		Return(createLedgerCloseMeta(1), nil)
 
@@ -139,7 +139,7 @@ func (s *ExportManagerSuite) TestRunContextCancel() {
 	}()
 
 	err = exporter.Run(ctx, 0, 255)
-	require.EqualError(s.T(), err, "failed to add ledgerCloseMeta for ledger 128: context canceled")
+	s.Require().EqualError(err, "failed to add ledgerCloseMeta for ledger 128: context canceled")
 
 }
 
@@ -148,12 +148,18 @@ func (s *ExportManagerSuite) TestRunWithCanceledContext() {
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
-	require.NoError(s.T(), err)
+	s.Require().NoError(err)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
+
+	s.mockBackend.On("PrepareRange", ctx, ledgerbackend.BoundedRange(1, 10)).
+		Return(context.Canceled).Run(func(args mock.Arguments) {
+		ctx := args.Get(0).(context.Context)
+		s.Require().ErrorIs(ctx.Err(), context.Canceled)
+	})
 	err = exporter.Run(ctx, 1, 10)
-	require.EqualError(s.T(), err, "context canceled")
+	s.Require().ErrorIs(err, context.Canceled)
 }
 
 func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
@@ -161,7 +167,7 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
-	require.NoError(s.T(), err)
+	s.Require().NoError(err)
 
 	expectedKeys := set.NewSet[string](10)
 	actualKeys := set.NewSet[string](10)
@@ -183,7 +189,7 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
 	start := uint32(0)
 	end := uint32(255)
 	for i := start; i <= end; i++ {
-		require.NoError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(i)))
+		s.Require().NoError(exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(i)))
 
 		key := config.GetObjectKeyFromSequenceNumber(i)
 		expectedKeys.Add(key)
@@ -191,7 +197,7 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
 
 	queue.Close()
 	wg.Wait()
-	require.Equal(s.T(), expectedKeys, actualKeys)
+	s.Require().Equal(expectedKeys, actualKeys)
 }
 
 func (s *ExportManagerSuite) TestAddLedgerCloseMetaContextCancel() {
@@ -199,7 +205,7 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMetaContextCancel() {
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
-	require.NoError(s.T(), err)
+	s.Require().NoError(err)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	go func() {
@@ -207,9 +213,9 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMetaContextCancel() {
 		cancel()
 	}()
 
-	require.NoError(s.T(), exporter.AddLedgerCloseMeta(ctx, createLedgerCloseMeta(1)))
+	s.Require().NoError(exporter.AddLedgerCloseMeta(ctx, createLedgerCloseMeta(1)))
 	err = exporter.AddLedgerCloseMeta(ctx, createLedgerCloseMeta(2))
-	require.EqualError(s.T(), err, "context canceled")
+	s.Require().EqualError(err, "context canceled")
 }
 
 func (s *ExportManagerSuite) TestAddLedgerCloseMetaKeyMismatch() {
@@ -217,9 +223,9 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMetaKeyMismatch() {
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
-	require.NoError(s.T(), err)
+	s.Require().NoError(err)
 
-	require.NoError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(16)))
-	require.EqualError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(21)),
+	s.Require().NoError(exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(16)))
+	s.Require().EqualError(exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(21)),
 		"Current meta archive object key mismatch")
 }