getLedgerEntries: optionally use high-performance Core server

.github/workflows/stellar-rpc.yml

@@ -98,16 +98,16 @@ jobs:
     strategy:
       matrix:
         os: [ ubuntu-20.04, ubuntu-22.04 ]
-        protocol-version: [ 21, 22 ]
+        protocol-version: [ 22, 23 ]
     runs-on: ${{ matrix.os }}
     env:
       STELLAR_RPC_INTEGRATION_TESTS_ENABLED: true
       STELLAR_RPC_INTEGRATION_TESTS_CORE_MAX_SUPPORTED_PROTOCOL: ${{ matrix.protocol-version }}
       STELLAR_RPC_INTEGRATION_TESTS_CAPTIVE_CORE_BIN: /usr/bin/stellar-core
-      PROTOCOL_21_CORE_DEBIAN_PKG_VERSION: 22.0.0-2138.721fd0a65.focal
-      PROTOCOL_21_CORE_DOCKER_IMG: stellar/stellar-core:22.0.0-2138.721fd0a65.focal
-      PROTOCOL_22_CORE_DEBIAN_PKG_VERSION: 22.0.0-2138.721fd0a65.focal
-      PROTOCOL_22_CORE_DOCKER_IMG: stellar/stellar-core:22.0.0-2138.721fd0a65.focal
+      PROTOCOL_22_CORE_DEBIAN_PKG_VERSION: 22.1.0-2194.0241e79f7.focal
+      PROTOCOL_22_CORE_DOCKER_IMG: stellar/stellar-core:22.1.0-2194.0241e79f7.focal
+      PROTOCOL_23_CORE_DEBIAN_PKG_VERSION: 22.1.1-2251.ac9f21ac7.focal-do-not-use-in-prd
+      PROTOCOL_23_CORE_DOCKER_IMG: stellar/unsafe-stellar-core:22.1.1-2251.ac9f21ac7.focal-do-not-use-in-prd
 
     steps:
       - uses: actions/checkout@v4

cmd/stellar-rpc/internal/config/main.go

