Add streaming-loadtest ledger backend for apply-load replay

cmd/ingest.go

@@ -85,11 +85,11 @@ func (c *ingestCmd) Command() *cobra.Command {
 		},
 		{
 			Name:        "archive-url",
-			Usage:       "Archive URL for history archives",
+			Usage:       "Archive URL for history archives. Required for every backend except 'streaming-loadtest', which runs without a history archive.",
 			OptType:     types.String,
 			ConfigKey:   &cfg.ArchiveURL,
 			FlagDefault: "https://history.stellar.org/prd/core-testnet/core_testnet_001/",
-			Required:    true,
+			Required:    false,
 		},
 		{
 			Name:        "checkpoint-frequency",
@@ -101,12 +101,30 @@ func (c *ingestCmd) Command() *cobra.Command {
 		},
 		{
 			Name:        "ledger-backend-type",
-			Usage:       "Type of ledger backend to use for fetching ledgers. Options: 'rpc' or 'datastore' (default)",
+			Usage:       "Type of ledger backend to use for fetching ledgers. Options: 'rpc', 'datastore' (default), or 'streaming-loadtest' (dev-only; reads apply-load ledger meta from named pipes)",
 			OptType:     types.String,
 			ConfigKey:   &ledgerBackendType,
 			FlagDefault: string(ingest.LedgerBackendTypeDatastore),
 			Required:    false,
 		},
+		{
+			Name:           "loadtest-meta-pipe-paths",
+			Usage:          "Dev-only. Comma-separated named pipe (FIFO) paths carrying apply-load ledger meta, one per apply-load process. Required when ledger-backend-type is 'streaming-loadtest', ignored otherwise.",
+			OptType:        types.String,
+			CustomSetValue: utils.SetConfigOptionStringList,
+			ConfigKey:      &cfg.LoadtestMetaPipePaths,
+			FlagDefault:    "",
+			Required:       false,
+		},
+		{
+			Name:           "loadtest-ledger-close-duration",
+			Usage:          "Dev-only. Minimum interval between ledgers in streaming-loadtest mode (Go duration string, e.g. \"5s\"). \"0s\" leaves the stream uncapped.",
+			OptType:        types.String,
+			CustomSetValue: utils.SetConfigOptionDuration,
+			ConfigKey:      &cfg.LoadtestLedgerCloseDuration,
+			FlagDefault:    "0s",
+			Required:       false,
+		},
 		{
 			Name:        "chunk-interval",
 			Usage:       "TimescaleDB chunk time interval for hypertables. Only affects future chunks. Uses PostgreSQL INTERVAL syntax.",
@@ -173,8 +191,20 @@ func (c *ingestCmd) Command() *cobra.Command {
 				cfg.LedgerBackendType = ingest.LedgerBackendTypeRPC
 			case string(ingest.LedgerBackendTypeDatastore):
 				cfg.LedgerBackendType = ingest.LedgerBackendTypeDatastore
+			case string(ingest.LedgerBackendTypeStreamingLoadtest):
+				cfg.LedgerBackendType = ingest.LedgerBackendTypeStreamingLoadtest
 			default:
-				return fmt.Errorf("invalid ledger-backend-type '%s', must be 'rpc' or 'datastore'", ledgerBackendType)
+				return fmt.Errorf("invalid ledger-backend-type '%s', must be 'rpc', 'datastore', or 'streaming-loadtest'", ledgerBackendType)
+			}
+
+			// The streaming-loadtest backend needs its pipes and runs without a history
+			// archive; every other backend reads a history archive to build initial state.
+			if cfg.LedgerBackendType == ingest.LedgerBackendTypeStreamingLoadtest {
+				if len(cfg.LoadtestMetaPipePaths) == 0 {
+					return fmt.Errorf("loadtest-meta-pipe-paths is required when ledger-backend-type is 'streaming-loadtest'")
+				}
+			} else if cfg.ArchiveURL == "" {
+				return fmt.Errorf("archive-url is required when ledger-backend-type is '%s'", ledgerBackendType)
 			}
 
 			appTracker, err := sentry.NewSentryTracker(sentryDSN, stellarEnvironment, 5)

cmd/protocol_migrate.go

@@ -131,6 +131,10 @@ func buildMigrationCommand(
 				}
 			case string(ingest.LedgerBackendTypeDatastore):
 				// datastore-bucket-path is validated via Required:true in DatastoreOptions.
+			case string(ingest.LedgerBackendTypeStreamingLoadtest):
+				// Migrations replay arbitrary historical ranges; the streaming backend only
+				// ever yields the synthetic ledgers currently being written to its pipes.
+				return fmt.Errorf("--ledger-backend-type %q is not supported for protocol migration", opts.ledgerBackendType)
 			default:
 				return fmt.Errorf("invalid --ledger-backend-type %q, must be 'rpc' or 'datastore'", opts.ledgerBackendType)
 			}

cmd/utils/custom_set_value.go

@@ -108,6 +108,27 @@ func SetConfigOptionAssets(co *config.ConfigOption) error {
 	return nil
 }
 
+// SetConfigOptionStringList parses a comma-separated list from the CLI flag or environment
+// variable and stores the result in a *[]string ConfigKey. Surrounding whitespace is trimmed
+// from each element and empty elements are dropped, so "a, b," yields ["a", "b"]. An empty
+// input yields an empty slice.
+func SetConfigOptionStringList(co *config.ConfigOption) error {
+	key, ok := co.ConfigKey.(*[]string)
+	if !ok {
+		return unexpectedTypeError(key, co)
+	}
+
+	values := []string{}
+	for _, value := range strings.Split(viper.GetString(co.Name), ",") {
+		if trimmed := strings.TrimSpace(value); trimmed != "" {
+			values = append(values, trimmed)
+		}
+	}
+	*key = values
+
+	return nil
+}
+
 // SetConfigOptionDuration parses a Go duration string (e.g. "5m", "10s") from the CLI flag
 // or environment variable and stores the result in a *time.Duration ConfigKey.
 func SetConfigOptionDuration(co *config.ConfigOption) error {

cmd/utils/custom_set_value_test.go

@@ -252,6 +252,46 @@ func TestSetConfigOptionAssets(t *testing.T) {
 	}
 }
 
+func TestSetConfigOptionStringList(t *testing.T) {
+	opts := struct{ paths []string }{}
+
+	co := config.ConfigOption{
+		Name:           "loadtest-meta-pipe-paths",
+		OptType:        types.String,
+		CustomSetValue: SetConfigOptionStringList,
+		ConfigKey:      &opts.paths,
+	}
+
+	testCases := []customSetterTestCase[[]string]{
+		{
+			name:       "yields an empty slice if the value is empty",
+			wantResult: []string{},
+		},
+		{
+			name:       "handles a single value through the CLI flag",
+			args:       []string{"--loadtest-meta-pipe-paths", "/tmp/a.pipe"},
+			wantResult: []string{"/tmp/a.pipe"},
+		},
+		{
+			name:       "trims whitespace and drops empty elements",
+			args:       []string{"--loadtest-meta-pipe-paths", " /tmp/a.pipe , ,/tmp/b.pipe,"},
+			wantResult: []string{"/tmp/a.pipe", "/tmp/b.pipe"},
+		},
+		{
+			name:       "handles a list through the ENV var",
+			envValue:   "/tmp/a.pipe,/tmp/b.pipe",
+			wantResult: []string{"/tmp/a.pipe", "/tmp/b.pipe"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			opts.paths = nil
+			customSetterTester(t, tc, co)
+		})
+	}
+}
+
 func TestSetConfigOptionDuration(t *testing.T) {
 	opts := struct{ d time.Duration }{}
 

go.mod

@@ -156,6 +156,7 @@ require (
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/xdrpp/goxdr v0.1.1 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/contrib/detectors/gcp v1.38.0 // indirect
@@ -172,6 +173,7 @@ require (
 	go.yaml.in/yaml/v2 v2.4.3 // indirect
 	golang.org/x/crypto v0.52.0 // indirect
 	golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect
+	golang.org/x/mod v0.35.0 // indirect
 	golang.org/x/net v0.55.0 // indirect
 	golang.org/x/oauth2 v0.32.0 // indirect
 	golang.org/x/sync v0.20.0 // indirect

go.sum

@@ -465,6 +465,8 @@ golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 h1:mgKeJMpvi0yx/sU5GsxQ7p6s2
 golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546/go.mod h1:j/pmGrbnkbPtQfxEe5D0VQhZC6qKbfKifgD0oM7sR70=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM=
+golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=

internal/ingest/ingest.go

@@ -44,6 +44,10 @@
 	LedgerBackendTypeRPC LedgerBackendType = "rpc"
 	// LedgerBackendTypeDatastore uses cloud storage (S3/GCS) to fetch ledgers
 	LedgerBackendTypeDatastore LedgerBackendType = "datastore"
+	// LedgerBackendTypeStreamingLoadtest reads synthetic ledgers from named
+	// pipes written by stellar-core apply-load. Dev-only, for load testing
+	// the standard ingestion path.
+	LedgerBackendTypeStreamingLoadtest LedgerBackendType = "streaming-loadtest"
 )
 
 type Configs struct {
@@ -66,6 +70,12 @@
 	LedgerBackendType LedgerBackendType
 	// Datastore holds the datastore ledger backend configuration (flag/env driven).
 	Datastore DatastoreConfig
+	// LoadtestMetaPipePaths are the FIFO paths for the streaming-loadtest
+	// backend, one per apply-load process (their frames are merged per ledger).
+	LoadtestMetaPipePaths []string
+	// LoadtestLedgerCloseDuration is the minimum interval between ledgers in
+	// streaming-loadtest mode. 0 = uncapped.
+	LoadtestLedgerCloseDuration time.Duration
 	// BackfillWorkers limits concurrent batch processing during backfill.
 	// Defaults to runtime.NumCPU(). Lower values reduce RAM usage.
 	BackfillWorkers int
@@ -171,7 +181,7 @@
 	// must not mutate or remove the policies the live pods rely on; and an active
 	// retention policy would drop the very history a backfill is writing.
 	if cfg.IngestionMode == services.IngestionModeLive {
 		if err := configureHypertableSettings(ctx, dbConnectionPool, cfg.ChunkInterval, cfg.RetentionPeriod, cfg.OldestLedgerCursorName, cfg.CompressionScheduleInterval, cfg.CompressAfter, cfg.MaxChunksToCompress); err != nil {
 			return nil, nil, fmt.Errorf("configuring hypertable settings: %w", err)
 		}
 	}
@@ -217,16 +227,24 @@
 		MetricsService:          m,
 	}
 
-	// Initialize history archive once for use by both TokenIngestionService and IngestService
-	archive, err := historyarchive.Connect(
-		cfg.ArchiveURL,
-		historyarchive.ArchiveOptions{
-			NetworkPassphrase:   cfg.NetworkPassphrase,
-			CheckpointFrequency: uint32(cfg.CheckpointFrequency),
-		},
-	)
-	if err != nil {
-		return nil, nil, fmt.Errorf("connecting to history archive: %w", err)
+	// Initialize history archive once for use by both TokenIngestionService and IngestService.
+	// The streaming-loadtest backend runs without one: apply-load's benchmark mode
+	// publishes no history archive, so ingestion starts from an empty database and
+	// balance state materializes from the ledger stream itself. archive must stay a
+	// nil interface (not a typed-nil *Archive) — downstream code branches on == nil.
+	var archive historyarchive.ArchiveInterface
+	if cfg.LedgerBackendType != LedgerBackendTypeStreamingLoadtest {
+		connectedArchive, err := historyarchive.Connect(
+			cfg.ArchiveURL,
+			historyarchive.ArchiveOptions{
+				NetworkPassphrase:   cfg.NetworkPassphrase,
+				CheckpointFrequency: uint32(cfg.CheckpointFrequency),
+			},
+		)
+		if err != nil {
+			return nil, nil, fmt.Errorf("connecting to history archive: %w", err)
+		}
+		archive = connectedArchive
 	}
 
 	tokenIngestionService := services.NewTokenIngestionService(services.TokenIngestionServiceConfig{

internal/ingest/ledger_backend.go

@@ -16,6 +16,11 @@ func NewLedgerBackend(ctx context.Context, cfg Configs) (ledgerbackend.LedgerBac
 		return newDatastoreLedgerBackend(ctx, cfg.Datastore, cfg.NetworkPassphrase)
 	case LedgerBackendTypeRPC:
 		return newRPCLedgerBackend(cfg)
+	case LedgerBackendTypeStreamingLoadtest:
+		return NewStreamingLoadtestLedgerBackend(StreamingLoadtestBackendConfig{
+			MetaPipePaths:       cfg.LoadtestMetaPipePaths,
+			LedgerCloseDuration: cfg.LoadtestLedgerCloseDuration,
+		})
 	default:
 		return nil, fmt.Errorf("unsupported ledger backend type: %s", cfg.LedgerBackendType)
 	}

internal/ingest/streaming_loadtest_corpus_test.go

@@ -0,0 +1,75 @@
+package ingest
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stellar/go-stellar-sdk/ingest/ledgerbackend"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/stellar/wallet-backend/internal/indexer"
+)
+
+// TestStreamingLoadtestBackendRealCorpus replays real `stellar-core apply-load`
+// meta through the backend and the production transaction reader. It proves,
+// against real core output rather than hand-built fixtures, that renumbered
+// and merged ledgers still parse through the exact code path live ingestion
+// uses (transaction-set-to-result pairing included).
+//
+// Opt-in: set STREAMING_LOADTEST_CORPUS to a comma-separated list of meta.xdr
+// files (regular files work; EOF exercises the reopen path by replaying the
+// file as a new stream epoch). Generate them by running
+// `stellar-core apply-load` (BUILD_TESTS image) with METADATA_OUTPUT_STREAM
+// pointed at a file, one run per transaction profile.
+func TestStreamingLoadtestBackendRealCorpus(t *testing.T) {
+	corpus := os.Getenv("STREAMING_LOADTEST_CORPUS")
+	if corpus == "" {
+		t.Skip("set STREAMING_LOADTEST_CORPUS=<meta.xdr>[,<meta.xdr>...] to run")
+	}
+	paths := strings.Split(corpus, ",")
+
+	// apply-load hard-overrides its network passphrase to this value.
+	const passphrase = "Apply Load"
+	// Enough ledgers to cross at least one EOF/reopen boundary per file with
+	// the reference smoke corpora (50 benchmark ledgers plus setup each).
+	const ledgersToRead = 200
+
+	backend, err := NewStreamingLoadtestLedgerBackend(StreamingLoadtestBackendConfig{
+		MetaPipePaths: paths,
+	})
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+	require.NoError(t, backend.PrepareRange(ctx, ledgerbackend.UnboundedRange(1)))
+	defer func() {
+		cancel()
+		require.NoError(t, backend.Close())
+	}()
+
+	var lastCloseTime int64
+	totalTxs := 0
+	for seq := uint32(1); seq <= ledgersToRead; seq++ {
+		lcm, err := backend.GetLedger(ctx, seq)
+		require.NoError(t, err, "ledger %d", seq)
+
+		require.Equal(t, seq, lcm.LedgerSequence())
+		ct := lcm.LedgerCloseTime()
+		require.Positive(t, ct, "ledger %d close time", seq)
+		require.GreaterOrEqual(t, ct, lastCloseTime, "ledger %d close time regressed", seq)
+		lastCloseTime = ct
+
+		// The production read path: this is what live ingestion runs on every
+		// ledger, so a merged ledger it cannot parse would fail here first.
+		txs, err := indexer.GetLedgerTransactions(ctx, passphrase, lcm)
+		require.NoError(t, err, "reading transactions of ledger %d", seq)
+		totalTxs += len(txs)
+	}
+
+	assert.Positive(t, totalTxs)
+	t.Logf("read %d ledgers, %d transactions total from %d file(s)", ledgersToRead, totalTxs, len(paths))
+}