Merge branch 'master' into friendbot-instrumentation-03

Author: satyamz

.github/workflows/horizon.yml

@@ -112,7 +112,7 @@ jobs:
           key: ${{ env.COMBINED_SOURCE_HASH }}
 
       - if: ${{ steps.horizon_binary_tests_hash.outputs.cache-hit != 'true' }}
-        run: go test -race -timeout 65m -v ./services/horizon/internal/integration/...
+        run: go test -race -timeout 75m -v ./services/horizon/internal/integration/...
 
       - name: Save Horizon binary and integration tests source hash to cache
         if: ${{ success() && steps.horizon_binary_tests_hash.outputs.cache-hit != 'true' }}

ingest/CHANGELOG.md

@@ -10,6 +10,7 @@ All notable changes to this project will be documented in this file. This projec
   - `DefaultBufferedStorageBackendConfig`
   - `PublisherConfig` 
   These are now relocated to `ingest` package. Will need to change references in existing code.
+* Moved the `ingest/verify` package to be internally located at `services/horizon/internal/ingest` package in `verify.go` for overall go repo restructuring. [5670](https://github.com/stellar/go/issues/5670)  
 
 
 ## v23.0.0

ingest/loadtest/ledger_backend.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"math"
 	"os"
+	"sync"
 	"time"
 
 	"github.com/klauspost/compress/zstd"
@@ -16,6 +17,9 @@ import (
 	"github.com/stellar/go/xdr"
 )
 
+// ErrLoadTestDone indicates that the load test has run to completion.
+var ErrLoadTestDone = fmt.Errorf("the load test is done")
+
 // LedgerBackend is used to load test ingestion.
 // LedgerBackend will take a file of synthetically generated ledgers (see
 // services/horizon/internal/integration/generate_ledgers_test.go) and merge those ledgers
@@ -31,6 +35,8 @@ type LedgerBackend struct {
 	latestLedgerSeq       uint32
 	preparedRange         ledgerbackend.Range
 	cachedLedger          xdr.LedgerCloseMeta
+	done                  bool
+	lock                  sync.RWMutex
 }
 
 // LedgerBackendConfig configures LedgerBackend
@@ -57,6 +63,9 @@ func NewLedgerBackend(config LedgerBackendConfig) *LedgerBackend {
 }
 
 func (r *LedgerBackend) GetLatestLedgerSequence(ctx context.Context) (uint32, error) {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+
 	if r.nextLedgerSeq == 0 {
 		return 0, fmt.Errorf("PrepareRange() must be called before GetLatestLedgerSequence()")
 	}
@@ -94,6 +103,12 @@ func readLedgerEntries(path string) ([]xdr.LedgerEntry, error) {
 }
 
 func (r *LedgerBackend) PrepareRange(ctx context.Context, ledgerRange ledgerbackend.Range) error {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	if r.done {
+		return ErrLoadTestDone
+	}
 	if r.nextLedgerSeq != 0 {
 		if r.isPrepared(ledgerRange) {
 			return nil
@@ -282,6 +297,9 @@ func validateNetworkPassphrase(networkPassphrase string, ledger xdr.LedgerCloseM
 }
 
 func (r *LedgerBackend) IsPrepared(ctx context.Context, ledgerRange ledgerbackend.Range) (bool, error) {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+
 	return r.isPrepared(ledgerRange), nil
 }
 
@@ -302,6 +320,15 @@ func (r *LedgerBackend) isPrepared(ledgerRange ledgerbackend.Range) bool {
 }
 
 func (r *LedgerBackend) GetLedger(ctx context.Context, sequence uint32) (xdr.LedgerCloseMeta, error) {
+	r.lock.RLock()
+	closeLedgerBackend := false
+	defer func() {
+		r.lock.RUnlock()
+		if closeLedgerBackend {
+			r.Close()
+		}
+	}()
+
 	if r.nextLedgerSeq == 0 {
 		return xdr.LedgerCloseMeta{}, fmt.Errorf("PrepareRange() must be called before GetLedger()")
 	}
@@ -312,6 +339,13 @@ func (r *LedgerBackend) GetLedger(ctx context.Context, sequence uint32) (xdr.Led
 			r.cachedLedger.LedgerSequence(),
 		)
 	}
+	if r.done {
+		return xdr.LedgerCloseMeta{}, ErrLoadTestDone
+	}
+	if sequence > r.latestLedgerSeq {
+		closeLedgerBackend = true
+		return xdr.LedgerCloseMeta{}, ErrLoadTestDone
+	}
 	for ; r.nextLedgerSeq <= sequence; r.nextLedgerSeq++ {
 		var ledger xdr.LedgerCloseMeta
 		if err := r.mergedLedgersStream.ReadOne(&ledger); err == io.EOF {
@@ -339,6 +373,10 @@ func (r *LedgerBackend) GetLedger(ctx context.Context, sequence uint32) (xdr.Led
 }
 
 func (r *LedgerBackend) Close() error {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	r.done = true
 	if err := r.config.LedgerBackend.Close(); err != nil {
 		return fmt.Errorf("could not close real ledger backend: %w", err)
 	}
@@ -347,9 +385,13 @@ func (r *LedgerBackend) Close() error {
 		if err := r.mergedLedgersStream.Close(); err != nil {
 			return fmt.Errorf("could not close merged ledgers xdr stream: %w", err)
 		}
+		r.mergedLedgersStream = nil
+	}
+	if r.mergedLedgersFilePath != "" {
 		if err := os.Remove(r.mergedLedgersFilePath); err != nil {
 			return fmt.Errorf("could not remove merged ledgers file: %w", err)
 		}
+		r.mergedLedgersFilePath = ""
 	}
 	return nil
 }