@@ -14,11 +14,14 @@ type Config struct {
 
 	Strict bool
 
-	StellarCoreURL         string
-	CaptiveCoreStoragePath string
-	StellarCoreBinaryPath  string
-	CaptiveCoreConfigPath  string
-	CaptiveCoreHTTPPort    uint
+	StellarCoreURL                      string
+	CaptiveCoreStoragePath              string
+	StellarCoreBinaryPath               string
+	CaptiveCoreConfigPath               string
+	CaptiveCoreHTTPPort                 uint
+	CaptiveCoreHTTPQueryPort            uint
+	CaptiveCoreHTTPQueryThreadPoolSize  uint
+	CaptiveCoreHTTPQuerySnapshotLedgers uint
 
 	Endpoint                                       string
 	AdminEndpoint                                  string

cmd/stellar-rpc/internal/config/options.go

@@ -20,7 +20,8 @@ const (
 	OneDayOfLedgers   = 17280
 	SevenDayOfLedgers = OneDayOfLedgers * 7
 
-	defaultHTTPEndpoint = "localhost:8000"
+	defaultHTTPEndpoint        = "localhost:8000"
+	defaultCaptiveCoreHTTPPort = 11626 // regular queries like /info
 )
 
 // TODO: refactor and remove the linter exceptions
@@ -60,6 +61,7 @@ func (cfg *Config) options() Options {
 			Usage:     "Admin endpoint to listen and serve on. WARNING: this should not be accessible from the Internet and does not use TLS. \"\" (default) disables the admin server",
 			ConfigKey: &cfg.AdminEndpoint,
 		},
+		// TODO: should be gave a similar stellar-core-query-url parameter or should we assume that all queries will happen to the local captive core?
 		{
 			Name:      "stellar-core-url",
 			Usage:     "URL used to query Stellar Core (local captive core by default)",
@@ -84,7 +86,25 @@ func (cfg *Config) options() Options {
 			Name:         "stellar-captive-core-http-port",
 			Usage:        "HTTP port for Captive Core to listen on (0 disables the HTTP server)",
 			ConfigKey:    &cfg.CaptiveCoreHTTPPort,
-			DefaultValue: uint(11626),
+			DefaultValue: uint(defaultCaptiveCoreHTTPPort),
+		},
+		{
+			Name:         "stellar-captive-core-http-query-port",
+			Usage:        "HTTP port for Captive Core to listen on for high-performance queries like /getledgerentry (0 disables the HTTP server, must not conflict with CAPTIVE_CORE_HTTP_PORT)",
+			ConfigKey:    &cfg.CaptiveCoreHTTPQueryPort,
+			DefaultValue: uint(0), // Disabled by default, although it normally uses 11628
+		},
+		{
+			Name:         "stellar-captive-core-http-query-thread-pool-size",
+			Usage:        "Number of threads to use by Captive Core's high-performance query server",
+			ConfigKey:    &cfg.CaptiveCoreHTTPQueryThreadPoolSize,
+			DefaultValue: uint(runtime.NumCPU()), //nolint:gosec
+		},
+		{
+			Name:         "stellar-captive-core-http-query-snapshot-ledgers",
+			Usage:        "Size of ledger history in Captive Core's high-performance query server (don't touch unless you know what you are doing)",
+			ConfigKey:    &cfg.CaptiveCoreHTTPQuerySnapshotLedgers,
+			DefaultValue: uint(4),
 		},
 		{
 			Name:         "log-level",

cmd/stellar-rpc/internal/config/test.soroban.rpc.config

@@ -8,4 +8,5 @@ STELLAR_CORE_BINARY_PATH="/usr/bin/stellar-core"
 HISTORY_ARCHIVE_URLS=["http://localhost:1570"]
 DB_PATH="/opt/stellar/stellar-rpc/rpc_db.sqlite"
 STELLAR_CAPTIVE_CORE_HTTP_PORT=0
+STELLAR_CAPTIVE_CORE_HTTP_QUERY_PORT=11628
 CHECKPOINT_FREQUENCY=64

cmd/stellar-rpc/internal/daemon/daemon.go

@@ -3,6 +3,7 @@
 import (
 	"context"
 	"errors"
+	"fmt"
 	"net"
 	"net/http"
 	"net/http/pprof"
@@ -49,6 +50,7 @@
 type Daemon struct {
 	core                *ledgerbackend.CaptiveStellarCore
 	coreClient          *CoreClientWithMetrics
+	coreQueryingClient  interfaces.FastCoreClient
 	ingestService       *ingest.Service
 	db                  *db.DB
 	jsonRPCHandler      *internal.Handler
@@ -120,6 +122,16 @@
 
 // newCaptiveCore creates a new captive core backend instance and returns it.
 func newCaptiveCore(cfg *config.Config, logger *supportlog.Entry) (*ledgerbackend.CaptiveStellarCore, error) {
+	var queryServerParams *ledgerbackend.HTTPQueryServerParams
+	if cfg.CaptiveCoreHTTPPort != 0 {
+		// Only try to enable the server if the port passed is non-zero
+		queryServerParams = &ledgerbackend.HTTPQueryServerParams{
+			Port:            uint16(cfg.CaptiveCoreHTTPQueryPort),
+			ThreadPoolSize:  uint16(cfg.CaptiveCoreHTTPQueryThreadPoolSize),
+			SnapshotLedgers: uint16(cfg.CaptiveCoreHTTPQuerySnapshotLedgers),
+		}
+	}
+
 	captiveCoreTomlParams := ledgerbackend.CaptiveCoreTomlParams{
 		HTTPPort:                           &cfg.CaptiveCoreHTTPPort,
 		HistoryArchiveURLs:                 cfg.HistoryArchiveURLs,
@@ -129,6 +141,7 @@
 		EnforceSorobanDiagnosticEvents:     true,
 		EnforceSorobanTransactionMetaExtV1: true,
 		CoreBinaryPath:                     cfg.StellarCoreBinaryPath,
+		HTTPQueryServerParams:              queryServerParams,
 	}
 	captiveCoreToml, err := ledgerbackend.NewCaptiveCoreTomlFromFile(cfg.CaptiveCoreConfigPath, captiveCoreTomlParams)
 	if err != nil {
@@ -156,12 +169,13 @@
 	metricsRegistry := prometheus.NewRegistry()
 
 	daemon := &Daemon{
-		logger:          logger,
-		core:            core,
-		db:              mustOpenDatabase(cfg, logger, metricsRegistry),
-		done:            make(chan struct{}),
-		metricsRegistry: metricsRegistry,
-		coreClient:      newCoreClientWithMetrics(createStellarCoreClient(cfg), metricsRegistry),
+		logger:             logger,
+		core:               core,
+		db:                 mustOpenDatabase(cfg, logger, metricsRegistry),
+		done:               make(chan struct{}),
+		metricsRegistry:    metricsRegistry,
+		coreClient:         newCoreClientWithMetrics(createStellarCoreClient(cfg), metricsRegistry),
+		coreQueryingClient: createHighperfStellarCoreClient(cfg),
 	}
 
 	feewindows := daemon.mustInitializeStorage(cfg)
@@ -235,6 +249,17 @@
 	}
 }
 
+func createHighperfStellarCoreClient(cfg *config.Config) interfaces.FastCoreClient {
+	// It doesn't make sense to create a client if the local server is not enabled
+	if cfg.CaptiveCoreHTTPQueryPort == 0 {
+		return nil
+	}
+	return &stellarcore.Client{
+		URL:  fmt.Sprintf("http://localhost:%d", cfg.CaptiveCoreHTTPQueryPort),
+		HTTP: &http.Client{Timeout: cfg.CoreRequestTimeout},
+	}
+}
+
 func createIngestService(cfg *config.Config, logger *supportlog.Entry, daemon *Daemon,
 	feewindows *feewindow.FeeWindows, historyArchive *historyarchive.ArchiveInterface,
 ) *ingest.Service {
@@ -486,3 +511,6 @@
 		return
 	}
 }
+
+// Ensure the daemon conforms to the interface
+var _ interfaces.Daemon = (*Daemon)(nil)

cmd/stellar-rpc/internal/daemon/interfaces/interfaces.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/stellar/go/ingest/ledgerbackend"
 	proto "github.com/stellar/go/protocols/stellarcore"
+	"github.com/stellar/go/xdr"
 )
 
 // Daemon defines the interface that the Daemon would be implementing.
@@ -16,10 +17,15 @@ type Daemon interface {
 	MetricsRegistry() *prometheus.Registry
 	MetricsNamespace() string
 	CoreClient() CoreClient
+	FastCoreClient() FastCoreClient
 	GetCore() *ledgerbackend.CaptiveStellarCore
 }
 
 type CoreClient interface {
 	Info(ctx context.Context) (*proto.InfoResponse, error)
 	SubmitTransaction(ctx context.Context, txBase64 string) (*proto.TXResponse, error)
 }
+
+type FastCoreClient interface {
+	GetLedgerEntries(ctx context.Context, ledgerSeq uint32, keys ...xdr.LedgerKey) (proto.GetLedgerEntryResponse, error)
+}

cmd/stellar-rpc/internal/daemon/interfaces/noOpDaemon.go

@@ -7,6 +7,7 @@
 
 	"github.com/stellar/go/ingest/ledgerbackend"
 	proto "github.com/stellar/go/protocols/stellarcore"
+	"github.com/stellar/go/xdr"
 )
 
 // TODO: deprecate and rename to stellar_rpc
@@ -41,6 +42,10 @@
 	return d.coreClient
 }
 
+func (d *NoOpDaemon) FastCoreClient() FastCoreClient {
+	return d.coreClient
+}
+
 func (d *NoOpDaemon) GetCore() *ledgerbackend.CaptiveStellarCore {
 	return d.core
 }
@@ -54,3 +59,7 @@
 func (s noOpCoreClient) SubmitTransaction(context.Context, string) (*proto.TXResponse, error) {
 	return &proto.TXResponse{Status: proto.PreflightStatusOk}, nil
 }
+
+func (s noOpCoreClient) GetLedgerEntries(context.Context, uint32, ...xdr.LedgerKey) (proto.GetLedgerEntryResponse, error) {
+	return proto.GetLedgerEntryResponse{}, nil
+}

cmd/stellar-rpc/internal/daemon/metrics.go

@@ -111,6 +111,10 @@ func (d *Daemon) CoreClient() interfaces.CoreClient {
 	return d.coreClient
 }
 
+func (d *Daemon) FastCoreClient() interfaces.FastCoreClient {
+	return d.coreQueryingClient
+}
+
 func (d *Daemon) GetCore() *ledgerbackend.CaptiveStellarCore {
 	return d.core
 }

cmd/stellar-rpc/internal/integrationtest/infrastructure/docker/captive-core-integration-tests.cfg.tmpl

@@ -21,4 +21,4 @@ PUBLIC_KEY="GD5KD2KEZJIGTC63IGW6UMUSMVUVG5IHG64HUTFWCHVZH2N2IBOQN7PS"
 
 # should be "core" when running RPC in a container or "localhost:port" when running RPC in the host
 ADDRESS="${CORE_HOST_PORT}"
-QUALITY="MEDIUM"
+QUALITY="MEDIUM"

cmd/stellar-rpc/internal/integrationtest/infrastructure/docker/docker-compose.yml

@@ -13,7 +13,8 @@ services:
     # Note: Please keep the image pinned to an immutable tag matching the Captive Core version.
     #       This avoids implicit updates which break compatibility between
     #       the Core container and captive core.
-    image: ${CORE_IMAGE:-stellar/stellar-core:22.0.0-2138.721fd0a65.focal}
+    image: ${CORE_IMAGE:-stellar/unsafe-stellar-core:22.1.1-2251.ac9f21ac7.focal-do-not-use-in-prd}
+
     depends_on:
       - core-postgres
     environment:
@@ -23,6 +24,8 @@ services:
       - "127.0.0.1:0:11625"
       # http
       - "127.0.0.1:0:11626"
+      # high-perf http
+      - "127.0.0.1:0:11628"
       # history archive
       - "127.0.0.1:0:1570"
     entrypoint: /usr/bin/env

cmd/stellar-rpc/internal/integrationtest/infrastructure/test.go

@@ -302,6 +302,7 @@
 	coreBinaryPath         string
 	captiveCoreConfigPath  string
 	captiveCoreStoragePath string
+	captiveCoreQueryPort   int
 	archiveURL             string
 	sqlitePath             string
 }
