feat(ingest): rewrite backfill as a streaming fetch → process → flush pipeline

cmd/ingest.go

@@ -68,18 +68,10 @@ func (c *ingestCmd) Command() *cobra.Command {
 			Required:    false,
 		},
 		{
-			Name:        "backfill-batch-size",
-			Usage:       "Number of ledgers per batch during backfill. Defaults to 250. Lower values reduce RAM usage at cost of more DB transactions.",
+			Name:        "backfill-flush-size",
+			Usage:       "Number of ledgers to accumulate before flushing to DB in one transaction during backfill. Defaults to 100.",
 			OptType:     types.Int,
-			ConfigKey:   &cfg.BackfillBatchSize,
-			FlagDefault: 250,
-			Required:    false,
-		},
-		{
-			Name:        "backfill-db-insert-batch-size",
-			Usage:       "Number of ledgers to process before flushing buffer to DB during backfill. Defaults to 100. Lower values reduce RAM usage at cost of more DB transactions.",
-			OptType:     types.Int,
-			ConfigKey:   &cfg.BackfillDBInsertBatchSize,
+			ConfigKey:   &cfg.BackfillFlushSize,
 			FlagDefault: 100,
 			Required:    false,
 		},

go.mod

@@ -32,6 +32,7 @@ require (
 	github.com/tetratelabs/wazero v1.10.1
 	github.com/vektah/gqlparser/v2 v2.5.32
 	github.com/vikstrous/dataloadgen v0.0.9
+	golang.org/x/sync v0.19.0
 	golang.org/x/text v0.34.0
 )
 
@@ -179,7 +180,6 @@ require (
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect
 	golang.org/x/net v0.50.0 // indirect
 	golang.org/x/oauth2 v0.30.0 // indirect
-	golang.org/x/sync v0.19.0 // indirect
 	golang.org/x/sys v0.41.0 // indirect
 	golang.org/x/time v0.8.0 // indirect
 	google.golang.org/api v0.215.0 // indirect

internal/indexer/indexer.go

@@ -116,10 +116,17 @@ func NewIndexer(networkPassphrase string, pool pond.Pool, ingestionMetrics *metr
 	}
 }
 
-// ProcessLedgerTransactions processes all transactions in a ledger in parallel.
-// It collects transaction data (participants, operations, state changes) and populates the buffer in a single pass.
+// ProcessLedgerTransactions processes all transactions in a ledger. When the indexer
+// was constructed with a non-nil pool, transactions are processed in parallel via the
+// pool. When the pool is nil, transactions are processed serially — callers supply
+// their own outer parallelism (e.g. per-ledger backfill workers) and the nil-pool
+// path avoids nested fan-out.
 // Returns the total participant count for metrics.
 func (i *Indexer) ProcessLedgerTransactions(ctx context.Context, transactions []ingest.LedgerTransaction, ledgerBuffer IndexerBufferInterface) (int, error) {
+	if i.pool == nil {
+		return i.processLedgerTransactionsSerial(ctx, transactions, ledgerBuffer)
+	}
+
 	group := i.pool.NewGroupContext(ctx)
 
 	txnBuffers := make([]*IndexerBuffer, len(transactions))
@@ -161,6 +168,23 @@ func (i *Indexer) ProcessLedgerTransactions(ctx context.Context, transactions []
 	return totalParticipants, nil
 }
 
+// processLedgerTransactionsSerial processes transactions one at a time with no worker
+// pool. A nil-pool indexer uses this so its only parallelism comes from the caller (the
+// backfill process workers), avoiding nested fan-out.
+func (i *Indexer) processLedgerTransactionsSerial(ctx context.Context, transactions []ingest.LedgerTransaction, ledgerBuffer IndexerBufferInterface) (int, error) {
+	total := 0
+	for _, tx := range transactions {
+		buffer := NewIndexerBuffer()
+		count, err := i.processTransaction(ctx, tx, buffer)
+		if err != nil {
+			return 0, fmt.Errorf("processing transaction at ledger=%d tx=%d: %w", tx.Ledger.LedgerSequence(), tx.Index, err)
+		}
+		ledgerBuffer.Merge(buffer)
+		total += count
+	}
+	return total, nil
+}
+
 // processTransaction processes a single transaction - collects data and populates buffer.
 // Returns participant count for metrics.
 func (i *Indexer) processTransaction(ctx context.Context, tx ingest.LedgerTransaction, buffer *IndexerBuffer) (int, error) {

internal/indexer/indexer_test.go

@@ -1129,6 +1129,25 @@ func TestExtractContractEventsForLedger_V4OperationsShorterThanResults(t *testin
 	require.Len(t, events, 1)
 }
 
+func TestIndexer_ProcessLedgerTransactions_serialMatchesPooled(t *testing.T) {
+	txs := []ingest.LedgerTransaction{testTx, testTx2}
+
+	pooled := NewIndexer(network.TestNetworkPassphrase, pond.NewPool(runtime.NumCPU()), nil)
+	pooledBuf := NewIndexerBuffer()
+	pooledCount, err := pooled.ProcessLedgerTransactions(context.Background(), txs, pooledBuf)
+	require.NoError(t, err)
+
+	serial := NewIndexer(network.TestNetworkPassphrase, nil, nil)
+	serialBuf := NewIndexerBuffer()
+	serialCount, err := serial.ProcessLedgerTransactions(context.Background(), txs, serialBuf)
+	require.NoError(t, err)
+
+	assert.Equal(t, pooledCount, serialCount)
+	assert.Equal(t, len(pooledBuf.GetTransactions()), len(serialBuf.GetTransactions()))
+	assert.Equal(t, len(pooledBuf.GetOperations()), len(serialBuf.GetOperations()))
+	assert.Equal(t, len(pooledBuf.GetStateChanges()), len(serialBuf.GetStateChanges()))
+}
+
 // TestExtractContractEventsForLedger_UnknownMetaVersionErrors confirms the one
 // intentional behavior change: an unsupported TransactionMeta version surfaces
 // as a propagated error (fail loud) rather than being silently dropped.

internal/ingest/ingest.go

@@ -64,12 +64,9 @@
 	// BackfillWorkers limits concurrent batch processing during backfill.
 	// Defaults to runtime.NumCPU(). Lower values reduce RAM usage.
 	BackfillWorkers int
-	// BackfillBatchSize is the number of ledgers processed per batch during backfill.
-	// Defaults to 250. Lower values reduce RAM usage at cost of more DB transactions.
-	BackfillBatchSize int
-	// BackfillDBInsertBatchSize is the number of ledgers to process before flushing to DB.
-	// Defaults to 50. Lower values reduce RAM usage at cost of more DB transactions.
-	BackfillDBInsertBatchSize int
+	// BackfillFlushSize is the number of ledgers to accumulate before flushing to DB.
+	// Defaults to 100. Lower values reduce RAM usage at cost of more DB transactions.
+	BackfillFlushSize int
 	// ChunkInterval sets the TimescaleDB chunk time interval for hypertables.
 	// Only affects future chunks. Uses PostgreSQL INTERVAL syntax (e.g., "1 day", "7 days").
 	ChunkInterval string
@@ -135,7 +132,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)
 		}
 	}
