diff --git a/cmd/ingest.go b/cmd/ingest.go index 35de96210..cdad33673 100644 --- a/cmd/ingest.go +++ b/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, }, diff --git a/go.mod b/go.mod index 6ad48cd27..c3bf56f34 100644 --- a/go.mod +++ b/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 diff --git a/internal/indexer/indexer.go b/internal/indexer/indexer.go index 0cce68f99..0a940f12d 100644 --- a/internal/indexer/indexer.go +++ b/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) { diff --git a/internal/indexer/indexer_test.go b/internal/indexer/indexer_test.go index 87b5cbdd1..81713569a 100644 --- a/internal/indexer/indexer_test.go +++ b/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. diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index e29d58cfc..5134c2445 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -64,12 +64,9 @@ type Configs struct { // 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 @@ -244,27 +241,26 @@ func setupDeps(cfg Configs) (services.IngestService, error) { }() 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) diff --git a/internal/services/backfill_compressor.go b/internal/services/backfill_compressor.go new file mode 100644 index 000000000..2ab3d2876 --- /dev/null +++ b/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 +} diff --git a/internal/services/backfill_compressor_test.go b/internal/services/backfill_compressor_test.go new file mode 100644 index 000000000..8a3237592 --- /dev/null +++ b/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)) +} diff --git a/internal/services/backfill_pipeline.go b/internal/services/backfill_pipeline.go new file mode 100644 index 000000000..fa1d34eff --- /dev/null +++ b/internal/services/backfill_pipeline.go @@ -0,0 +1,251 @@ +package services + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/jackc/pgx/v5" + "golang.org/x/sync/errgroup" + + "github.com/stellar/go-stellar-sdk/ingest/ledgerbackend" + "github.com/stellar/go-stellar-sdk/support/log" + "github.com/stellar/go-stellar-sdk/xdr" + + "github.com/stellar/wallet-backend/internal/data" + "github.com/stellar/wallet-backend/internal/db" + "github.com/stellar/wallet-backend/internal/indexer" +) + +// rawLedger is a decoded ledger handed from the fetcher to a process worker. +type rawLedger struct { + seq uint32 + meta xdr.LedgerCloseMeta +} + +// processedLedger is the indexer output for one ledger, handed from a process worker to +// the flusher. +type processedLedger struct { + seq uint32 + closeTime time.Time + buffer *indexer.IndexerBuffer +} + +// orderedBuffer turns out-of-order processed ledgers into contiguous ascending runs. +// Process workers finish in arbitrary order; orderedBuffer releases a ledger only once +// every earlier ledger has arrived, so the flusher always sees a gapless prefix. +type orderedBuffer struct { + next uint32 + pending map[uint32]processedLedger +} + +func newOrderedBuffer(start uint32) *orderedBuffer { + return &orderedBuffer{next: start, pending: make(map[uint32]processedLedger)} +} + +// add records p and returns the contiguous run (possibly empty) now releasable. +func (o *orderedBuffer) add(p processedLedger) []processedLedger { + o.pending[p.seq] = p + var run []processedLedger + for { + next, ok := o.pending[o.next] + if !ok { + break + } + run = append(run, next) + delete(o.pending, o.next) + o.next++ + } + return run +} + +// fetchLedgers consumes [start,end] in ascending order from a single prepared backend, +// emitting decoded ledgers onto rawCh. It closes rawCh on return. A GetLedger error is +// fatal and returned immediately — the fork retries transient downloads internally, so a +// surfaced error is non-recoverable (fail-fast). +func (m *ingestService) fetchLedgers(ctx context.Context, backend ledgerbackend.LedgerBackend, start, end uint32, rawCh chan<- rawLedger) error { + for seq := start; seq <= end; seq++ { + meta, err := backend.GetLedger(ctx, seq) + if err != nil { + return fmt.Errorf("getting ledger %d: %w", seq, err) + } + select { + case rawCh <- rawLedger{seq: seq, meta: meta}: + case <-ctx.Done(): + return fmt.Errorf("context cancelled fetching ledger %d: %w", seq, ctx.Err()) + } + } + return nil +} + +// processWorker reads decoded ledgers from rawCh, runs the indexer into a fresh buffer +// with its own serial indexer, and emits the result onto resultCh. Returns on rawCh close +// or context cancellation. +func (m *ingestService) processWorker(ctx context.Context, idx *indexer.Indexer, rawCh <-chan rawLedger, resultCh chan<- processedLedger) error { + for raw := range rawCh { + buffer := indexer.NewIndexerBuffer() + if err := m.processLedgerWith(ctx, idx, raw.meta, buffer); err != nil { + return err + } + select { + case resultCh <- processedLedger{seq: raw.seq, closeTime: raw.meta.ClosedAt(), buffer: buffer}: + case <-ctx.Done(): + return fmt.Errorf("context cancelled emitting ledger %d: %w", raw.seq, ctx.Err()) + } + } + return nil +} + +// runFlusher consumes processed ledgers, reorders them into contiguous runs, flushes every +// flush-size ledgers in one transaction, advances the oldest cursor, and signals the +// compressor with the highest contiguously-persisted close time. start is the gap's first +// ledger sequence. +func (m *ingestService) runFlusher(ctx context.Context, start uint32, resultCh <-chan processedLedger, compressor *backfillCompressor) error { + ob := newOrderedBuffer(start) + merged := indexer.NewIndexerBuffer() + var ( + count uint32 + lowestSeq uint32 + contiguousEnd time.Time + ) + + flush := func() error { + if count == 0 { + return nil + } + if err := m.flushBackfillBuffer(ctx, merged, lowestSeq); err != nil { + return err + } + m.appMetrics.Ingestion.OldestLedger.Set(float64(lowestSeq)) + compressor.trigger(contiguousEnd) + merged = indexer.NewIndexerBuffer() + count = 0 + return nil + } + + for { + select { + case <-ctx.Done(): + return fmt.Errorf("context cancelled in flusher: %w", ctx.Err()) + case p, ok := <-resultCh: + if !ok { + return flush() // final partial flush at stream end + } + for _, r := range ob.add(p) { + if count == 0 { + lowestSeq = r.seq + } + merged.Merge(r.buffer) + contiguousEnd = r.closeTime + count++ + if count >= m.backfillFlushSize { + if err := flush(); err != nil { + return err + } + } + } + } + } +} + +// flushBackfillBuffer persists one merged buffer and advances the oldest cursor in a single +// transaction, with retry. synchronous_commit is disabled for the transaction — safe for +// backfill because a crash before WAL flush just re-creates the gap, which the next run +// detects and refills. +func (m *ingestService) flushBackfillBuffer(ctx context.Context, buffer *indexer.IndexerBuffer, lowestSeq uint32) error { + var lastErr error + for attempt := 0; attempt < maxIngestProcessedDataRetries; attempt++ { + select { + case <-ctx.Done(): + return fmt.Errorf("context cancelled: %w", ctx.Err()) + default: + } + + err := db.RunInTransaction(ctx, m.models.DB, func(dbTx pgx.Tx) error { + if _, txErr := dbTx.Exec(ctx, "SET LOCAL synchronous_commit = off"); txErr != nil { + return fmt.Errorf("setting synchronous_commit=off: %w", txErr) + } + if _, _, insErr := m.insertIntoDB(ctx, dbTx, buffer); insErr != nil { + return fmt.Errorf("inserting processed data: %w", insErr) + } + if curErr := m.models.IngestStore.UpdateMin(ctx, dbTx, m.oldestLedgerCursorName, lowestSeq); curErr != nil { + return fmt.Errorf("updating oldest cursor: %w", curErr) + } + return nil + }) + if err == nil { + return nil + } + lastErr = err + m.appMetrics.Ingestion.RetriesTotal.WithLabelValues("batch_flush").Inc() + if attempt == maxIngestProcessedDataRetries-1 { + break + } + backoff := time.Duration(1< maxIngestProcessedDataRetryBackoff { + backoff = maxIngestProcessedDataRetryBackoff + } + log.Ctx(ctx).Warnf("Error flushing backfill buffer (attempt %d/%d): %v, retrying in %v...", + attempt+1, maxIngestProcessedDataRetries, lastErr, backoff) + select { + case <-ctx.Done(): + return fmt.Errorf("context cancelled during backoff: %w", ctx.Err()) + case <-time.After(backoff): + } + } + m.appMetrics.Ingestion.RetryExhaustionsTotal.WithLabelValues("batch_flush").Inc() + return lastErr +} + +// runGapPipeline processes one contiguous gap through fetch → process → flush. It owns a +// single backend (the fork's single-consumer contract requires exactly one sequential +// GetLedger caller) and tears the pipeline down on the first error (fail-fast). +func (m *ingestService) runGapPipeline(ctx context.Context, gap data.LedgerRange, compressor *backfillCompressor) error { + backend, err := m.ledgerBackendFactory(ctx) + if err != nil { + return fmt.Errorf("creating ledger backend: %w", err) + } + defer func() { + if closeErr := backend.Close(); closeErr != nil { + log.Ctx(ctx).Warnf("Error closing backend for gap [%d-%d]: %v", gap.GapStart, gap.GapEnd, closeErr) + } + }() + if err := backend.PrepareRange(ctx, ledgerbackend.BoundedRange(gap.GapStart, gap.GapEnd)); err != nil { + return fmt.Errorf("preparing backend range [%d-%d]: %w", gap.GapStart, gap.GapEnd, err) + } + + chanCap := max(2*int(m.backfillWorkers), 64) + rawCh := make(chan rawLedger, chanCap) + resultCh := make(chan processedLedger, chanCap) + + g, gctx := errgroup.WithContext(ctx) + + g.Go(func() error { + defer close(rawCh) + return m.fetchLedgers(gctx, backend, gap.GapStart, gap.GapEnd, rawCh) + }) + + var wg sync.WaitGroup + for range m.backfillWorkers { + wg.Add(1) + idx := indexer.NewIndexer(m.networkPassphrase, nil, m.appMetrics.Ingestion) + g.Go(func() error { + defer wg.Done() + return m.processWorker(gctx, idx, rawCh, resultCh) + }) + } + // Close resultCh once every worker (its only senders) has returned — the canonical + // fan-in close. This goroutine ends when wg drains, which happens on completion or + // cancellation, so it never leaks. + go func() { + wg.Wait() + close(resultCh) + }() + + g.Go(func() error { + return m.runFlusher(gctx, gap.GapStart, resultCh, compressor) + }) + + return g.Wait() //nolint:wrapcheck // errgroup stages already wrap their own errors; startBackfilling adds the gap context +} diff --git a/internal/services/backfill_pipeline_test.go b/internal/services/backfill_pipeline_test.go new file mode 100644 index 000000000..e80738f27 --- /dev/null +++ b/internal/services/backfill_pipeline_test.go @@ -0,0 +1,33 @@ +package services + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func seqs(run []processedLedger) []uint32 { + out := make([]uint32, len(run)) + for i, p := range run { + out[i] = p.seq + } + return out +} + +func Test_orderedBuffer_yieldsContiguousAscendingRuns(t *testing.T) { + ct := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) + ob := newOrderedBuffer(100) + + assert.Empty(t, seqs(ob.add(processedLedger{seq: 102, closeTime: ct}))) + assert.Empty(t, seqs(ob.add(processedLedger{seq: 101, closeTime: ct}))) + + run := ob.add(processedLedger{seq: 100, closeTime: ct}) + require.Equal(t, []uint32{100, 101, 102}, seqs(run)) + + assert.Empty(t, seqs(ob.add(processedLedger{seq: 104, closeTime: ct}))) + + run = ob.add(processedLedger{seq: 103, closeTime: ct}) + require.Equal(t, []uint32{103, 104}, seqs(run)) +} diff --git a/internal/services/ingest.go b/internal/services/ingest.go index 99605a501..73e39d585 100644 --- a/internal/services/ingest.go +++ b/internal/services/ingest.go @@ -86,9 +86,8 @@ type IngestServiceConfig struct { GetLedgersLimit int // === Backfill Tuning === - BackfillWorkers int - BackfillBatchSize int - BackfillDBInsertBatchSize int + BackfillWorkers int + BackfillFlushSize int } // generateAdvisoryLockID creates a deterministic advisory lock ID based on the network name. @@ -110,29 +109,28 @@ type IngestService interface { var _ IngestService = (*ingestService)(nil) type ingestService struct { - ingestionMode string - models *data.Models - oldestLedgerCursorName string - advisoryLockID int - appTracker apptracker.AppTracker - rpcService RPCService - ledgerBackend ledgerbackend.LedgerBackend - ledgerBackendFactory LedgerBackendFactory - tokenIngestionService TokenIngestionService - checkpointService CheckpointService - appMetrics *metrics.Metrics - networkPassphrase string - getLedgersLimit int - ledgerIndexer *indexer.Indexer - archive historyarchive.ArchiveInterface - backfillPool pond.Pool - backfillBatchSize uint32 - backfillDBInsertBatchSize uint32 - knownContractIDs set.Set[string] - contractMetadataService ContractMetadataService - protocolProcessors map[string]ProtocolProcessor - protocolValidators []ProtocolValidator - wasmSpecExtractor WasmSpecExtractor + ingestionMode string + models *data.Models + oldestLedgerCursorName string + advisoryLockID int + appTracker apptracker.AppTracker + rpcService RPCService + ledgerBackend ledgerbackend.LedgerBackend + ledgerBackendFactory LedgerBackendFactory + tokenIngestionService TokenIngestionService + checkpointService CheckpointService + appMetrics *metrics.Metrics + networkPassphrase string + getLedgersLimit int + ledgerIndexer *indexer.Indexer + archive historyarchive.ArchiveInterface + backfillFlushSize uint32 + backfillWorkers uint32 + knownContractIDs set.Set[string] + contractMetadataService ContractMetadataService + protocolProcessors map[string]ProtocolProcessor + protocolValidators []ProtocolValidator + wasmSpecExtractor WasmSpecExtractor } func NewIngestService(cfg IngestServiceConfig) (*ingestService, error) { @@ -140,14 +138,12 @@ func NewIngestService(cfg IngestServiceConfig) (*ingestService, error) { ledgerIndexerPool := pond.NewPool(0) cfg.Metrics.RegisterPoolMetrics("ledger_indexer", ledgerIndexerPool) - // Create backfill pool with bounded size to control memory usage. + // Bound backfill worker concurrency to control memory usage. // Default to NumCPU if not specified. backfillWorkers := cfg.BackfillWorkers if backfillWorkers <= 0 { backfillWorkers = runtime.NumCPU() } - backfillPool := pond.NewPool(backfillWorkers) - cfg.Metrics.RegisterPoolMetrics("backfill", backfillPool) // Build protocol processor map from slice for i, p := range cfg.ProtocolProcessors { @@ -163,29 +159,28 @@ func NewIngestService(cfg IngestServiceConfig) (*ingestService, error) { } return &ingestService{ - ingestionMode: cfg.IngestionMode, - models: cfg.Models, - oldestLedgerCursorName: cfg.OldestLedgerCursorName, - advisoryLockID: generateAdvisoryLockID(cfg.Network), - appTracker: cfg.AppTracker, - rpcService: cfg.RPCService, - ledgerBackend: cfg.LedgerBackend, - ledgerBackendFactory: cfg.LedgerBackendFactory, - tokenIngestionService: cfg.TokenIngestionService, - checkpointService: cfg.CheckpointService, - appMetrics: cfg.Metrics, - networkPassphrase: cfg.NetworkPassphrase, - getLedgersLimit: cfg.GetLedgersLimit, - ledgerIndexer: indexer.NewIndexer(cfg.NetworkPassphrase, ledgerIndexerPool, cfg.Metrics.Ingestion), - contractMetadataService: cfg.ContractMetadataService, - protocolValidators: cfg.ProtocolValidators, - wasmSpecExtractor: cfg.WasmSpecExtractor, - archive: cfg.Archive, - backfillPool: backfillPool, - backfillBatchSize: uint32(cfg.BackfillBatchSize), - backfillDBInsertBatchSize: uint32(cfg.BackfillDBInsertBatchSize), - knownContractIDs: set.NewSet[string](), - protocolProcessors: ppMap, + ingestionMode: cfg.IngestionMode, + models: cfg.Models, + oldestLedgerCursorName: cfg.OldestLedgerCursorName, + advisoryLockID: generateAdvisoryLockID(cfg.Network), + appTracker: cfg.AppTracker, + rpcService: cfg.RPCService, + ledgerBackend: cfg.LedgerBackend, + ledgerBackendFactory: cfg.LedgerBackendFactory, + tokenIngestionService: cfg.TokenIngestionService, + checkpointService: cfg.CheckpointService, + appMetrics: cfg.Metrics, + networkPassphrase: cfg.NetworkPassphrase, + getLedgersLimit: cfg.GetLedgersLimit, + ledgerIndexer: indexer.NewIndexer(cfg.NetworkPassphrase, ledgerIndexerPool, cfg.Metrics.Ingestion), + contractMetadataService: cfg.ContractMetadataService, + protocolValidators: cfg.ProtocolValidators, + wasmSpecExtractor: cfg.WasmSpecExtractor, + archive: cfg.Archive, + backfillFlushSize: uint32(cfg.BackfillFlushSize), + backfillWorkers: uint32(backfillWorkers), + knownContractIDs: set.NewSet[string](), + protocolProcessors: ppMap, }, nil } @@ -203,9 +198,15 @@ func (m *ingestService) Run(ctx context.Context, startLedger uint32, endLedger u } } -// processLedger processes a single ledger - gets the transactions and processes them using indexer processors. +// processLedger processes a single ledger using the shared (live) indexer. func (m *ingestService) processLedger(ctx context.Context, ledgerMeta xdr.LedgerCloseMeta, buffer *indexer.IndexerBuffer) error { - participantCount, err := indexer.ProcessLedger(ctx, m.networkPassphrase, ledgerMeta, m.ledgerIndexer, buffer) + return m.processLedgerWith(ctx, m.ledgerIndexer, ledgerMeta, buffer) +} + +// processLedgerWith processes a single ledger with the given indexer. Backfill passes a +// per-worker serial indexer so the only parallelism axis is the worker pool itself. +func (m *ingestService) processLedgerWith(ctx context.Context, idx *indexer.Indexer, ledgerMeta xdr.LedgerCloseMeta, buffer *indexer.IndexerBuffer) error { + participantCount, err := indexer.ProcessLedger(ctx, m.networkPassphrase, ledgerMeta, idx, buffer) if err != nil { return fmt.Errorf("processing ledger %d: %w", ledgerMeta.LedgerSequence(), err) } diff --git a/internal/services/ingest_backfill.go b/internal/services/ingest_backfill.go index a632225b0..3b27c92fb 100644 --- a/internal/services/ingest_backfill.go +++ b/internal/services/ingest_backfill.go @@ -3,54 +3,16 @@ package services import ( "context" "fmt" - "sync" "time" - "github.com/jackc/pgx/v5" - "github.com/jackc/pgx/v5/pgxpool" - "github.com/stellar/go-stellar-sdk/ingest/ledgerbackend" "github.com/stellar/go-stellar-sdk/support/log" - "github.com/stellar/go-stellar-sdk/xdr" "github.com/stellar/wallet-backend/internal/data" - "github.com/stellar/wallet-backend/internal/db" - "github.com/stellar/wallet-backend/internal/indexer" - "github.com/stellar/wallet-backend/internal/utils" ) -// BackfillBatch represents a contiguous range of ledgers to process as a unit. -type BackfillBatch struct { - StartLedger uint32 - EndLedger uint32 -} - -// BackfillResult tracks the outcome of processing a single batch. -type BackfillResult struct { - Batch BackfillBatch - LedgersCount int - Duration time.Duration - Error error - StartTime time.Time // First ledger close time in batch (for compression) - EndTime time.Time // Last ledger close time in batch (for compression) -} - -// analyzeBatchResults aggregates backfill batch results and logs any failures. -func analyzeBatchResults(ctx context.Context, results []BackfillResult) int { - numFailed := 0 - for _, result := range results { - if result.Error != nil { - numFailed++ - log.Ctx(ctx).Errorf("Batch [%d-%d] failed: %v", - result.Batch.StartLedger, result.Batch.EndLedger, result.Error) - } - } - log.Ctx(ctx).Infof("Backfilling completed: %d/%d batches failed", numFailed, len(results)) - return numFailed -} - -// startBackfilling fills gaps within the already-ingested range [startLedger, endLedger], -// processing them in parallel batches. Data is inserted uncompressed and a progressive -// recompressor compresses chunks as contiguous batches complete. +// startBackfilling fills gaps within the already-ingested range [startLedger, endLedger] +// using a streaming fetch → process → flush pipeline. Data is inserted uncompressed and a +// background compressor compresses chunks as the flusher reports contiguous progress. func (m *ingestService) startBackfilling(ctx context.Context, startLedger, endLedger uint32) error { if startLedger > endLedger { return fmt.Errorf("start ledger cannot be greater than end ledger") @@ -73,27 +35,22 @@ func (m *ingestService) startBackfilling(ctx context.Context, startLedger, endLe return nil } - backfillBatches := m.splitGapsIntoBatches(gaps) - - // Recompress chunks as contiguous batches complete rather than waiting until the end. - tables := []string{ + compressor := newBackfillCompressor(ctx, m.models.DB, []string{ "transactions", "transactions_accounts", "operations", "operations_accounts", "state_changes", - } - recompressor := newProgressiveRecompressor(ctx, m.models.DB, tables, len(backfillBatches)) - - startTime := time.Now() - results := m.processBackfillBatchesParallel(ctx, backfillBatches, recompressor) - duration := time.Since(startTime) - - analyzeBatchResults(ctx, results) + }) - // Wait for progressive compression to finish. Compression proceeds even if some - // batches failed — already-compressed chunks contain valid data and compress_chunk - // is idempotent. - recompressor.Wait() + start := time.Now() + for _, gap := range gaps { + if err := m.runGapPipeline(ctx, gap, compressor); err != nil { + compressor.Wait() + return fmt.Errorf("backfilling gap [%d-%d]: %w", gap.GapStart, gap.GapEnd, err) + } + log.Ctx(ctx).Infof("Backfilled gap [%d-%d]", gap.GapStart, gap.GapEnd) + } - log.Ctx(ctx).Infof("Backfilling completed in %v: %d batches", duration, len(backfillBatches)) + compressor.Wait() + log.Ctx(ctx).Infof("Backfilling completed in %v: %d gaps", time.Since(start), len(gaps)) return nil } @@ -157,371 +114,3 @@ func (m *ingestService) calculateBackfillGaps(ctx context.Context, startLedger, return newGaps, nil } - -// splitGapsIntoBatches divides ledger gaps into fixed-size batches for parallel processing. -func (m *ingestService) splitGapsIntoBatches(gaps []data.LedgerRange) []BackfillBatch { - var batches []BackfillBatch - - for _, gap := range gaps { - start := gap.GapStart - for start <= gap.GapEnd { - end := min(start+m.backfillBatchSize-1, gap.GapEnd) - batches = append(batches, BackfillBatch{ - StartLedger: start, - EndLedger: end, - }) - start = end + 1 - } - } - - return batches -} - -// processBackfillBatchesParallel processes backfill batches in parallel using a worker pool. -// Data is inserted uncompressed; the progressive recompressor compresses chunks via -// compress_chunk() as contiguous batches complete. -func (m *ingestService) processBackfillBatchesParallel(ctx context.Context, batches []BackfillBatch, recompressor *progressiveRecompressor) []BackfillResult { - results := make([]BackfillResult, len(batches)) - group := m.backfillPool.NewGroupContext(ctx) - - for i, batch := range batches { - group.Submit(func() { - results[i] = m.processSingleBatch(ctx, batch, i, len(batches)) - if recompressor != nil && results[i].Error == nil { - recompressor.MarkDone(i, results[i].StartTime, results[i].EndTime) - } - }) - } - - if err := group.Wait(); err != nil { - log.Ctx(ctx).Warnf("Backfill batch group wait returned error: %v", err) - } - - return results -} - -// processSingleBatch processes a single backfill batch with its own ledger backend. -func (m *ingestService) processSingleBatch(ctx context.Context, batch BackfillBatch, batchIndex, totalBatches int) BackfillResult { - start := time.Now() - result := BackfillResult{Batch: batch} - - // Setup backend - backend, err := m.setupBatchBackend(ctx, batch) - if err != nil { - result.Error = err - result.Duration = time.Since(start) - return result - } - defer func() { - if closeErr := backend.Close(); closeErr != nil { - log.Ctx(ctx).Warnf("Error closing ledger backend for batch [%d-%d]: %v", batch.StartLedger, batch.EndLedger, closeErr) - } - }() - - // Process all ledgers in batch (cursor is updated atomically with the final flush) - ledgersCount, batchStartTime, batchEndTime, err := m.processLedgersInBatch(ctx, backend, batch) - result.LedgersCount = ledgersCount - result.StartTime = batchStartTime - result.EndTime = batchEndTime - if err != nil { - result.Error = err - result.Duration = time.Since(start) - return result - } - - m.appMetrics.Ingestion.OldestLedger.Set(float64(batch.StartLedger)) - - result.Duration = time.Since(start) - log.Ctx(ctx).Infof("Batch %d/%d [%d - %d] completed: %d ledgers in %v", - batchIndex+1, totalBatches, batch.StartLedger, batch.EndLedger, result.LedgersCount, result.Duration) - - return result -} - -// setupBatchBackend creates and prepares a ledger backend for a batch range. -// Caller is responsible for calling Close() on the returned backend. -func (m *ingestService) setupBatchBackend(ctx context.Context, batch BackfillBatch) (ledgerbackend.LedgerBackend, error) { - backend, err := m.ledgerBackendFactory(ctx) - if err != nil { - return nil, fmt.Errorf("creating ledger backend: %w", err) - } - - ledgerRange := ledgerbackend.BoundedRange(batch.StartLedger, batch.EndLedger) - if err := backend.PrepareRange(ctx, ledgerRange); err != nil { - return nil, fmt.Errorf("preparing backend range: %w", err) - } - - return backend, nil -} - -// flushBatchBufferWithRetry persists buffered data to the database within a transaction. -// If updateCursorTo is non-nil, it also updates the oldest cursor atomically. -func (m *ingestService) flushBatchBufferWithRetry(ctx context.Context, buffer *indexer.IndexerBuffer, updateCursorTo *uint32) error { - var lastErr error - for attempt := 0; attempt < maxIngestProcessedDataRetries; attempt++ { - select { - case <-ctx.Done(): - return fmt.Errorf("context cancelled: %w", ctx.Err()) - default: - } - - err := db.RunInTransaction(ctx, m.models.DB, func(dbTx pgx.Tx) error { - // Disable synchronous commit for this transaction only — safe for backfill - // since data can be re-ingested if a crash occurs before WAL flush. - if _, txErr := dbTx.Exec(ctx, "SET LOCAL synchronous_commit = off"); txErr != nil { - return fmt.Errorf("setting synchronous_commit=off: %w", txErr) - } - if _, _, err := m.insertIntoDB(ctx, dbTx, buffer); err != nil { - return fmt.Errorf("inserting processed data into db: %w", err) - } - // Update cursor atomically with data insertion if requested - if updateCursorTo != nil { - if err := m.models.IngestStore.UpdateMin(ctx, dbTx, m.oldestLedgerCursorName, *updateCursorTo); err != nil { - return fmt.Errorf("updating oldest cursor: %w", err) - } - } - return nil - }) - if err == nil { - return nil - } - lastErr = err - m.appMetrics.Ingestion.RetriesTotal.WithLabelValues("batch_flush").Inc() - if attempt == maxIngestProcessedDataRetries-1 { - break - } - - backoff := time.Duration(1< maxIngestProcessedDataRetryBackoff { - backoff = maxIngestProcessedDataRetryBackoff - } - log.Ctx(ctx).Warnf("Error flushing batch buffer (attempt %d/%d): %v, retrying in %v...", - attempt+1, maxIngestProcessedDataRetries, lastErr, backoff) - - select { - case <-ctx.Done(): - return fmt.Errorf("context cancelled during backoff: %w", ctx.Err()) - case <-time.After(backoff): - } - } - m.appMetrics.Ingestion.RetryExhaustionsTotal.WithLabelValues("batch_flush").Inc() - return lastErr -} - -// processLedgersInBatch processes all ledgers in a batch, flushing to DB periodically. -// The oldest-ledger cursor is updated atomically with the final data flush. -// Returns the count of ledgers processed and the batch's ledger close-time range. -func (m *ingestService) processLedgersInBatch( - ctx context.Context, - backend ledgerbackend.LedgerBackend, - batch BackfillBatch, -) (int, time.Time, time.Time, error) { - batchBuffer := indexer.NewIndexerBuffer() - ledgersInBuffer := uint32(0) - ledgersProcessed := 0 - var startTime, endTime time.Time - - for ledgerSeq := batch.StartLedger; ledgerSeq <= batch.EndLedger; ledgerSeq++ { - ledgerMeta, err := utils.RetryWithBackoff(ctx, maxLedgerFetchRetries, maxRetryBackoff, - func(ctx context.Context) (xdr.LedgerCloseMeta, error) { - return backend.GetLedger(ctx, ledgerSeq) - }, - func(attempt int, err error, backoff time.Duration) { - log.Ctx(ctx).Warnf("Error fetching ledger %d (attempt %d/%d): %v, retrying in %v...", - ledgerSeq, attempt+1, maxLedgerFetchRetries, err, backoff) - }, - ) - if err != nil { - return ledgersProcessed, startTime, endTime, fmt.Errorf("getting ledger %d: %w", ledgerSeq, err) - } - - // Track time range for compression - ledgerTime := ledgerMeta.ClosedAt() - if startTime.IsZero() { - startTime = ledgerTime - } - endTime = ledgerTime - - if err := m.processLedger(ctx, ledgerMeta, batchBuffer); err != nil { - return ledgersProcessed, startTime, endTime, fmt.Errorf("processing ledger %d: %w", ledgerSeq, err) - } - ledgersProcessed++ - ledgersInBuffer++ - - // Flush buffer periodically to control memory usage (intermediate flushes, no cursor update) - if ledgersInBuffer >= m.backfillDBInsertBatchSize { - if err := m.flushBatchBufferWithRetry(ctx, batchBuffer, nil); err != nil { - return ledgersProcessed, startTime, endTime, err - } - batchBuffer.Clear() - ledgersInBuffer = 0 - } - } - - // Final flush with cursor update - if ledgersInBuffer > 0 { - if err := m.flushBatchBufferWithRetry(ctx, batchBuffer, &batch.StartLedger); err != nil { - return ledgersProcessed, startTime, endTime, err - } - } else { - // All data was flushed in intermediate batches, but we still need to update the cursor. - // This happens when ledgersInBuffer == 0 (exact multiple of batch size). - if err := m.updateOldestCursor(ctx, batch.StartLedger); err != nil { - return ledgersProcessed, startTime, endTime, err - } - } - - return ledgersProcessed, startTime, endTime, nil -} - -// updateOldestCursor updates the oldest ledger cursor to the given ledger. -func (m *ingestService) updateOldestCursor(ctx context.Context, ledgerSeq uint32) error { - err := db.RunInTransaction(ctx, m.models.DB, func(dbTx pgx.Tx) error { - return m.models.IngestStore.UpdateMin(ctx, dbTx, m.oldestLedgerCursorName, ledgerSeq) - }) - if err != nil { - return fmt.Errorf("updating oldest ledger cursor: %w", err) - } - return nil -} - -// progressiveRecompressor compresses uncompressed TimescaleDB chunks as they become safe during backfill. -// Tracks batch completion via a watermark to determine when chunks are fully written. -type progressiveRecompressor struct { - pool *pgxpool.Pool - tables []string - ctx context.Context - - mu sync.Mutex - completed []bool - endTimes []time.Time - watermarkIdx int // index of highest contiguous completed batch (-1 = none) - globalStart time.Time // lower bound for chunk queries (batch 0's StartTime) - - triggerCh chan time.Time // safeEnd for recompression window - done chan struct{} -} - -// newProgressiveRecompressor creates a compressor that progressively compresses uncompressed chunks -// as contiguous batches complete. Starts a background goroutine for compression work. -func newProgressiveRecompressor(ctx context.Context, pool *pgxpool.Pool, tables []string, totalBatches int) *progressiveRecompressor { - r := &progressiveRecompressor{ - pool: pool, - tables: tables, - ctx: ctx, - completed: make([]bool, totalBatches), - endTimes: make([]time.Time, totalBatches), - watermarkIdx: -1, - triggerCh: make(chan time.Time, totalBatches), - done: make(chan struct{}), - } - go r.runCompression() - return r -} - -// MarkDone records a batch as complete and advances the watermark if possible. -// If the watermark advances, triggers recompression of chunks in the safe window. -func (r *progressiveRecompressor) MarkDone(batchIdx int, startTime, endTime time.Time) { - var safeEnd time.Time - r.mu.Lock() - r.completed[batchIdx] = true - r.endTimes[batchIdx] = endTime - - // Record global start from batch 0 (earliest time boundary for queries) - if batchIdx == 0 { - r.globalStart = startTime - } - - // Advance watermark past contiguous completed batches - oldWatermark := r.watermarkIdx - for r.watermarkIdx+1 < len(r.completed) && r.completed[r.watermarkIdx+1] { - r.watermarkIdx++ - } - - sendToChannel := (r.watermarkIdx > oldWatermark) - if sendToChannel { - safeEnd = r.endTimes[r.watermarkIdx] - } - r.mu.Unlock() - - // If watermark advanced then we trigger recompression outside the lock - if sendToChannel { - r.triggerCh <- safeEnd - } -} - -// Wait closes the trigger channel and waits for background compression to finish. -func (r *progressiveRecompressor) Wait() { - close(r.triggerCh) - <-r.done -} - -// runCompression processes compression triggers in the background. -// For each safe window, queries and compresses uncompressed chunks per table. -func (r *progressiveRecompressor) runCompression() { - defer close(r.done) - - totalCompressed := 0 - for safeEnd := range r.triggerCh { - for _, table := range r.tables { - count := r.compressTableChunks(table, safeEnd) - totalCompressed += count - } - } - - log.Ctx(r.ctx).Infof("Progressive compression complete: %d total chunks compressed", totalCompressed) -} - -// compressTableChunks compresses uncompressed chunks for a single table within the safe window. -// Queries chunks where range_end falls within (globalStart, safeEnd) to catch the leading -// boundary chunk that overlaps globalStart. -func (r *progressiveRecompressor) compressTableChunks(table string, safeEnd time.Time) int { - rows, err := r.pool.Query(r.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 - AND c.range_end > $3::timestamptz`, - table, safeEnd, r.globalStart) - if err != nil { - log.Ctx(r.ctx).Warnf("Failed to get chunks for %s: %v", table, err) - return 0 - } - defer rows.Close() - - var chunks []string - for rows.Next() { - var chunk string - if err := rows.Scan(&chunk); err != nil { - log.Ctx(r.ctx).Warnf("Failed to scan chunk row for table %s: %v", table, err) - continue - } - chunks = append(chunks, chunk) - } - - compressed := 0 - for _, chunk := range chunks { - select { - case <-r.ctx.Done(): - log.Ctx(r.ctx).Warnf("Compression cancelled for %s after %d chunks", table, compressed) - return compressed - default: - } - - _, err := r.pool.Exec(r.ctx, `SELECT compress_chunk($1::regclass)`, chunk) - if err != nil { - log.Ctx(r.ctx).Warnf("Failed to compress chunk %s: %v", chunk, err) - continue - } - compressed++ - log.Ctx(r.ctx).Debugf("Compressed chunk for %s: %s", table, chunk) - } - - if compressed > 0 { - log.Ctx(r.ctx).Infof("Compressed %d chunks for table %s (window end: %s)", - compressed, table, safeEnd.Format(time.RFC3339)) - } - - return compressed -} diff --git a/internal/services/ingest_backfill_test.go b/internal/services/ingest_backfill_test.go deleted file mode 100644 index afbcf4c73..000000000 --- a/internal/services/ingest_backfill_test.go +++ /dev/null @@ -1,176 +0,0 @@ -// Tests for progressive recompression watermark logic during historical backfill. -package services - -import ( - "testing" - "time" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// newTestRecompressor creates a progressiveRecompressor for testing watermark logic. -// No background goroutine is started; triggerCh is buffered for direct inspection. -func newTestRecompressor(totalBatches int) *progressiveRecompressor { - return &progressiveRecompressor{ - completed: make([]bool, totalBatches), - endTimes: make([]time.Time, totalBatches), - watermarkIdx: -1, - triggerCh: make(chan time.Time, totalBatches), - } -} - -// drainWindows reads all available safeEnd values from triggerCh without blocking. -func drainWindows(r *progressiveRecompressor) []time.Time { - var windows []time.Time - for { - select { - case w := <-r.triggerCh: - windows = append(windows, w) - default: - return windows - } - } -} - -func Test_progressiveRecompressor_MarkDone_sequential(t *testing.T) { - r := newTestRecompressor(5) - base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) - - // Complete batches 0-4 in order - for i := 0; i < 5; i++ { - startTime := base.Add(time.Duration(i) * time.Hour) - endTime := base.Add(time.Duration(i+1) * time.Hour) - r.MarkDone(i, startTime, endTime) - } - - windows := drainWindows(r) - // Each call advances the watermark — expect 5 windows - require.Len(t, windows, 5) - - // First window: safeEnd = batch 0 endTime - assert.Equal(t, base.Add(1*time.Hour), windows[0]) - - // Second window: safeEnd = batch 1 endTime - assert.Equal(t, base.Add(2*time.Hour), windows[1]) - - // Last window: safeEnd = batch 4 endTime - assert.Equal(t, base.Add(5*time.Hour), windows[4]) - - // Verify globalStart set from batch 0 - assert.Equal(t, base, r.globalStart) -} - -func Test_progressiveRecompressor_MarkDone_outOfOrder(t *testing.T) { - r := newTestRecompressor(5) - base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) - endTime := func(i int) time.Time { return base.Add(time.Duration(i+1) * time.Hour) } - startTime := func(i int) time.Time { return base.Add(time.Duration(i) * time.Hour) } - - // Complete batch 2 first — watermark can't advance (batch 0 not done) - r.MarkDone(2, startTime(2), endTime(2)) - windows := drainWindows(r) - assert.Empty(t, windows) - assert.Equal(t, -1, r.watermarkIdx) - - // Complete batch 4 — still no advancement - r.MarkDone(4, startTime(4), endTime(4)) - windows = drainWindows(r) - assert.Empty(t, windows) - - // Complete batch 0 — watermark advances to 0 only (batch 1 missing) - r.MarkDone(0, startTime(0), endTime(0)) - windows = drainWindows(r) - require.Len(t, windows, 1) - assert.Equal(t, endTime(0), windows[0]) - assert.Equal(t, 0, r.watermarkIdx) - - // Complete batch 1 — watermark jumps from 0 to 2 (batch 2 was already done) - r.MarkDone(1, startTime(1), endTime(1)) - windows = drainWindows(r) - require.Len(t, windows, 1) - assert.Equal(t, endTime(2), windows[0]) // safeEnd = batch 2's endTime - assert.Equal(t, 2, r.watermarkIdx) - - // Complete batch 3 — watermark jumps from 2 to 4 (batch 4 was already done) - r.MarkDone(3, startTime(3), endTime(3)) - windows = drainWindows(r) - require.Len(t, windows, 1) - assert.Equal(t, endTime(4), windows[0]) - assert.Equal(t, 4, r.watermarkIdx) -} - -func Test_progressiveRecompressor_MarkDone_failedBatchBlocksWatermark(t *testing.T) { - r := newTestRecompressor(5) - base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) - endTime := func(i int) time.Time { return base.Add(time.Duration(i+1) * time.Hour) } - startTime := func(i int) time.Time { return base.Add(time.Duration(i) * time.Hour) } - - // Complete batches 0 and 1 - r.MarkDone(0, startTime(0), endTime(0)) - r.MarkDone(1, startTime(1), endTime(1)) - _ = drainWindows(r) // consume windows - - // Batch 2 fails (never call MarkDone for it) - - // Complete batches 3 and 4 - r.MarkDone(3, startTime(3), endTime(3)) - r.MarkDone(4, startTime(4), endTime(4)) - - // No new windows — watermark stuck at 1 - windows := drainWindows(r) - assert.Empty(t, windows) - assert.Equal(t, 1, r.watermarkIdx) -} - -func Test_progressiveRecompressor_MarkDone_singleBatch(t *testing.T) { - r := newTestRecompressor(1) - start := time.Date(2025, 6, 1, 0, 0, 0, 0, time.UTC) - end := time.Date(2025, 6, 1, 1, 0, 0, 0, time.UTC) - - r.MarkDone(0, start, end) - - windows := drainWindows(r) - require.Len(t, windows, 1) - assert.Equal(t, end, windows[0]) - assert.Equal(t, 0, r.watermarkIdx) - assert.Equal(t, start, r.globalStart) -} - -func Test_progressiveRecompressor_MarkDone_globalStartSetFromBatchZero(t *testing.T) { - r := newTestRecompressor(3) - base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) - - // Complete batch 1 first — globalStart should NOT be set - r.MarkDone(1, base.Add(1*time.Hour), base.Add(2*time.Hour)) - assert.True(t, r.globalStart.IsZero()) - - // Complete batch 0 — globalStart should be set to batch 0's startTime - r.MarkDone(0, base, base.Add(1*time.Hour)) - assert.Equal(t, base, r.globalStart) - - // Complete batch 2 — globalStart unchanged - r.MarkDone(2, base.Add(2*time.Hour), base.Add(3*time.Hour)) - assert.Equal(t, base, r.globalStart) -} - -func Test_progressiveRecompressor_MarkDone_allSimultaneous(t *testing.T) { - r := newTestRecompressor(4) - base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) - endTime := func(i int) time.Time { return base.Add(time.Duration(i+1) * time.Hour) } - startTime := func(i int) time.Time { return base.Add(time.Duration(i) * time.Hour) } - - // Complete all batches in reverse order except batch 0 - r.MarkDone(3, startTime(3), endTime(3)) - r.MarkDone(2, startTime(2), endTime(2)) - r.MarkDone(1, startTime(1), endTime(1)) - windows := drainWindows(r) - assert.Empty(t, windows) // Nothing yet — batch 0 missing - - // Complete batch 0 — watermark jumps from -1 to 3 in one step - r.MarkDone(0, startTime(0), endTime(0)) - windows = drainWindows(r) - require.Len(t, windows, 1) - assert.Equal(t, endTime(3), windows[0]) // safeEnd = last batch - assert.Equal(t, 3, r.watermarkIdx) -} diff --git a/internal/services/ingest_test.go b/internal/services/ingest_test.go index 489454664..8349307df 100644 --- a/internal/services/ingest_test.go +++ b/internal/services/ingest_test.go @@ -12,7 +12,6 @@ import ( "github.com/jackc/pgx/v5/pgxpool" "github.com/prometheus/client_golang/prometheus" "github.com/stellar/go-stellar-sdk/ingest/ledgerbackend" - "github.com/stellar/go-stellar-sdk/keypair" "github.com/stellar/go-stellar-sdk/network" "github.com/stellar/go-stellar-sdk/xdr" "github.com/stretchr/testify/assert" @@ -29,13 +28,6 @@ import ( "github.com/stellar/wallet-backend/internal/utils" ) -var ( - testKP1 = keypair.MustRandom() - testKP2 = keypair.MustRandom() - testAddr1 = testKP1.Address() - testAddr2 = testKP2.Address() -) - const ( defaultGetLedgersLimit = 50 @@ -46,9 +38,6 @@ const ( flushTxHash4 = "f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f4f404" txHash1 = "1111111111111111111111111111111111111111111111111111111111111111" txHash2 = "2222222222222222222222222222222222222222222222222222222222222222" - - // Test fixtures for ledger metadata - ledgerMetadataWith1Tx = "AAAAAQAAAAD8G2qemHnBKFkbq90RTagxAypNnA7DXDc63Giipq9mNwAAABYLEZ5DrTv6njXTOAFEdOO0yeLtJjCRyH4ryJkgpRh7VPJvwbisrc9A0yzFxxCdkICgB3Gv7qHOi8ZdsK2CNks2AAAAAGhTTAsAAAAAAAAAAQAAAACoJM0YvJ11Bk0pmltbrKQ7w6ovMmk4FT2ML5u1y23wMwAAAEAunZtorOSbnRpgnykoDe4kzAvLwNXefncy1R/1ynBWyDv0DfdnqJ6Hcy/0AJf6DkBZlRayg775h3HjV0GKF/oPua7l8wkLlJBtSk1kRDt55qSf6btSrgcupB/8bnpJfUUgZJ76saUrj29HukYHS1bq7SyuoCAY+5F9iBYTmW1G9QAAEX4N4Lazp2QAAAAAAAMtS3veAAAAAAAAAAAAAAAMAAAAZABMS0AAAADIXukLfWC53MCmzxKd/+LBbaYxQkgxATFDLI3hWj7EqWgAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAELEZ5DrTv6njXTOAFEdOO0yeLtJjCRyH4ryJkgpRh7VAAAAAIAAAAAAAAAAQAAAAAAAAABAAAAAAAAAGQAAAABAAAAAgAAAADg4mtiLKjJVgrmOpO9+Ff3XAmnycHyNUKu/v9KhHevAAAAAGQAAA7FAAAAGgAAAAAAAAAAAAAAAQAAAAAAAAABAAAAALvqzdVyRxgBMcLzbw1wNWcJYHPNPok1GdVSgmy4sjR2AAAAAVVTREMAAAAA4OJrYiyoyVYK5jqTvfhX91wJp8nB8jVCrv7/SoR3rwAAAAACVAvkAAAAAAAAAAABhHevAAAAAEDq2yIDzXUoLboBHQkbr8U2oKqLzf0gfpwXbmRPLB6Ek3G8uCEYyry1vt5Sb+LCEd81fefFQcQN0nydr1FmiXcDAAAAAAAAAAAAAAABXFSiWcxpDRa8frBs1wbEaMUw4hMe7ctFtdw3Ci73IEwAAAAAAAAAZAAAAAAAAAABAAAAAAAAAAEAAAAAAAAAAAAAAAIAAAADAAARfQAAAAAAAAAA4OJrYiyoyVYK5jqTvfhX91wJp8nB8jVCrv7/SoR3rwAAAAAukO3GPAAADsUAAAAZAAAAAAAAAAAAAAAAAAAAAAEAAAAAAAAAAAAAAQAAAAAAAAAAAAAAAAAAAAAAAAACAAAAAAAAAAAAAAAAAAAAAwAAAAAAABF9AAAAAGhTTAYAAAAAAAAAAQAAEX4AAAAAAAAAAODia2IsqMlWCuY6k734V/dcCafJwfI1Qq7+/0qEd68AAAAALpDtxdgAAA7FAAAAGQAAAAAAAAAAAAAAAAAAAAABAAAAAAAAAAAAAAEAAAAAAAAAAAAAAAAAAAAAAAAAAgAAAAAAAAAAAAAAAAAAAAMAAAAAAAARfQAAAABoU0wGAAAAAAAAAAMAAAAAAAAAAgAAAAMAABF+AAAAAAAAAADg4mtiLKjJVgrmOpO9+Ff3XAmnycHyNUKu/v9KhHevAAAAAC6Q7cXYAAAOxQAAABkAAAAAAAAAAAAAAAAAAAAAAQAAAAAAAAAAAAABAAAAAAAAAAAAAAAAAAAAAAAAAAIAAAAAAAAAAAAAAAAAAAADAAAAAAAAEX0AAAAAaFNMBgAAAAAAAAABAAARfgAAAAAAAAAA4OJrYiyoyVYK5jqTvfhX91wJp8nB8jVCrv7/SoR3rwAAAAAukO3F2AAADsUAAAAaAAAAAAAAAAAAAAAAAAAAAAEAAAAAAAAAAAAAAQAAAAAAAAAAAAAAAAAAAAAAAAACAAAAAAAAAAAAAAAAAAAAAwAAAAAAABF+AAAAAGhTTAsAAAAAAAAAAQAAAAIAAAADAAARcwAAAAEAAAAAu+rN1XJHGAExwvNvDXA1Zwlgc80+iTUZ1VKCbLiyNHYAAAABVVNEQwAAAADg4mtiLKjJVgrmOpO9+Ff3XAmnycHyNUKu/v9KhHevAAAAAAlQL5AAf/////////8AAAABAAAAAAAAAAAAAAABAAARfgAAAAEAAAAAu+rN1XJHGAExwvNvDXA1Zwlgc80+iTUZ1VKCbLiyNHYAAAABVVNEQwAAAADg4mtiLKjJVgrmOpO9+Ff3XAmnycHyNUKu/v9KhHevAAAAAAukO3QAf/////////8AAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA8RxEAAAAAAAAAAA==" ) func TestLagLedgers(t *testing.T) { @@ -131,99 +120,6 @@ func Test_generateAdvisoryLockID(t *testing.T) { } } -func Test_ingestService_splitGapsIntoBatches(t *testing.T) { - svc := &ingestService{} - - testCases := []struct { - name string - gaps []data.LedgerRange - batchSize uint32 - expected []BackfillBatch - }{ - { - name: "empty_gaps", - gaps: []data.LedgerRange{}, - batchSize: 100, - expected: nil, - }, - { - name: "single_gap_smaller_than_batch", - gaps: []data.LedgerRange{ - {GapStart: 100, GapEnd: 150}, - }, - batchSize: 200, - expected: []BackfillBatch{ - {StartLedger: 100, EndLedger: 150}, - }, - }, - { - name: "single_gap_larger_than_batch", - gaps: []data.LedgerRange{ - {GapStart: 100, GapEnd: 399}, - }, - batchSize: 100, - expected: []BackfillBatch{ - {StartLedger: 100, EndLedger: 199}, - {StartLedger: 200, EndLedger: 299}, - {StartLedger: 300, EndLedger: 399}, - }, - }, - { - name: "single_gap_exact_batch_size", - gaps: []data.LedgerRange{ - {GapStart: 100, GapEnd: 199}, - }, - batchSize: 100, - expected: []BackfillBatch{ - {StartLedger: 100, EndLedger: 199}, - }, - }, - { - name: "multiple_gaps", - gaps: []data.LedgerRange{ - {GapStart: 100, GapEnd: 149}, - {GapStart: 300, GapEnd: 349}, - }, - batchSize: 100, - expected: []BackfillBatch{ - {StartLedger: 100, EndLedger: 149}, - {StartLedger: 300, EndLedger: 349}, - }, - }, - { - name: "multiple_gaps_with_splits", - gaps: []data.LedgerRange{ - {GapStart: 100, GapEnd: 249}, - {GapStart: 500, GapEnd: 599}, - }, - batchSize: 100, - expected: []BackfillBatch{ - {StartLedger: 100, EndLedger: 199}, - {StartLedger: 200, EndLedger: 249}, - {StartLedger: 500, EndLedger: 599}, - }, - }, - { - name: "single_ledger_gap", - gaps: []data.LedgerRange{ - {GapStart: 100, GapEnd: 100}, - }, - batchSize: 100, - expected: []BackfillBatch{ - {StartLedger: 100, EndLedger: 100}, - }, - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - svc.backfillBatchSize = tc.batchSize - result := svc.splitGapsIntoBatches(tc.gaps) - assert.Equal(t, tc.expected, result) - }) - } -} - func Test_ingestService_calculateBackfillGaps(t *testing.T) { dbt := dbtest.Open(t) defer dbt.Close() @@ -418,27 +314,25 @@ func Test_startBackfilling_Validation(t *testing.T) { defer dbConnectionPool.Close() testCases := []struct { - name string - startLedger uint32 - endLedger uint32 - latestIngested uint32 - expectValidationError bool - errorContains string + name string + startLedger uint32 + endLedger uint32 + latestIngested uint32 + errorContains string }{ { - name: "valid_range", - startLedger: 50, - endLedger: 80, - latestIngested: 100, - expectValidationError: false, + name: "valid_range_fails_fast_on_backend_error", + startLedger: 50, + endLedger: 80, + latestIngested: 100, + errorContains: "creating ledger backend", }, { - name: "end_exceeds_latest", - startLedger: 50, - endLedger: 150, - latestIngested: 100, - expectValidationError: true, - errorContains: "end ledger 150 cannot be greater than latest ingested ledger 100", + name: "end_exceeds_latest", + startLedger: 50, + endLedger: 150, + latestIngested: 100, + errorContains: "end ledger 150 cannot be greater than latest ingested ledger 100", }, } @@ -460,6 +354,15 @@ func Test_startBackfilling_Validation(t *testing.T) { fmt.Sprintf("%d", tc.latestIngested)) require.NoError(t, err) + // Required for the valid-range/fail-fast subtest: anchoring the oldest ingested + // ledger makes GetOldestLedger return latestIngested so a real gap forms and the + // backend factory is reached. + _, err = dbConnectionPool.Exec(ctx, + `INSERT INTO transactions (hash, to_id, fee_charged, result_code, ledger_number, ledger_created_at) + VALUES ('anchor_hash', 1, 100, 'TransactionResultCodeTxSuccess', $1, NOW())`, + tc.latestIngested) + require.NoError(t, err) + m := metrics.NewMetrics(prometheus.NewRegistry()) models, err := data.NewModels(dbConnectionPool, m.DB) @@ -471,8 +374,8 @@ func Test_startBackfilling_Validation(t *testing.T) { mockLedgerBackend := &LedgerBackendMock{} mockArchive := &HistoryArchiveMock{} - // Create a mock ledger backend factory that returns an error immediately - // This allows validation to pass but stops batch processing early + // Create a mock ledger backend factory that returns an error immediately. + // This lets validation + gap calculation pass, then the pipeline fails fast on backend creation. mockBackendFactory := func(ctx context.Context) (ledgerbackend.LedgerBackend, error) { return nil, fmt.Errorf("mock backend factory error") } @@ -490,19 +393,12 @@ func Test_startBackfilling_Validation(t *testing.T) { Network: network.TestNetworkPassphrase, NetworkPassphrase: network.TestNetworkPassphrase, Archive: mockArchive, - BackfillBatchSize: 100, }) require.NoError(t, err) err = svc.startBackfilling(ctx, tc.startLedger, tc.endLedger) - if tc.expectValidationError { - require.Error(t, err) - assert.Contains(t, err.Error(), tc.errorContains) - } else { - // For valid cases, validation passes but batch processing fails. - // Backfill logs batch failures and returns nil rather than erroring. - require.NoError(t, err) - } + require.Error(t, err) + assert.Contains(t, err.Error(), tc.errorContains) }) } } @@ -523,269 +419,6 @@ func setupDBCursors(t *testing.T, ctx context.Context, pool *pgxpool.Pool, lates } } -// ptrUint32 returns a pointer to the given uint32 value -func ptrUint32(v uint32) *uint32 { return &v } - -// createTestTransaction creates a transaction with required fields for testing. -func createTestTransaction(hash string, toID int64) types.Transaction { - now := time.Now() - return types.Transaction{ - Hash: types.HashBytea(hash), - ToID: toID, - FeeCharged: 100, - ResultCode: "TransactionResultCodeTxSuccess", - LedgerNumber: 1000, - LedgerCreatedAt: now, - IngestedAt: now, - } -} - -// createTestOperation creates an operation with required fields for testing. -func createTestOperation(id int64) types.Operation { - now := time.Now() - return types.Operation{ - ID: id, - OperationType: types.OperationTypePayment, - OperationXDR: types.XDRBytea([]byte("test_operation_xdr")), - LedgerNumber: 1000, - LedgerCreatedAt: now, - IngestedAt: now, - } -} - -// createTestStateChange creates a state change with required fields for testing. -func createTestStateChange(toID int64, accountID string, opID int64) types.StateChange { - now := time.Now() - reason := types.StateChangeReasonCredit - return types.StateChange{ - ToID: toID, - StateChangeID: 1, - StateChangeCategory: types.StateChangeCategoryBalance, - StateChangeReason: reason, - AccountID: types.AddressBytea(accountID), - OperationID: opID, - LedgerNumber: 1000, - LedgerCreatedAt: now, - IngestedAt: now, - } -} - -// ==================== New Tests ==================== - -func Test_analyzeBatchResults(t *testing.T) { - ctx := context.Background() - - testCases := []struct { - name string - results []BackfillResult - wantFailures int - }{ - { - name: "empty_results", - results: []BackfillResult{}, - wantFailures: 0, - }, - { - name: "all_success", - results: []BackfillResult{ - {Batch: BackfillBatch{StartLedger: 100, EndLedger: 199}, LedgersCount: 100}, - {Batch: BackfillBatch{StartLedger: 200, EndLedger: 299}, LedgersCount: 100}, - }, - wantFailures: 0, - }, - { - name: "some_failures", - results: []BackfillResult{ - {Batch: BackfillBatch{StartLedger: 100, EndLedger: 199}, LedgersCount: 100}, - {Batch: BackfillBatch{StartLedger: 200, EndLedger: 299}, Error: fmt.Errorf("failed")}, - {Batch: BackfillBatch{StartLedger: 300, EndLedger: 399}, LedgersCount: 100}, - }, - wantFailures: 1, - }, - { - name: "all_failures", - results: []BackfillResult{ - {Batch: BackfillBatch{StartLedger: 100, EndLedger: 199}, Error: fmt.Errorf("failed1")}, - {Batch: BackfillBatch{StartLedger: 200, EndLedger: 299}, Error: fmt.Errorf("failed2")}, - }, - wantFailures: 2, - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - numFailed := analyzeBatchResults(ctx, tc.results) - assert.Equal(t, tc.wantFailures, numFailed) - }) - } -} - -func Test_ingestService_setupBatchBackend(t *testing.T) { - dbt := dbtest.Open(t) - defer dbt.Close() - ctx := context.Background() - dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) - require.NoError(t, err) - defer dbConnectionPool.Close() - - testCases := []struct { - name string - batch BackfillBatch - setupFactory func() LedgerBackendFactory - wantErr bool - wantErrContains string - }{ - { - name: "success", - batch: BackfillBatch{StartLedger: 100, EndLedger: 199}, - setupFactory: func() LedgerBackendFactory { - return func(ctx context.Context) (ledgerbackend.LedgerBackend, error) { - mockBackend := &LedgerBackendMock{} - mockBackend.On("PrepareRange", mock.Anything, ledgerbackend.BoundedRange(100, 199)).Return(nil) - return mockBackend, nil - } - }, - wantErr: false, - }, - { - name: "factory_error", - batch: BackfillBatch{StartLedger: 100, EndLedger: 199}, - setupFactory: func() LedgerBackendFactory { - return func(ctx context.Context) (ledgerbackend.LedgerBackend, error) { - return nil, fmt.Errorf("factory failed") - } - }, - wantErr: true, - wantErrContains: "creating ledger backend", - }, - { - name: "prepare_range_error", - batch: BackfillBatch{StartLedger: 100, EndLedger: 199}, - setupFactory: func() LedgerBackendFactory { - return func(ctx context.Context) (ledgerbackend.LedgerBackend, error) { - mockBackend := &LedgerBackendMock{} - mockBackend.On("PrepareRange", mock.Anything, mock.Anything).Return(fmt.Errorf("prepare failed")) - return mockBackend, nil - } - }, - wantErr: true, - wantErrContains: "preparing backend range", - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - m := metrics.NewMetrics(prometheus.NewRegistry()) - - models, err := data.NewModels(dbConnectionPool, m.DB) - require.NoError(t, err) - - mockRPCService := &RPCServiceMock{} - mockRPCService.On("NetworkPassphrase").Return(network.TestNetworkPassphrase).Maybe() - - svc, err := NewIngestService(IngestServiceConfig{ - IngestionMode: IngestionModeBackfill, - Models: models, - OldestLedgerCursorName: "oldest_ledger_cursor", - AppTracker: &apptracker.MockAppTracker{}, - RPCService: mockRPCService, - LedgerBackend: &LedgerBackendMock{}, - LedgerBackendFactory: tc.setupFactory(), - Metrics: m, - GetLedgersLimit: defaultGetLedgersLimit, - Network: network.TestNetworkPassphrase, - NetworkPassphrase: network.TestNetworkPassphrase, - Archive: &HistoryArchiveMock{}, - }) - require.NoError(t, err) - - backend, err := svc.setupBatchBackend(ctx, tc.batch) - if tc.wantErr { - require.Error(t, err) - if tc.wantErrContains != "" { - assert.Contains(t, err.Error(), tc.wantErrContains) - } - } else { - require.NoError(t, err) - require.NotNil(t, backend) - } - }) - } -} - -func Test_ingestService_updateOldestCursor(t *testing.T) { - dbt := dbtest.Open(t) - defer dbt.Close() - ctx := context.Background() - dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) - require.NoError(t, err) - defer dbConnectionPool.Close() - - testCases := []struct { - name string - initialCursor uint32 - updateTo uint32 - wantCursor uint32 // Expected cursor after update (UpdateMin keeps lower) - }{ - { - name: "updates_to_lower_value", - initialCursor: 100, - updateTo: 50, - wantCursor: 50, - }, - { - name: "keeps_existing_lower_value", - initialCursor: 50, - updateTo: 100, - wantCursor: 50, // UpdateMin keeps the lower value - }, - { - name: "same_value_no_change", - initialCursor: 100, - updateTo: 100, - wantCursor: 100, - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - // Clean up and set initial cursor - setupDBCursors(t, ctx, dbConnectionPool, 200, tc.initialCursor) - - m := metrics.NewMetrics(prometheus.NewRegistry()) - - models, err := data.NewModels(dbConnectionPool, m.DB) - require.NoError(t, err) - - mockRPCService := &RPCServiceMock{} - mockRPCService.On("NetworkPassphrase").Return(network.TestNetworkPassphrase).Maybe() - - svc, err := NewIngestService(IngestServiceConfig{ - IngestionMode: IngestionModeBackfill, - Models: models, - OldestLedgerCursorName: "oldest_ledger_cursor", - AppTracker: &apptracker.MockAppTracker{}, - RPCService: mockRPCService, - LedgerBackend: &LedgerBackendMock{}, - Metrics: m, - GetLedgersLimit: defaultGetLedgersLimit, - Network: network.TestNetworkPassphrase, - NetworkPassphrase: network.TestNetworkPassphrase, - Archive: &HistoryArchiveMock{}, - }) - require.NoError(t, err) - - err = svc.updateOldestCursor(ctx, tc.updateTo) - require.NoError(t, err) - - // Verify the cursor value - cursor, err := models.IngestStore.Get(ctx, "oldest_ledger_cursor") - require.NoError(t, err) - assert.Equal(t, tc.wantCursor, cursor) - }) - } -} - func Test_ingestService_initializeCursors(t *testing.T) { dbt := dbtest.Open(t) defer dbt.Close() @@ -912,632 +545,6 @@ func Test_ingestService_Run(t *testing.T) { } } -func Test_ingestService_flushBatchBufferWithRetry(t *testing.T) { - dbt := dbtest.Open(t) - defer dbt.Close() - ctx := context.Background() - dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) - require.NoError(t, err) - defer dbConnectionPool.Close() - - testCases := []struct { - name string - setupBuffer func() *indexer.IndexerBuffer - updateCursorTo *uint32 - initialCursor uint32 - wantCursor uint32 - wantTxCount int - wantOpCount int - wantStateChangeCount int - txHashes []string // For verification queries - }{ - { - name: "flush_empty_buffer_no_cursor_update", - setupBuffer: func() *indexer.IndexerBuffer { return indexer.NewIndexerBuffer() }, - updateCursorTo: nil, - initialCursor: 100, - wantCursor: 100, - wantTxCount: 0, - wantOpCount: 0, - wantStateChangeCount: 0, - txHashes: []string{}, - }, - { - name: "flush_with_data_inserts_to_database", - setupBuffer: func() *indexer.IndexerBuffer { - buf := indexer.NewIndexerBuffer() - tx1 := createTestTransaction(flushTxHash1, 1) - tx2 := createTestTransaction(flushTxHash2, 2) - op1 := createTestOperation(200) - op2 := createTestOperation(201) - sc1 := createTestStateChange(1, testAddr1, 200) - sc2 := createTestStateChange(2, testAddr2, 201) - - buf.PushTransaction(testAddr1, tx1) - buf.PushTransaction(testAddr2, tx2) - buf.PushOperation(testAddr1, op1, tx1) - buf.PushOperation(testAddr2, op2, tx2) - buf.PushStateChange(tx1, op1, sc1) - buf.PushStateChange(tx2, op2, sc2) - return buf - }, - updateCursorTo: nil, - initialCursor: 100, - wantCursor: 100, - wantTxCount: 2, - wantOpCount: 2, - wantStateChangeCount: 2, - txHashes: []string{flushTxHash1, flushTxHash2}, - }, - { - name: "flush_with_cursor_update_to_lower_value", - setupBuffer: func() *indexer.IndexerBuffer { - buf := indexer.NewIndexerBuffer() - tx1 := createTestTransaction(flushTxHash3, 3) - buf.PushTransaction(testAddr1, tx1) - return buf - }, - updateCursorTo: ptrUint32(50), - initialCursor: 100, - wantCursor: 50, - wantTxCount: 1, - wantOpCount: 0, - wantStateChangeCount: 0, - txHashes: []string{flushTxHash3}, - }, - { - name: "flush_with_cursor_update_to_higher_value_keeps_existing", - setupBuffer: func() *indexer.IndexerBuffer { - buf := indexer.NewIndexerBuffer() - tx1 := createTestTransaction(flushTxHash4, 4) - buf.PushTransaction(testAddr1, tx1) - return buf - }, - updateCursorTo: ptrUint32(150), - initialCursor: 100, - wantCursor: 100, // UpdateMin keeps lower - wantTxCount: 1, - wantOpCount: 0, - wantStateChangeCount: 0, - txHashes: []string{flushTxHash4}, - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - // Clean up test data from previous runs (using HashBytea for BYTEA column) - for _, hash := range []string{flushTxHash1, flushTxHash2, flushTxHash3, flushTxHash4} { - _, err = dbConnectionPool.Exec(ctx, `DELETE FROM state_changes WHERE to_id IN (SELECT to_id FROM transactions WHERE hash = $1)`, types.HashBytea(hash)) - require.NoError(t, err) - _, err = dbConnectionPool.Exec(ctx, `DELETE FROM transactions WHERE hash = $1`, types.HashBytea(hash)) - require.NoError(t, err) - } - // Also clean up any orphan operations - _, err = dbConnectionPool.Exec(ctx, `TRUNCATE operations, operations_accounts CASCADE`) - require.NoError(t, err) - - // Set up initial cursor - setupDBCursors(t, ctx, dbConnectionPool, 200, tc.initialCursor) - - m := metrics.NewMetrics(prometheus.NewRegistry()) - - models, err := data.NewModels(dbConnectionPool, m.DB) - require.NoError(t, err) - - mockRPCService := &RPCServiceMock{} - mockRPCService.On("NetworkPassphrase").Return(network.TestNetworkPassphrase).Maybe() - - // Use variadic mock.Anything for any number of tx hashes - - svc, err := NewIngestService(IngestServiceConfig{ - IngestionMode: IngestionModeBackfill, - Models: models, - OldestLedgerCursorName: "oldest_ledger_cursor", - AppTracker: &apptracker.MockAppTracker{}, - RPCService: mockRPCService, - LedgerBackend: &LedgerBackendMock{}, - Metrics: m, - GetLedgersLimit: defaultGetLedgersLimit, - Network: network.TestNetworkPassphrase, - NetworkPassphrase: network.TestNetworkPassphrase, - Archive: &HistoryArchiveMock{}, - }) - require.NoError(t, err) - - buffer := tc.setupBuffer() - - // Call flushBatchBuffer - err = svc.flushBatchBufferWithRetry(ctx, buffer, tc.updateCursorTo) - require.NoError(t, err) - - // Verify the cursor value - cursor, err := models.IngestStore.Get(ctx, "oldest_ledger_cursor") - require.NoError(t, err) - assert.Equal(t, tc.wantCursor, cursor) - - // Verify transaction count in database - if len(tc.txHashes) > 0 { - hashBytes := make([][]byte, len(tc.txHashes)) - for i, h := range tc.txHashes { - val, err := types.HashBytea(h).Value() - require.NoError(t, err) - hashBytes[i] = val.([]byte) - } - var txCount int - err = dbConnectionPool.QueryRow(ctx, - `SELECT COUNT(*) FROM transactions WHERE hash = ANY($1)`, - hashBytes).Scan(&txCount) - require.NoError(t, err) - assert.Equal(t, tc.wantTxCount, txCount, "transaction count mismatch") - } - - // Verify operation count in database - if tc.wantOpCount > 0 { - var opCount int - err = dbConnectionPool.QueryRow(ctx, - `SELECT COUNT(*) FROM operations`).Scan(&opCount) - require.NoError(t, err) - assert.Equal(t, tc.wantOpCount, opCount, "operation count mismatch") - } - - // Verify state change count in database - if tc.wantStateChangeCount > 0 { - scHashBytes := make([][]byte, len(tc.txHashes)) - for i, h := range tc.txHashes { - val, err := types.HashBytea(h).Value() - require.NoError(t, err) - scHashBytes[i] = val.([]byte) - } - var scCount int - err = dbConnectionPool.QueryRow(ctx, - `SELECT COUNT(*) FROM state_changes WHERE to_id IN (SELECT to_id FROM transactions WHERE hash = ANY($1))`, - scHashBytes).Scan(&scCount) - require.NoError(t, err) - assert.Equal(t, tc.wantStateChangeCount, scCount, "state change count mismatch") - } - }) - } -} - -// ==================== Backfill Failure Scenario Tests ==================== - -// Test_ingestService_processBackfillBatchesParallel_PartialFailure verifies that when one -// batch fails during parallel processing, other batches still complete successfully. -func Test_ingestService_processBackfillBatchesParallel_PartialFailure(t *testing.T) { - dbt := dbtest.Open(t) - defer dbt.Close() - ctx := context.Background() - dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) - require.NoError(t, err) - defer dbConnectionPool.Close() - - testCases := []struct { - name string - batches []BackfillBatch - failBatchIndex int // which batch index should fail (0-based) - wantFailedCount int - }{ - { - name: "middle_batch_fails", - batches: []BackfillBatch{ - {StartLedger: 100, EndLedger: 109}, - {StartLedger: 110, EndLedger: 119}, // This one fails - {StartLedger: 120, EndLedger: 129}, - }, - failBatchIndex: 1, - wantFailedCount: 1, - }, - { - name: "first_batch_fails", - batches: []BackfillBatch{ - {StartLedger: 100, EndLedger: 109}, // This one fails - {StartLedger: 110, EndLedger: 119}, - {StartLedger: 120, EndLedger: 129}, - }, - failBatchIndex: 0, - wantFailedCount: 1, - }, - { - name: "last_batch_fails", - batches: []BackfillBatch{ - {StartLedger: 100, EndLedger: 109}, - {StartLedger: 110, EndLedger: 119}, - {StartLedger: 120, EndLedger: 129}, // This one fails - }, - failBatchIndex: 2, - wantFailedCount: 1, - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - m := metrics.NewMetrics(prometheus.NewRegistry()) - - models, modelsErr := data.NewModels(dbConnectionPool, m.DB) - require.NoError(t, modelsErr) - - mockRPCService := &RPCServiceMock{} - mockRPCService.On("NetworkPassphrase").Return(network.TestNetworkPassphrase).Maybe() - - failBatch := tc.batches[tc.failBatchIndex] - - // Factory that returns a backend that fails PrepareRange for the specified batch - factory := func(ctx context.Context) (ledgerbackend.LedgerBackend, error) { - mockBackend := &LedgerBackendMock{} - // Use MatchedBy to check if this is the failing batch - mockBackend.On("PrepareRange", mock.Anything, mock.MatchedBy(func(r ledgerbackend.Range) bool { - return r.From() == failBatch.StartLedger - })).Return(fmt.Errorf("simulated failure for batch starting at %d", failBatch.StartLedger)) - // All other batches succeed - return error to prevent processing - // This avoids nil pointer issues with empty LedgerCloseMeta - mockBackend.On("PrepareRange", mock.Anything, mock.Anything).Return(nil).Maybe() - // Return proper empty ledger meta to avoid nil pointer issues - mockBackend.On("GetLedger", mock.Anything, mock.Anything).Return(xdr.LedgerCloseMeta{ - V: 0, - V0: &xdr.LedgerCloseMetaV0{ - LedgerHeader: xdr.LedgerHeaderHistoryEntry{ - Header: xdr.LedgerHeader{ - LedgerSeq: xdr.Uint32(100), - }, - }, - }, - }, nil).Maybe() - mockBackend.On("Close").Return(nil).Maybe() - return mockBackend, nil - } - - svc, svcErr := NewIngestService(IngestServiceConfig{ - IngestionMode: IngestionModeBackfill, - Models: models, - OldestLedgerCursorName: "oldest_ledger_cursor", - AppTracker: &apptracker.MockAppTracker{}, - RPCService: mockRPCService, - LedgerBackend: &LedgerBackendMock{}, - LedgerBackendFactory: factory, - Metrics: m, - GetLedgersLimit: defaultGetLedgersLimit, - Network: network.TestNetworkPassphrase, - NetworkPassphrase: network.TestNetworkPassphrase, - Archive: &HistoryArchiveMock{}, - BackfillBatchSize: 10, - }) - require.NoError(t, svcErr) - - results := svc.processBackfillBatchesParallel(ctx, tc.batches, nil) - - // Verify results - require.Len(t, results, len(tc.batches)) - - // Count actual failures - actualFailed := 0 - for i, result := range results { - if result.Error != nil { - actualFailed++ - assert.Equal(t, tc.failBatchIndex, i, "expected batch at index %d to fail", tc.failBatchIndex) - assert.Contains(t, result.Error.Error(), "preparing backend range") - } - } - assert.Equal(t, tc.wantFailedCount, actualFailed) - - // Verify analyzeBatchResults returns correct count - numFailed := analyzeBatchResults(ctx, results) - assert.Equal(t, tc.wantFailedCount, numFailed) - }) - } -} - -// Test_ingestService_startBackfilling_HistoricalMode_PartialFailure_CursorUpdate verifies -// that the oldest_ingest_ledger cursor is updated to the minimum of successful batches -// when some batches fail during historical backfill. -func Test_ingestService_startBackfilling_HistoricalMode_PartialFailure_CursorUpdate(t *testing.T) { - dbt := dbtest.Open(t) - defer dbt.Close() - ctx := context.Background() - dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) - require.NoError(t, err) - defer dbConnectionPool.Close() - - testCases := []struct { - name string - initialOldest uint32 - initialLatest uint32 - startLedger uint32 - endLedger uint32 - batchSize uint32 - failBatchStart uint32 // start ledger of batch to fail - wantFinalOldest uint32 - wantError bool - wantErrContains string - }{ - { - name: "middle_batch_fails_cursor_updates_to_minimum", - initialOldest: 100, - initialLatest: 100, - startLedger: 50, - endLedger: 99, - batchSize: 10, - failBatchStart: 70, // Batch [70-79] fails - wantFinalOldest: 50, // Minimum of successful batches (50, 60, 80, 90) - wantError: false, - }, - { - name: "first_batch_fails_cursor_updates_to_second_batch", - initialOldest: 100, - initialLatest: 100, - startLedger: 50, - endLedger: 79, - batchSize: 10, - failBatchStart: 50, // Batch [50-59] fails - wantFinalOldest: 60, // Next successful batch - wantError: false, - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - // Clean up database - _, err = dbConnectionPool.Exec(ctx, `DELETE FROM transactions`) - require.NoError(t, err) - _, err = dbConnectionPool.Exec(ctx, `DELETE FROM operations`) - require.NoError(t, err) - _, err = dbConnectionPool.Exec(ctx, `DELETE FROM state_changes`) - require.NoError(t, err) - - // Set up initial cursors - setupDBCursors(t, ctx, dbConnectionPool, tc.initialLatest, tc.initialOldest) - - // Insert anchor transaction so GetOldestLedger() returns initialOldest - _, err = dbConnectionPool.Exec(ctx, - `INSERT INTO transactions (hash, to_id, fee_charged, result_code, ledger_number, ledger_created_at) - VALUES ('anchor_hash', 1, 100, 'TransactionResultCodeTxSuccess', $1, NOW())`, - tc.initialOldest) - require.NoError(t, err) - - m := metrics.NewMetrics(prometheus.NewRegistry()) - - models, modelsErr := data.NewModels(dbConnectionPool, m.DB) - require.NoError(t, modelsErr) - - mockRPCService := &RPCServiceMock{} - mockRPCService.On("NetworkPassphrase").Return(network.TestNetworkPassphrase).Maybe() - - // Factory that fails for the specified batch - factory := func(ctx context.Context) (ledgerbackend.LedgerBackend, error) { - mockBackend := &LedgerBackendMock{} - mockBackend.On("PrepareRange", mock.Anything, mock.MatchedBy(func(r ledgerbackend.Range) bool { - return r.From() == tc.failBatchStart - })).Return(fmt.Errorf("simulated network failure")) - mockBackend.On("PrepareRange", mock.Anything, mock.Anything).Return(nil).Maybe() - // For successful batches, return empty ledgers (no transactions) - mockBackend.On("GetLedger", mock.Anything, mock.Anything).Return(xdr.LedgerCloseMeta{ - V: 0, - V0: &xdr.LedgerCloseMetaV0{ - LedgerHeader: xdr.LedgerHeaderHistoryEntry{ - Header: xdr.LedgerHeader{ - LedgerSeq: xdr.Uint32(50), - }, - }, - }, - }, nil).Maybe() - mockBackend.On("Close").Return(nil).Maybe() - return mockBackend, nil - } - - svc, svcErr := NewIngestService(IngestServiceConfig{ - IngestionMode: IngestionModeBackfill, - Models: models, - OldestLedgerCursorName: "oldest_ledger_cursor", - AppTracker: &apptracker.MockAppTracker{}, - RPCService: mockRPCService, - LedgerBackend: &LedgerBackendMock{}, - LedgerBackendFactory: factory, - Metrics: m, - GetLedgersLimit: defaultGetLedgersLimit, - Network: network.TestNetworkPassphrase, - NetworkPassphrase: network.TestNetworkPassphrase, - Archive: &HistoryArchiveMock{}, - BackfillBatchSize: int(tc.batchSize), - }) - require.NoError(t, svcErr) - - // Run backfilling - backfillErr := svc.startBackfilling(ctx, tc.startLedger, tc.endLedger) - - if tc.wantError { - require.Error(t, backfillErr) - if tc.wantErrContains != "" { - assert.Contains(t, backfillErr.Error(), tc.wantErrContains) - } - } else { - // Historical mode should not return error even with partial failures - require.NoError(t, backfillErr) - } - - // Verify cursor was updated correctly using IngestStore.Get - finalOldest, getErr := models.IngestStore.Get(ctx, "oldest_ledger_cursor") - require.NoError(t, getErr) - assert.Equal(t, tc.wantFinalOldest, finalOldest, - "oldest cursor should be updated to minimum of successful batches") - }) - } -} - -// Test_ingestService_processBackfillBatches_PartialFailure_OnlySuccessfulBatchPersisted verifies -// that when one batch fails and another succeeds: -// - The failed batch does not persist any data -// - The successful batch persists its transactions -// - Proper error handling for both cases -func Test_ingestService_processBackfillBatches_PartialFailure_OnlySuccessfulBatchPersisted(t *testing.T) { - dbt := dbtest.Open(t) - defer dbt.Close() - ctx := context.Background() - dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) - require.NoError(t, err) - defer dbConnectionPool.Close() - - // Clean up database - _, err = dbConnectionPool.Exec(ctx, `DELETE FROM transactions`) - require.NoError(t, err) - _, err = dbConnectionPool.Exec(ctx, `DELETE FROM operations`) - require.NoError(t, err) - _, err = dbConnectionPool.Exec(ctx, `DELETE FROM state_changes`) - require.NoError(t, err) - - // Set up initial cursors - setupDBCursors(t, ctx, dbConnectionPool, 200, 200) - - m := metrics.NewMetrics(prometheus.NewRegistry()) - - models, modelsErr := data.NewModels(dbConnectionPool, m.DB) - require.NoError(t, modelsErr) - - mockRPCService := &RPCServiceMock{} - mockRPCService.On("NetworkPassphrase").Return(network.TestNetworkPassphrase).Maybe() - - // Parse ledger metadata with a transaction for the successful batch - var metaWithTx xdr.LedgerCloseMeta - err = xdr.SafeUnmarshalBase64(ledgerMetadataWith1Tx, &metaWithTx) - require.NoError(t, err) - - // Two batches: first fails, second succeeds - // Use single-ledger batches to avoid duplicate key issues with test fixtures - batches := []BackfillBatch{ - {StartLedger: 100, EndLedger: 100}, // Will fail - {StartLedger: 110, EndLedger: 110}, // Will succeed - single ledger to avoid duplicate tx hash - } - - // Factory that returns a mock backend configured based on the batch range - factory := func(ctx context.Context) (ledgerbackend.LedgerBackend, error) { - mockBackend := &LedgerBackendMock{} - // Fail batch 1 (100-109) at PrepareRange - mockBackend.On("PrepareRange", mock.Anything, mock.MatchedBy(func(r ledgerbackend.Range) bool { - return r.From() == 100 - })).Return(fmt.Errorf("ledger range unavailable")) - // Succeed batch 2 (110-119) - mockBackend.On("PrepareRange", mock.Anything, mock.Anything).Return(nil).Maybe() - mockBackend.On("GetLedger", mock.Anything, mock.Anything).Return(metaWithTx, nil).Maybe() - mockBackend.On("Close").Return(nil).Maybe() - return mockBackend, nil - } - - svc, svcErr := NewIngestService(IngestServiceConfig{ - IngestionMode: IngestionModeBackfill, - Models: models, - OldestLedgerCursorName: "oldest_ledger_cursor", - AppTracker: &apptracker.MockAppTracker{}, - RPCService: mockRPCService, - LedgerBackend: &LedgerBackendMock{}, - LedgerBackendFactory: factory, - Metrics: m, - GetLedgersLimit: defaultGetLedgersLimit, - Network: network.TestNetworkPassphrase, - NetworkPassphrase: network.TestNetworkPassphrase, - Archive: &HistoryArchiveMock{}, - BackfillBatchSize: 10, - BackfillDBInsertBatchSize: 50, - }) - require.NoError(t, svcErr) - - // Process both batches in parallel - results := svc.processBackfillBatchesParallel(ctx, batches, nil) - - // Verify we got results for both batches - require.Len(t, results, 2) - - // Verify batch 1 (100-109) failed - require.Error(t, results[0].Error, "batch 1 should have failed") - assert.Contains(t, results[0].Error.Error(), "preparing backend range") - assert.Contains(t, results[0].Error.Error(), "ledger range unavailable") - - // Verify batch 2 (110-119) succeeded - require.NoError(t, results[1].Error, "batch 2 should have succeeded") - - // Verify no transactions were persisted for failed batch (ledger 100) - var failedBatchTxCount int - err = dbConnectionPool.QueryRow(ctx, - `SELECT COUNT(*) FROM transactions WHERE ledger_number BETWEEN $1 AND $2`, - 100, 109).Scan(&failedBatchTxCount) - require.NoError(t, err) - assert.Equal(t, 0, failedBatchTxCount, "no transactions should be persisted for failed batch") - - // Verify transactions were persisted for successful batch - // The ledgerMetadataWith1Tx fixture has a fixed ledger sequence of 4478 - // so we query for that ledger number (the XDR metadata determines the stored ledger) - var successBatchTxCount int - err = dbConnectionPool.QueryRow(ctx, - `SELECT COUNT(*) FROM transactions WHERE ledger_number = $1`, - 4478).Scan(&successBatchTxCount) - require.NoError(t, err) - assert.Equal(t, 1, successBatchTxCount, "1 transaction should be persisted for successful batch") -} - -// Test_ingestService_startBackfilling_HistoricalMode_AllBatchesFail_CursorUnchanged -// verifies that the cursor remains unchanged when all batches fail during historical backfill. -func Test_ingestService_startBackfilling_HistoricalMode_AllBatchesFail_CursorUnchanged(t *testing.T) { - dbt := dbtest.Open(t) - defer dbt.Close() - ctx := context.Background() - dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) - require.NoError(t, err) - defer dbConnectionPool.Close() - - // Clean up database - _, err = dbConnectionPool.Exec(ctx, `DELETE FROM transactions`) - require.NoError(t, err) - - // Set up initial cursors - initialOldest := uint32(100) - initialLatest := uint32(100) - setupDBCursors(t, ctx, dbConnectionPool, initialLatest, initialOldest) - - m := metrics.NewMetrics(prometheus.NewRegistry()) - - models, modelsErr := data.NewModels(dbConnectionPool, m.DB) - require.NoError(t, modelsErr) - - mockRPCService := &RPCServiceMock{} - mockRPCService.On("NetworkPassphrase").Return(network.TestNetworkPassphrase).Maybe() - - // Factory that always fails - factory := func(ctx context.Context) (ledgerbackend.LedgerBackend, error) { - return nil, fmt.Errorf("all backends unavailable") - } - - svc, svcErr := NewIngestService(IngestServiceConfig{ - IngestionMode: IngestionModeBackfill, - Models: models, - OldestLedgerCursorName: "oldest_ledger_cursor", - AppTracker: &apptracker.MockAppTracker{}, - RPCService: mockRPCService, - LedgerBackend: &LedgerBackendMock{}, - LedgerBackendFactory: factory, - Metrics: m, - GetLedgersLimit: defaultGetLedgersLimit, - Network: network.TestNetworkPassphrase, - NetworkPassphrase: network.TestNetworkPassphrase, - Archive: &HistoryArchiveMock{}, - BackfillBatchSize: 10, - }) - require.NoError(t, svcErr) - - // Run backfilling with all batches failing - backfillErr := svc.startBackfilling(ctx, 50, 99) - - // Historical mode should NOT return error even when all batches fail - require.NoError(t, backfillErr, "historical mode should not return error even when all batches fail") - - // Verify cursor remains unchanged - finalOldest, getErr := models.IngestStore.Get(ctx, "oldest_ledger_cursor") - require.NoError(t, getErr) - assert.Equal(t, initialOldest, finalOldest, - "oldest cursor should remain unchanged when all batches fail") - - finalLatest, getErr := models.IngestStore.Get(ctx, data.LatestLedgerCursorName) - require.NoError(t, getErr) - assert.Equal(t, initialLatest, finalLatest, - "latest cursor should remain unchanged when all batches fail") -} - // Test_ingestProcessedDataWithRetry tests the ingestProcessedDataWithRetry function covering success, failure, and retry scenarios. func Test_ingestProcessedDataWithRetry(t *testing.T) { t.Run("success - processes data and updates cursor", func(t *testing.T) { @@ -1789,73 +796,6 @@ func Test_ingestProcessedDataWithRetry(t *testing.T) { }) } -// Test_ingestService_processBackfillBatchesParallel_Success verifies that -// backfill batches process successfully. -func Test_ingestService_processBackfillBatchesParallel_Success(t *testing.T) { - dbt := dbtest.Open(t) - defer dbt.Close() - ctx := context.Background() - dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) - require.NoError(t, err) - defer dbConnectionPool.Close() - - m := metrics.NewMetrics(prometheus.NewRegistry()) - - models, modelsErr := data.NewModels(dbConnectionPool, m.DB) - require.NoError(t, modelsErr) - - mockRPCService := &RPCServiceMock{} - mockRPCService.On("NetworkPassphrase").Return(network.TestNetworkPassphrase).Maybe() - - // Factory that returns a backend with minimal valid ledger data - factory := func(ctx context.Context) (ledgerbackend.LedgerBackend, error) { - mockBackend := &LedgerBackendMock{} - mockBackend.On("PrepareRange", mock.Anything, mock.Anything).Return(nil) - mockBackend.On("GetLedger", mock.Anything, mock.Anything).Return(xdr.LedgerCloseMeta{ - V: 0, - V0: &xdr.LedgerCloseMetaV0{ - LedgerHeader: xdr.LedgerHeaderHistoryEntry{ - Header: xdr.LedgerHeader{ - LedgerSeq: xdr.Uint32(100), - }, - }, - }, - }, nil) - mockBackend.On("Close").Return(nil) - return mockBackend, nil - } - - svc, svcErr := NewIngestService(IngestServiceConfig{ - IngestionMode: IngestionModeBackfill, - Models: models, - OldestLedgerCursorName: "oldest_ledger_cursor", - AppTracker: &apptracker.MockAppTracker{}, - RPCService: mockRPCService, - LedgerBackend: &LedgerBackendMock{}, - LedgerBackendFactory: factory, - Metrics: m, - GetLedgersLimit: defaultGetLedgersLimit, - Network: network.TestNetworkPassphrase, - NetworkPassphrase: network.TestNetworkPassphrase, - Archive: &HistoryArchiveMock{}, - BackfillBatchSize: 10, - }) - require.NoError(t, svcErr) - - batches := []BackfillBatch{ - {StartLedger: 100, EndLedger: 100}, - {StartLedger: 101, EndLedger: 101}, - } - - results := svc.processBackfillBatchesParallel(ctx, batches, nil) - - // All batches should succeed - require.Len(t, results, 2) - for i, result := range results { - assert.NoError(t, result.Error, "batch %d should succeed", i) - } -} - // testProtocolProcessor is a test-only ProtocolProcessor that writes sentinel // values into ingest_store within the DB transaction, proving PersistHistory // and PersistCurrentState were called and committed atomically.