@@ -316,16 +317,18 @@
 		"CAPTIVE_CORE_CONFIG_PATH":       vars.captiveCoreConfigPath,
 		"CAPTIVE_CORE_STORAGE_PATH":      vars.captiveCoreStoragePath,
 		"STELLAR_CAPTIVE_CORE_HTTP_PORT": "0",
-		"FRIENDBOT_URL":                  FriendbotURL,
-		"NETWORK_PASSPHRASE":             StandaloneNetworkPassphrase,
-		"HISTORY_ARCHIVE_URLS":           vars.archiveURL,
-		"LOG_LEVEL":                      "debug",
-		"DB_PATH":                        vars.sqlitePath,
-		"INGESTION_TIMEOUT":              "10m",
-		"HISTORY_RETENTION_WINDOW":       strconv.Itoa(config.OneDayOfLedgers),
-		"CHECKPOINT_FREQUENCY":           strconv.Itoa(checkpointFrequency),
-		"MAX_HEALTHY_LEDGER_LATENCY":     "10s",
-		"PREFLIGHT_ENABLE_DEBUG":         "true",
+		// TODO: allow setting it for getledgerentry tests
+		// "STELLAR_CAPTIVE_CORE_HTTP_QUERY_PORT": strconv.Itoa(vars.captiveCoreQueryPort),
+		"FRIENDBOT_URL":              FriendbotURL,
+		"NETWORK_PASSPHRASE":         StandaloneNetworkPassphrase,
+		"HISTORY_ARCHIVE_URLS":       vars.archiveURL,
+		"LOG_LEVEL":                  "debug",
+		"DB_PATH":                    vars.sqlitePath,
+		"INGESTION_TIMEOUT":          "10m",
+		"HISTORY_RETENTION_WINDOW":   strconv.Itoa(config.OneDayOfLedgers),
+		"CHECKPOINT_FREQUENCY":       strconv.Itoa(checkpointFrequency),
+		"MAX_HEALTHY_LEDGER_LATENCY": "10s",
+		"PREFLIGHT_ENABLE_DEBUG":     "true",
 	}
 }
 

