services/horizon/internal/integration: Fix horizon integration tests (#…

…4734)

* Fix flaky TestReingestDB

* Fix reuse of captive core storage path

* Increase timeout for integration tests

.github/workflows/horizon.yml

@@ -104,7 +104,7 @@ jobs:
         key: horizon-hash-${{ hashFiles('./horizon') }}-${{ hashFiles('./clients/horizonclient/**') }}-${{ hashFiles('./protocols/horizon/**') }}-${{ hashFiles('./txnbuild/**') }}-${{ hashFiles('./services/horizon/internal/integration/**') }}-${{ env.PROTOCOL_19_CORE_DOCKER_IMG }}-${{ env.PROTOCOL_18_CORE_DOCKER_IMG }}
 
     - if: ${{ steps.horizon_binary_tests_hash.outputs.cache-hit != 'true' }}
-      run: go test -race -timeout 25m -v ./services/horizon/internal/integration/...
+      run: go test -race -timeout 35m -v ./services/horizon/internal/integration/...
 
   verify-range:
     name: Test (and push) verify-range image

integration.sh

@@ -20,4 +20,4 @@ if [[ "$(docker inspect integration_postgres -f '{{.State.Running}}')" != "true"
     circleci/postgres:9.6.5-alpine
 fi
 
-exec go test -timeout 25m github.com/stellar/go/services/horizon/internal/integration/... "$@"
+exec go test -timeout 35m github.com/stellar/go/services/horizon/internal/integration/... "$@"

services/horizon/internal/docs/notes_for_developers.md

@@ -70,12 +70,12 @@ go test github.com/stellar/go/services/horizon/...
 To run the integration tests, move to top folder of working copy of `go` repo to run all integration tests 
 or /services/horizon to run just Horizon integration tests:
 ```
-HORIZON_INTEGRATION_TESTS_ENABLED=true go test -race -timeout 25m -v ./...
+HORIZON_INTEGRATION_TESTS_ENABLED=true go test -race -timeout 35m -v ./...
 ```
 
 To run just one specific integration test, e.g. like `TestTxSub`:
 ```
-HORIZON_INTEGRATION_TESTS_ENABLED=true go test -run TestTxsub -race -timeout 25m -v ./...
+HORIZON_INTEGRATION_TESTS_ENABLED=true go test -run TestTxsub -race -timeout 5m -v ./...
 ```
 
 ## <a name="logging"></a> Logging

services/horizon/internal/integration/db_test.go

@@ -3,7 +3,6 @@ package integration
 import (
 	"context"
 	"fmt"
-	"os"
 	"path/filepath"
 	"strconv"
 	"testing"
@@ -396,46 +395,53 @@ func submitAccountOps(itest *integration.Test, tt *assert.Assertions) (submitted
 	return allOps, txResp.Ledger
 }
 
-func initializeDBIntegrationTest(t *testing.T) (itest *integration.Test, reachedLedger int32) {
-	itest = integration.NewTest(t, integration.Config{})
+func initializeDBIntegrationTest(t *testing.T) (*integration.Test, int32) {
+	itest := integration.NewTest(t, integration.Config{})
 	tt := assert.New(t)
 
-	// submit all possible operations
-	ops, _ := submitAccountOps(itest, tt)
-	submittedOps := ops
-	ops, _ = submitPaymentOps(itest, tt)
-	submittedOps = append(submittedOps, ops...)
-	ops, _ = submitOfferAndTrustlineOps(itest, tt)
-	submittedOps = append(submittedOps, ops...)
-	ops, _ = submitSponsorshipOps(itest, tt)
-	submittedOps = append(submittedOps, ops...)
-	ops, _ = submitClaimableBalanceOps(itest, tt)
-	submittedOps = append(submittedOps, ops...)
-	ops, _ = submitClawbackOps(itest, tt)
-	submittedOps = append(submittedOps, ops...)
-	ops, reachedLedger = submitLiquidityPoolOps(itest, tt)
-	submittedOps = append(submittedOps, ops...)
-
 	// Make sure all possible operations are covered by reingestion
 	allOpTypes := set.Set[xdr.OperationType]{}
 	for typ := range xdr.OperationTypeToStringMap {
 		allOpTypes.Add(xdr.OperationType(typ))
 	}
+
+	submitters := []func(*integration.Test, *assert.Assertions) ([]txnbuild.Operation, int32){
+		submitAccountOps,
+		submitPaymentOps,
+		submitOfferAndTrustlineOps,
+		submitSponsorshipOps,
+		submitClaimableBalanceOps,
+		submitClawbackOps,
+		submitLiquidityPoolOps,
+	}
 	// Inflation is not supported
 	delete(allOpTypes, xdr.OperationTypeInflation)
 
+	var submittedOps []txnbuild.Operation
+	var ledgerOfLastSubmittedTx int32
+	// submit all possible operations
+	for i, f := range submitters {
+		var ops []txnbuild.Operation
+		ops, ledgerOfLastSubmittedTx = f(itest, tt)
+		t.Logf("%v ledgerOfLastSubmittedTx %v", i, ledgerOfLastSubmittedTx)
+		submittedOps = append(submittedOps, ops...)
+	}
+
 	for _, op := range submittedOps {
 		opXDR, err := op.BuildXDR()
 		tt.NoError(err)
 		delete(allOpTypes, opXDR.Body.Type)
 	}
 	tt.Empty(allOpTypes)
 
-	root, err := itest.Client().Root()
-	tt.NoError(err)
-	tt.LessOrEqual(reachedLedger, root.HorizonSequence)
+	reachedLedger := func() bool {
+		root, err := itest.Client().Root()
+		tt.NoError(err)
+		return root.HorizonSequence >= ledgerOfLastSubmittedTx
+	}
+	tt.Eventually(reachedLedger, 15*time.Second, 5*time.Second)
 
-	return
+	return itest, ledgerOfLastSubmittedTx
 }
 
 func TestReingestDB(t *testing.T) {
@@ -444,7 +450,7 @@ func TestReingestDB(t *testing.T) {
 
 	horizonConfig := itest.GetHorizonIngestConfig()
 	t.Run("validate parallel range", func(t *testing.T) {
-		horizoncmd.RootCmd.SetArgs(command(horizonConfig,
+		horizoncmd.RootCmd.SetArgs(command(t, horizonConfig,
 			"db",
 			"reingest",
 			"range",
@@ -456,6 +462,7 @@ func TestReingestDB(t *testing.T) {
 		assert.EqualError(t, horizoncmd.RootCmd.Execute(), "Invalid range: {10 2} from > to")
 	})
 
+	t.Logf("reached ledger is %v", reachedLedger)
 	// cap reachedLedger to the nearest checkpoint ledger because reingest range cannot ingest past the most
 	// recent checkpoint ledger when using captive core
 	toLedger := uint32(reachedLedger)
@@ -469,7 +476,10 @@ func TestReingestDB(t *testing.T) {
 	var latestCheckpoint uint32
 	publishedFirstCheckpoint := func() bool {
 		has, requestErr := archive.GetRootHAS()
-		tt.NoError(requestErr)
+		if requestErr != nil {
+			t.Logf("request to fetch checkpoint failed: %v", requestErr)
+			return false
+		}
 		latestCheckpoint = has.CurrentLedger
 		return latestCheckpoint > 1
 	}
@@ -489,7 +499,7 @@ func TestReingestDB(t *testing.T) {
 		"captive-core-reingest-range-integration-tests.cfg",
 	)
 
-	horizoncmd.RootCmd.SetArgs(command(horizonConfig, "db",
+	horizoncmd.RootCmd.SetArgs(command(t, horizonConfig, "db",
 		"reingest",
 		"range",
 		"--parallel-workers=1",
@@ -501,7 +511,7 @@ func TestReingestDB(t *testing.T) {
 	tt.NoError(horizoncmd.RootCmd.Execute(), "Repeat the same reingest range against db, should not have errors.")
 }
 
-func command(horizonConfig horizon.Config, args ...string) []string {
+func command(t *testing.T, horizonConfig horizon.Config, args ...string) []string {
 	return append([]string{
 		"--stellar-core-url",
 		horizonConfig.StellarCoreURL,
@@ -525,7 +535,7 @@ func command(horizonConfig horizon.Config, args ...string) []string {
 		"8",
 		// Create the storage directory outside of the source repo,
 		// otherwise it will break Golang test caching.
-		"--captive-core-storage-path=" + os.TempDir(),
+		"--captive-core-storage-path=" + t.TempDir(),
 	}, args...)
 }
 
@@ -602,7 +612,7 @@ func TestFillGaps(t *testing.T) {
 	tt.NoError(err)
 
 	t.Run("validate parallel range", func(t *testing.T) {
-		horizoncmd.RootCmd.SetArgs(command(horizonConfig,
+		horizoncmd.RootCmd.SetArgs(command(t, horizonConfig,
 			"db",
 			"fill-gaps",
 			"--parallel-workers=2",
@@ -641,20 +651,20 @@ func TestFillGaps(t *testing.T) {
 		filepath.Dir(horizonConfig.CaptiveCoreConfigPath),
 		"captive-core-reingest-range-integration-tests.cfg",
 	)
-	horizoncmd.RootCmd.SetArgs(command(horizonConfig, "db", "fill-gaps", "--parallel-workers=1"))
+	horizoncmd.RootCmd.SetArgs(command(t, horizonConfig, "db", "fill-gaps", "--parallel-workers=1"))
 	tt.NoError(horizoncmd.RootCmd.Execute())
 
 	tt.NoError(historyQ.LatestLedger(context.Background(), &latestLedger))
 	tt.Equal(int64(0), latestLedger)
 
-	horizoncmd.RootCmd.SetArgs(command(horizonConfig, "db", "fill-gaps", "3", "4"))
+	horizoncmd.RootCmd.SetArgs(command(t, horizonConfig, "db", "fill-gaps", "3", "4"))
 	tt.NoError(horizoncmd.RootCmd.Execute())
 	tt.NoError(historyQ.LatestLedger(context.Background(), &latestLedger))
 	tt.NoError(historyQ.ElderLedger(context.Background(), &oldestLedger))
 	tt.Equal(int64(3), oldestLedger)
 	tt.Equal(int64(4), latestLedger)
 
-	horizoncmd.RootCmd.SetArgs(command(horizonConfig, "db", "fill-gaps", "6", "7"))
+	horizoncmd.RootCmd.SetArgs(command(t, horizonConfig, "db", "fill-gaps", "6", "7"))
 	tt.NoError(horizoncmd.RootCmd.Execute())
 	tt.NoError(historyQ.LatestLedger(context.Background(), &latestLedger))
 	tt.NoError(historyQ.ElderLedger(context.Background(), &oldestLedger))
@@ -665,7 +675,7 @@ func TestFillGaps(t *testing.T) {
 	tt.NoError(err)
 	tt.Equal([]history.LedgerRange{{StartSequence: 5, EndSequence: 5}}, gaps)
 
-	horizoncmd.RootCmd.SetArgs(command(horizonConfig, "db", "fill-gaps"))
+	horizoncmd.RootCmd.SetArgs(command(t, horizonConfig, "db", "fill-gaps"))
 	tt.NoError(horizoncmd.RootCmd.Execute())
 	tt.NoError(historyQ.LatestLedger(context.Background(), &latestLedger))
 	tt.NoError(historyQ.ElderLedger(context.Background(), &oldestLedger))
@@ -675,7 +685,7 @@ func TestFillGaps(t *testing.T) {
 	tt.NoError(err)
 	tt.Empty(gaps)
 
-	horizoncmd.RootCmd.SetArgs(command(horizonConfig, "db", "fill-gaps", "2", "8"))
+	horizoncmd.RootCmd.SetArgs(command(t, horizonConfig, "db", "fill-gaps", "2", "8"))
 	tt.NoError(horizoncmd.RootCmd.Execute())
 	tt.NoError(historyQ.LatestLedger(context.Background(), &latestLedger))
 	tt.NoError(historyQ.ElderLedger(context.Background(), &oldestLedger))

services/horizon/internal/test/integration/integration.go

@@ -69,9 +69,10 @@ type Config struct {
 }
 
 type CaptiveConfig struct {
-	binaryPath string
-	configPath string
-	useDB      bool
+	binaryPath  string
+	configPath  string
+	storagePath string
+	useDB       bool
 }
 
 type Test struct {
@@ -169,6 +170,7 @@ func (i *Test) configureCaptiveCore() {
 		composePath := findDockerComposePath()
 		i.coreConfig.binaryPath = os.Getenv("HORIZON_INTEGRATION_TESTS_CAPTIVE_CORE_BIN")
 		i.coreConfig.configPath = filepath.Join(composePath, "captive-core-integration-tests.cfg")
+		i.coreConfig.storagePath = i.CurrentTest().TempDir()
 		if RunWithCaptiveCoreUseDB {
 			i.coreConfig.useDB = true
 		}
@@ -332,6 +334,7 @@ func (i *Test) StartHorizon() error {
 	coreBinaryPath := i.coreConfig.binaryPath
 	captiveCoreConfigPath := i.coreConfig.configPath
 	captiveCoreUseDB := strconv.FormatBool(i.coreConfig.useDB)
+	captiveCoreStoragePath := i.coreConfig.storagePath
 
 	defaultArgs := map[string]string{
 		"ingest":                        "false",
@@ -366,7 +369,7 @@ func (i *Test) StartHorizon() error {
 			"captive-core-config-path":  captiveCoreConfigPath,
 			"captive-core-http-port":    "21626",
 			"captive-core-use-db":       captiveCoreUseDB,
-			"captive-core-storage-path": os.TempDir(),
+			"captive-core-storage-path": captiveCoreStoragePath,
 			"ingest":                    "true"},
 		i.config.HorizonIngestParameters)
 	ingestArgs := mapToFlags(mergedIngest)