horizon/integration/test: added test for reingest with --ledgerbackend datastore

go.mod

@@ -63,6 +63,7 @@ require (
  github.com/docker/docker v27.0.3+incompatible
  github.com/docker/go-connections v0.5.0
  github.com/fsouza/fake-gcs-server v1.49.2
+ github.com/otiai10/copy v1.14.0
 )
 
 require (

go.sum

@@ -401,6 +401,10 @@ github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQ
 github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM=
 github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU=
 github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
+github.com/otiai10/copy v1.14.0 h1:dCI/t1iTdYGtkvCuBG2BgR6KZa83PTclw4U5n2wAllU=
+github.com/otiai10/copy v1.14.0/go.mod h1:ECfuL02W+/FkTWZWgQqXPWZgW9oeKCSQ5qVfSc4qc4w=
+github.com/otiai10/mint v1.5.1 h1:XaPLeE+9vGbuyEHem1JNk3bYc7KKqyI/na0/mLd/Kks=
+github.com/otiai10/mint v1.5.1/go.mod h1:MJm72SBthJjz8qhefc4z1PYEieWmy8Bku7CjcAqyUSM=
 github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3ve8=
 github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c=
 github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=

services/horizon/cmd/db.go

@@ -101,6 +101,7 @@ func migrate(dir schema.MigrateDir, count int) error {
  if err != nil {
  return err
  }
+ defer dbConn.Close()
 
  numMigrationsRun, err := schema.Migrate(dbConn.DB.DB, dir, count)
  if err != nil {

services/horizon/cmd/db_test.go

@@ -18,7 +18,7 @@ func TestDBCommandsTestSuite(t *testing.T) {
 
 type DBCommandsTestSuite struct {
  suite.Suite
- dsn string
+ db *dbtest.DB
 }
 
 func (s *DBCommandsTestSuite) SetupSuite() {
@@ -27,18 +27,21 @@ func (s *DBCommandsTestSuite) SetupSuite() {
  return nil
  }
 
- newDB := dbtest.Postgres(s.T())
- s.dsn = newDB.DSN
+ s.db = dbtest.Postgres(s.T())
 
  RootCmd.SetArgs([]string{
- "db", "migrate", "up", "--db-url", s.dsn})
+ "db", "migrate", "up", "--db-url", s.db.DSN})
  require.NoError(s.T(), RootCmd.Execute())
 }
 
+func (s *DBCommandsTestSuite) TearDownSuite() {
+ s.db.Close()
+}
+
 func (s *DBCommandsTestSuite) TestDefaultParallelJobSizeForBufferedBackend() {
  RootCmd.SetArgs([]string{
  "db", "reingest", "range",
- "--db-url", s.dsn,
+ "--db-url", s.db.DSN,
  "--network", "testnet",
  "--parallel-workers", "2",
  "--ledgerbackend", "datastore",
@@ -53,7 +56,7 @@ func (s *DBCommandsTestSuite) TestDefaultParallelJobSizeForBufferedBackend() {
 func (s *DBCommandsTestSuite) TestDefaultParallelJobSizeForCaptiveBackend() {
  RootCmd.SetArgs([]string{
  "db", "reingest", "range",
- "--db-url", s.dsn,
+ "--db-url", s.db.DSN,
  "--network", "testnet",
  "--stellar-core-binary-path", "/test/core/bin/path",
  "--parallel-workers", "2",
@@ -68,7 +71,7 @@ func (s *DBCommandsTestSuite) TestDefaultParallelJobSizeForCaptiveBackend() {
 func (s *DBCommandsTestSuite) TestUsesParallelJobSizeWhenSetForCaptive() {
  RootCmd.SetArgs([]string{
  "db", "reingest", "range",
- "--db-url", s.dsn,
+ "--db-url", s.db.DSN,
  "--network", "testnet",
  "--stellar-core-binary-path", "/test/core/bin/path",
  "--parallel-workers", "2",
@@ -84,7 +87,7 @@ func (s *DBCommandsTestSuite) TestUsesParallelJobSizeWhenSetForCaptive() {
 func (s *DBCommandsTestSuite) TestUsesParallelJobSizeWhenSetForBuffered() {
  RootCmd.SetArgs([]string{
  "db", "reingest", "range",
- "--db-url", s.dsn,
+ "--db-url", s.db.DSN,
  "--network", "testnet",
  "--parallel-workers", "2",
  "--parallel-job-size", "5",

services/horizon/internal/integration/db_test.go

@@ -3,16 +3,24 @@ package integration
 import (
  "context"
  "fmt"
+ "net"
+ "os"
  "path/filepath"
  "strconv"
  "testing"
  "time"
 
+ "github.com/fsouza/fake-gcs-server/fakestorage"
+ cp "github.com/otiai10/copy"
+ "github.com/sirupsen/logrus"
  "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
 
  "github.com/stellar/go/clients/horizonclient"
+ sdk "github.com/stellar/go/clients/horizonclient"
  "github.com/stellar/go/historyarchive"
  "github.com/stellar/go/keypair"
+ "github.com/stellar/go/network"
  hProtocol "github.com/stellar/go/protocols/horizon"
  horizoncmd "github.com/stellar/go/services/horizon/cmd"
  horizon "github.com/stellar/go/services/horizon/internal"
@@ -25,6 +33,7 @@ import (
  "github.com/stellar/go/support/db/dbtest"
  "github.com/stellar/go/txnbuild"
  "github.com/stellar/go/xdr"
+ "github.com/stellar/throttled"
 )
 
 func submitLiquidityPoolOps(itest *integration.Test, tt *assert.Assertions) (submittedOperations []txnbuild.Operation, lastLedger int32) {
@@ -549,6 +558,131 @@ func TestReingestDB(t *testing.T) {
  tt.NoError(horizoncmd.RootCmd.Execute(), "Repeat the same reingest range against db, should not have errors.")
 }
 
+func TestReingestDatastore(t *testing.T) {
+ if os.Getenv("HORIZON_INTEGRATION_TESTS_ENABLED") == "" {
+ t.Skip("skipping integration test: HORIZON_INTEGRATION_TESTS_ENABLED not set")
+ }
+
+ newDB := dbtest.Postgres(t)
+ defer newDB.Close()
+ horizoncmd.RootCmd.SetArgs([]string{
+ "db", "migrate", "up", "--db-url", newDB.DSN})
+ require.NoError(t, horizoncmd.RootCmd.Execute())
+
+ testTempDir := t.TempDir()
+ tempSeedDataPath := filepath.Join(testTempDir, "data")
+ tempSeedBucketPath := filepath.Join(tempSeedDataPath, "path", "to", "my", "bucket")
+ tempSeedBucketFolder := filepath.Join(tempSeedBucketPath, "FFFFFFFF--0-63999")
+ if err := os.MkdirAll(tempSeedBucketFolder, 0777); err != nil {
+ t.Fatalf("unable to create seed data in temp path, %v", err)
+ }
+
+ err := cp.Copy("./testdata/testbucket", tempSeedBucketFolder)
+ if err != nil {
+ t.Fatalf("unable to copy seed data files for fake gcs, %v", err)
+ }
+
+ testWriter := &testWriter{test: t}
+ opts := fakestorage.Options{
+ Scheme: "http",
+ Host: "127.0.0.1",
+ Port: uint16(0),
+ Writer: testWriter,
+ Seed: tempSeedDataPath,
+ StorageRoot: filepath.Join(testTempDir, "bucket"),
+ PublicHost: "127.0.0.1",
+ }
+
+ gcsServer, err := fakestorage.NewServerWithOptions(opts)
+
+ if err != nil {
+ t.Fatalf("couldn't start the fake gcs http server %v", err)
+ }
+
+ defer gcsServer.Stop()
+ t.Logf("fake gcs server started at %v", gcsServer.URL())
+ t.Setenv("STORAGE_EMULATOR_HOST", gcsServer.URL())
+
+ horizoncmd.RootCmd.SetArgs([]string{"db",
+ "reingest",
+ "range",
+ "--db-url", newDB.DSN,
+ "--network", "testnet",
+ "--parallel-workers", "1",
+ "--ledgerbackend", "datastore",
+ "--datastore-config", "../../config.storagebackend.toml",
+ "997",
+ "999"})
+
+ require.NoError(t, horizoncmd.RootCmd.Execute())
+
+ listener, webApp, webPort, err := dynamicHorizonWeb(newDB.DSN)
+ if err != nil {
+ t.Fatalf("couldn't create and start horizon web app on dynamic port %v", err)
+ }
+
+ webAppDone := make(chan struct{})
+ go func() {
+ defer close(webAppDone)
+ if err = listener.Close(); err != nil {
+ return
+ }
+ webApp.Serve()
+ }()
+
+ defer func() {
+ webApp.Close()
+ select {
+ case <-webAppDone:
+ return
+ default:
+ }
+ }()
+
+ horizonClient := &sdk.Client{
+ HorizonURL: fmt.Sprintf("http:https://localhost:%v", webPort),
+ }
+
+ // wait until the web server is up before continuing to test requests
+ require.Eventually(t, func() bool {
+ if _, horizonErr := horizonClient.Root(); horizonErr != nil {
+ return false
+ }
+ return true
+ }, time.Second*15, time.Millisecond*100)
+
+ _, err = horizonClient.LedgerDetail(998)
+ require.NoError(t, err)
+}
+
+func dynamicHorizonWeb(dsn string) (net.Listener, *horizon.App, uint, error) {
+ listener, err := net.Listen("tcp", "localhost:0")
+ if err != nil {
+ return nil, nil, 0, err
+ }
+ webPort := uint(listener.Addr().(*net.TCPAddr).Port)
+
+ webApp, err := horizon.NewApp(horizon.Config{
+ DatabaseURL: dsn,
+ Port: webPort,
+ NetworkPassphrase: network.TestNetworkPassphrase,
+ LogLevel: logrus.InfoLevel,
+ DisableTxSub: true,
+ Ingest: false,
+ ConnectionTimeout: 10 * time.Second,
+ RateQuota: &throttled.RateQuota{
+ MaxRate: throttled.PerHour(1000),
+ MaxBurst: 100,
+ },
+ })
+ if err != nil {
+ listener.Close()
+ return nil, nil, 0, err
+ }
+
+ return listener, webApp, webPort, nil
+}
+
 func TestReingestDBWithFilterRules(t *testing.T) {
  itest, _ := initializeDBIntegrationTest(t)
  tt := assert.New(t)
@@ -905,3 +1039,12 @@ func TestResumeFromInitializedDB(t *testing.T) {
 
  tt.Eventually(successfullyResumed, 1*time.Minute, 1*time.Second)
 }
+
+type testWriter struct {
+ test *testing.T
+}
+
+func (w *testWriter) Write(p []byte) (n int, err error) {
+ w.test.Log(string(p))
+ return len(p), nil
+}