cmd/stellar-rpc/internal/jsonrpc.go

@@ -156,6 +156,15 @@ func NewJSONRPCHandler(cfg *config.Config, params HandlerParams) Handler {
 
 	retentionWindow := cfg.HistoryRetentionWindow
 
+	getLedgerEntriesHandler := methods.NewGetLedgerEntriesFromDBHandler(params.Logger, params.LedgerEntryReader)
+	if params.Daemon.FastCoreClient() != nil {
+		// Prioritize getting ledger entries from core if available
+		getLedgerEntriesHandler = methods.NewGetLedgerEntriesFromCoreHandler(
+			params.Logger,
+			params.Daemon.FastCoreClient(),
+			params.LedgerEntryReader)
+	}
+
 	handlers := []struct {
 		methodName           string
 		underlyingHandler    jrpc2.Handler
@@ -222,7 +231,7 @@ func NewJSONRPCHandler(cfg *config.Config, params HandlerParams) Handler {
 		},
 		{
 			methodName:           protocol.GetLedgerEntriesMethodName,
-			underlyingHandler:    methods.NewGetLedgerEntriesHandler(params.Logger, params.LedgerEntryReader),
+			underlyingHandler:    getLedgerEntriesHandler,
 			longName:             toSnakeCase(protocol.GetLedgerEntriesMethodName),
 			queueLimit:           cfg.RequestBacklogGetLedgerEntriesQueueLimit,
 			requestDurationLimit: cfg.MaxGetLedgerEntriesExecutionDuration,