@@ -238,33 +235,32 @@
 	wasmExtractor := services.NewWasmSpecExtractor()
 	go func() {
 		<-ctx.Done()
 		if err := wasmExtractor.Close(context.Background()); err != nil {
 			log.Ctx(ctx).Warnf("closing wasm spec extractor on shutdown: %v", err)
 		}
 	}()
 
 	ingestService, err := services.NewIngestService(services.IngestServiceConfig{
-		IngestionMode:             cfg.IngestionMode,
-		Models:                    models,
-		OldestLedgerCursorName:    cfg.OldestLedgerCursorName,
-		AppTracker:                cfg.AppTracker,
-		RPCService:                rpcService,
-		LedgerBackend:             ledgerBackend,
-		LedgerBackendFactory:      ledgerBackendFactory,
-		TokenIngestionService:     tokenIngestionService,
-		CheckpointService:         checkpointService,
-		Metrics:                   m,
-		GetLedgersLimit:           cfg.GetLedgersLimit,
-		Network:                   cfg.Network,
-		NetworkPassphrase:         cfg.NetworkPassphrase,
-		Archive:                   archive,
-		BackfillWorkers:           cfg.BackfillWorkers,
-		BackfillBatchSize:         cfg.BackfillBatchSize,
-		BackfillDBInsertBatchSize: cfg.BackfillDBInsertBatchSize,
-		ProtocolProcessors:        protocolProcessors,
-		ProtocolValidators:        protocolValidators,
-		WasmSpecExtractor:         wasmExtractor,
-		ContractMetadataService:   contractMetadataService,
+		IngestionMode:           cfg.IngestionMode,
+		Models:                  models,
+		OldestLedgerCursorName:  cfg.OldestLedgerCursorName,
+		AppTracker:              cfg.AppTracker,
+		RPCService:              rpcService,
+		LedgerBackend:           ledgerBackend,
+		LedgerBackendFactory:    ledgerBackendFactory,
+		TokenIngestionService:   tokenIngestionService,
+		CheckpointService:       checkpointService,
+		Metrics:                 m,
+		GetLedgersLimit:         cfg.GetLedgersLimit,
+		Network:                 cfg.Network,
+		NetworkPassphrase:       cfg.NetworkPassphrase,
+		Archive:                 archive,
+		BackfillWorkers:         cfg.BackfillWorkers,
+		BackfillFlushSize:       cfg.BackfillFlushSize,
+		ProtocolProcessors:      protocolProcessors,
+		ProtocolValidators:      protocolValidators,
+		WasmSpecExtractor:       wasmExtractor,
+		ContractMetadataService: contractMetadataService,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("instantiating ingest service: %w", err)

internal/services/backfill_compressor.go

@@ -0,0 +1,124 @@
+package services
+
+import (
+	"context"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/stellar/go-stellar-sdk/support/log"
+)
+
+// backfillCompressor compresses uncompressed TimescaleDB chunks as the flusher reports
+// contiguous progress. The flusher sends the close time of the highest contiguously-
+// persisted ledger ("safe end") on triggerCh; the compressor compresses any uncompressed
+// chunk ending before it. The signal is monotonic, so no watermark or ordering state is
+// needed.
+type backfillCompressor struct {
+	pool      *pgxpool.Pool
+	tables    []string
+	triggerCh chan time.Time
+	done      chan struct{}
+}
+
+func newBackfillCompressor(ctx context.Context, pool *pgxpool.Pool, tables []string) *backfillCompressor {
+	c := &backfillCompressor{
+		pool:      pool,
+		tables:    tables,
+		triggerCh: make(chan time.Time, 256),
+		done:      make(chan struct{}),
+	}
+	go c.run(ctx)
+	return c
+}
+
+// trigger reports that every ledger with close time <= safeEnd is persisted. The send is
+// intentionally blocking: it applies backpressure if compression falls behind, and it
+// guarantees the final (highest) safe-end is delivered so the tail always compresses.
+func (c *backfillCompressor) trigger(safeEnd time.Time) {
+	c.triggerCh <- safeEnd
+}
+
+// Wait stops accepting triggers and blocks until the background goroutine drains.
+func (c *backfillCompressor) Wait() {
+	close(c.triggerCh)
+	<-c.done
+}
+
+func (c *backfillCompressor) run(ctx context.Context) {
+	defer close(c.done)
+	total := 0
+	for safeEnd := range c.triggerCh {
+		safeEnd = drainLatest(c.triggerCh, safeEnd) // collapse a burst into one pass
+		for _, table := range c.tables {
+			total += c.compressTable(ctx, table, safeEnd)
+		}
+	}
+	log.Ctx(ctx).Infof("Progressive compression complete: %d chunks compressed", total)
+}
+
+// drainLatest collapses all currently-queued safe-end values into the maximum, so a burst
+// of flush signals triggers a single compression pass instead of one per flush.
+func drainLatest(ch <-chan time.Time, end time.Time) time.Time {
+	for {
+		select {
+		case next, ok := <-ch:
+			if !ok {
+				return end
+			}
+			if next.After(end) {
+				end = next
+			}
+		default:
+			return end
+		}
+	}
+}
+
+// compressTable compresses uncompressed chunks for one table whose range_end < safeEnd.
+func (c *backfillCompressor) compressTable(ctx context.Context, table string, safeEnd time.Time) int {
+	rows, err := c.pool.Query(ctx,
+		`SELECT c.chunk_schema || '.' || c.chunk_name
+		 FROM timescaledb_information.chunks c
+		 WHERE c.hypertable_name = $1
+		   AND NOT c.is_compressed
+		   AND c.range_end < $2::timestamptz`,
+		table, safeEnd)
+	if err != nil {
+		log.Ctx(ctx).Warnf("Failed to list chunks for %s: %v", table, err)
+		return 0
+	}
+	var chunks []string
+	for rows.Next() {
+		var chunk string
+		if scanErr := rows.Scan(&chunk); scanErr != nil {
+			log.Ctx(ctx).Warnf("Failed to scan chunk row for %s: %v", table, scanErr)
+			continue
+		}
+		chunks = append(chunks, chunk)
+	}
+	if err := rows.Err(); err != nil {
+		log.Ctx(ctx).Warnf("Error iterating chunk rows for %s: %v", table, err)
+		rows.Close()
+		return 0
+	}
+	rows.Close()
+
+	compressed := 0
+	for _, chunk := range chunks {
+		select {
+		case <-ctx.Done():
+			return compressed
+		default:
+		}
+		if _, err := c.pool.Exec(ctx, `SELECT compress_chunk($1::regclass)`, chunk); err != nil {
+			log.Ctx(ctx).Warnf("Failed to compress chunk %s: %v", chunk, err)
+			continue
+		}
+		compressed++
+	}
+	if compressed > 0 {
+		log.Ctx(ctx).Infof("Compressed %d chunks for %s (safe end: %s)", compressed, table, safeEnd.Format(time.RFC3339))
+	}
+	return compressed
+}

internal/services/backfill_compressor_test.go

@@ -0,0 +1,27 @@
+package services
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_drainLatest_coalescesToMax(t *testing.T) {
+	base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
+	ch := make(chan time.Time, 8)
+	ch <- base.Add(2 * time.Hour)
+	ch <- base.Add(1 * time.Hour)
+	ch <- base.Add(3 * time.Hour)
+
+	got := drainLatest(ch, base)
+
+	assert.Equal(t, base.Add(3*time.Hour), got) // max of seed + queued
+	assert.Empty(t, ch)                         // channel fully drained
+}
+
+func Test_drainLatest_returnsSeedWhenEmpty(t *testing.T) {
+	base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
+	ch := make(chan time.Time, 1)
+	assert.Equal(t, base, drainLatest(ch, base))
+}