Add streaming load test backend

cmd/ingest.go

@@ -108,7 +108,7 @@ 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 from named pipe)",
 			OptType:     types.String,
 			ConfigKey:   &ledgerBackendType,
 			FlagDefault: string(ingest.LedgerBackendTypeDatastore),
@@ -122,6 +122,23 @@ func (c *ingestCmd) Command() *cobra.Command {
 			FlagDefault: "config/datastore-pubnet.toml",
 			Required:    false,
 		},
+		{
+			Name:        "loadtest-meta-pipe-path",
+			Usage:       "Filesystem path of the named pipe (FIFO) for streaming-loadtest backend. Required when ledger-backend-type is 'streaming-loadtest'.",
+			OptType:     types.String,
+			ConfigKey:   &cfg.MetaPipePath,
+			FlagDefault: "",
+			Required:    false,
+		},
+		{
+			Name:           "loadtest-ledger-close-duration",
+			Usage:          "Minimum duration between ledger emits in streaming-loadtest mode. Accepts Go duration syntax (e.g., 1s, 200ms, 0 = uncapped). Only used with streaming-loadtest backend.",
+			OptType:        types.String,
+			ConfigKey:      &cfg.LedgerCloseDuration,
+			FlagDefault:    "0s",
+			Required:       false,
+			CustomSetValue: utils.SetConfigOptionDuration,
+		},
 		{
 			Name:        "chunk-interval",
 			Usage:       "TimescaleDB chunk time interval for hypertables. Only affects future chunks. Uses PostgreSQL INTERVAL syntax.",
@@ -183,8 +200,10 @@ 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)
 			}
 
 			appTracker, err := sentry.NewSentryTracker(sentryDSN, stellarEnvironment, 5)

internal/ingest/ingest.go

@@ -39,6 +39,9 @@
 	LedgerBackendTypeRPC LedgerBackendType = "rpc"
 	// LedgerBackendTypeDatastore uses cloud storage (S3/GCS) to fetch ledgers
 	LedgerBackendTypeDatastore LedgerBackendType = "datastore"
+	// LedgerBackendTypeStreamingLoadtest reads stream-framed XDR LedgerCloseMeta
+	// from a named pipe fed by stellar-core apply-load. Dev-only.
+	LedgerBackendTypeStreamingLoadtest LedgerBackendType = "streaming-loadtest"
 )
 
 // StorageBackendConfig holds configuration for the datastore-based ledger backend
@@ -101,6 +104,14 @@
 	DBMinConns        int
 	DBMaxConnLifetime time.Duration
 	DBMaxConnIdleTime time.Duration
+	// Streaming-loadtest backend options.
+	// MetaPipePath is the filesystem path of the named pipe (FIFO) that carries
+	// stream-framed XDR LedgerCloseMeta from stellar-core apply-load. Only consulted
+	// when LedgerBackendType == LedgerBackendTypeStreamingLoadtest.
+	MetaPipePath string
+	// LedgerCloseDuration paces the streaming-loadtest backend: GetLedger sleeps
+	// until this duration has elapsed since the previous emit. 0 = uncapped.
+	LedgerCloseDuration time.Duration
 }
 
 func (c Configs) BuildPoolConfig() db.PoolConfig {
@@ -145,7 +156,7 @@
 	}
 
 	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, fmt.Errorf("configuring hypertable settings: %w", err)
 		}
 	}

internal/ingest/ingest_test.go

@@ -0,0 +1,21 @@
+package ingest
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestConfigsStreamingLoadtestFields(t *testing.T) {
+	cfg := Configs{
+		MetaPipePath:        "/tmp/fake.pipe",
+		LedgerCloseDuration: 2 * time.Second,
+	}
+	assert.Equal(t, "/tmp/fake.pipe", cfg.MetaPipePath)
+	assert.Equal(t, 2*time.Second, cfg.LedgerCloseDuration)
+}
+
+func TestLedgerBackendTypeStreamingLoadtestConstant(t *testing.T) {
+	assert.Equal(t, LedgerBackendType("streaming-loadtest"), LedgerBackendTypeStreamingLoadtest)
+}

internal/ingest/ledger_backend.go

@@ -18,6 +18,13 @@ func NewLedgerBackend(ctx context.Context, cfg Configs) (ledgerbackend.LedgerBac
 		return newDatastoreLedgerBackend(ctx, cfg.DatastoreConfigPath, cfg.NetworkPassphrase)
 	case LedgerBackendTypeRPC:
 		return newRPCLedgerBackend(cfg)
+	case LedgerBackendTypeStreamingLoadtest:
+		return NewStreamingLoadtestLedgerBackend(StreamingLoadtestBackendConfig{
+			MetaPipePath:        cfg.MetaPipePath,
+			LedgerCloseDuration: cfg.LedgerCloseDuration,
+			NetworkPassphrase:   cfg.NetworkPassphrase,
+			ArchiveURL:          cfg.ArchiveURL,
+		})
 	default:
 		return nil, fmt.Errorf("unsupported ledger backend type: %s", cfg.LedgerBackendType)
 	}