Add backfilling prometheus metrics

Dockerfile

@@ -30,6 +30,7 @@ RUN apt-get update && \
     echo "deb https://apt.stellar.org jammy unstable" >/etc/apt/sources.list.d/SDF-unstable.list
 
 COPY --from=api-build /bin/wallet-backend /app/
+COPY --from=api-build /src/wallet-backend/config /app/config
 
 EXPOSE 8001
 WORKDIR /app

cmd/ingest.go

@@ -37,21 +37,61 @@ func (c *ingestCmd) Command() *cobra.Command {
 		utils.RedisHostOption(&cfg.RedisHost),
 		utils.RedisPortOption(&cfg.RedisPort),
 		{
-			Name:        "ledger-cursor-name",
+			Name:        "ingestion-mode",
+			Usage:       "What mode to run ingestion in - live or backfill",
+			OptType:     types.String,
+			ConfigKey:   &cfg.IngestionMode,
+			FlagDefault: string(ingest.IngestionModeLive),
+			Required:    true,
+		},
+		{
+			Name:        "latest-ledger-cursor-name",
 			Usage:       "Name of last synced ledger cursor, used to keep track of the last ledger ingested by the service. When starting up, ingestion will resume from the ledger number stored in this record. It should be an unique name per container as different containers would overwrite the cursor value of its peers when using the same cursor name.",
 			OptType:     types.String,
-			ConfigKey:   &cfg.LedgerCursorName,
-			FlagDefault: "live_ingest_cursor",
+			ConfigKey:   &cfg.LatestLedgerCursorName,
+			FlagDefault: "latest_ingest_ledger",
 			Required:    true,
 		},
 		{
-			Name:        "account-tokens-cursor-name",
-			Usage:       "Name of last synced account tokens ledger cursor, used to keep track of the last ledger ingested by the service.",
+			Name:        "oldest-ledger-cursor-name",
+			Usage:       "Name of the oldest ledger cursor, used to track the earliest ledger ingested by the service. Used for backfill operations to know where historical data begins.",
 			OptType:     types.String,
-			ConfigKey:   &cfg.AccountTokensCursorName,
-			FlagDefault: "live_account_tokens_ingest_cursor",
+			ConfigKey:   &cfg.OldestLedgerCursorName,
+			FlagDefault: "oldest_ingest_ledger",
 			Required:    true,
 		},
+		{
+			Name:        "backfill-workers",
+			Usage:       "Maximum concurrent workers for backfill processing. Defaults to number of CPUs. Lower values reduce RAM usage at cost of throughput.",
+			OptType:     types.Int,
+			ConfigKey:   &cfg.BackfillWorkers,
+			FlagDefault: 0,
+			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.",
+			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 50. Lower values reduce RAM usage at cost of more DB transactions.",
+			OptType:     types.Int,
+			ConfigKey:   &cfg.BackfillDBInsertBatchSize,
+			FlagDefault: 100,
+			Required:    false,
+		},
+		{
+			Name:        "catchup-threshold",
+			Usage:       "Number of ledgers behind network tip that triggers fast catchup via backfilling. Defaults to 100.",
+			OptType:     types.Int,
+			ConfigKey:   &cfg.CatchupThreshold,
+			FlagDefault: 100,
+			Required:    false,
+		},
 		{
 			Name:        "archive-url",
 			Usage:       "Archive URL for history archives",
@@ -73,15 +113,39 @@ func (c *ingestCmd) Command() *cobra.Command {
 			Usage:       "Type of ledger backend to use for fetching ledgers. Options: 'rpc' (default) or 'datastore'",
 			OptType:     types.String,
 			ConfigKey:   &ledgerBackendType,
-			FlagDefault: string(ingest.LedgerBackendTypeRPC),
+			FlagDefault: string(ingest.LedgerBackendTypeDatastore),
 			Required:    false,
 		},
 		{
 			Name:        "datastore-config-path",
 			Usage:       "Path to TOML config file for datastore backend. Required when ledger-backend-type is 'datastore'",
 			OptType:     types.String,
 			ConfigKey:   &cfg.DatastoreConfigPath,
-			FlagDefault: "",
+			FlagDefault: "config/datastore-pubnet.toml",
+			Required:    false,
+		},
+		{
+			Name:        "skip-tx-meta",
+			Usage:       "Skip storing transaction metadata (meta_xdr) to reduce storage space and improve insertion performance.",
+			OptType:     types.Bool,
+			ConfigKey:   &cfg.SkipTxMeta,
+			FlagDefault: true,
+			Required:    false,
+		},
+		{
+			Name:        "skip-tx-envelope",
+			Usage:       "Skip storing transaction envelope (envelope_xdr) to reduce storage space and improve insertion performance.",
+			OptType:     types.Bool,
+			ConfigKey:   &cfg.SkipTxEnvelope,
+			FlagDefault: true,
+			Required:    false,
+		},
+		{
+			Name:        "enable-participant-filtering",
+			Usage:       "When enabled, only store transactions, operations, and state changes for pre-registered accounts. When disabled (default), store all data.",
+			OptType:     types.Bool,
+			ConfigKey:   &cfg.EnableParticipantFiltering,
+			FlagDefault: false,
 			Required:    false,
 		},
 	}

docker-compose.yaml

@@ -23,24 +23,9 @@ services:
       - postgres-db:/data/postgres
     ports:
       - 5432:5432
-    command: |
-      bash -c "
-      # Run default postgres initialization
-      /usr/local/bin/docker-entrypoint.sh postgres &
-
-      # Wait for startup
-      sleep 10
-
-      # Create schemas (only on first run)
-      if [ ! -f /data/postgres/schemas_created ]; then
-        psql -U postgres -d wallet-backend -c 'CREATE SCHEMA IF NOT EXISTS wallet_backend_mainnet;'
-        psql -U postgres -d wallet-backend -c 'CREATE SCHEMA IF NOT EXISTS wallet_backend_testnet;'
-        touch /data/postgres/schemas_created
-      fi
-
-      # Keep postgres running
-      wait
-      "
+    configs:
+      - source: postgres_init
+        target: /docker-entrypoint-initdb.d/init-schemas.sql
 
   redis:
     container_name: redis
@@ -305,3 +290,9 @@ volumes:
     driver: local
   redis-data:
     driver: local
+
+configs:
+  postgres_init:
+    content: |
+      CREATE SCHEMA IF NOT EXISTS wallet_backend_mainnet;
+      CREATE SCHEMA IF NOT EXISTS wallet_backend_testnet;

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/go-chi/chi v4.1.2+incompatible
 	github.com/go-playground/validator/v10 v10.27.0
 	github.com/golang-jwt/jwt/v5 v5.2.3
+	github.com/jackc/pgx/v5 v5.7.6
 	github.com/jmoiron/sqlx v1.4.0
 	github.com/lib/pq v1.10.9
 	github.com/mattn/go-sqlite3 v1.14.28
@@ -122,6 +123,9 @@ require (
 	github.com/hashicorp/golang-lru v1.0.2 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect

go.sum

@@ -298,6 +298,14 @@ github.com/imkira/go-interpol v1.1.0 h1:KIiKr0VSG2CUW1hl1jpiyuzuJeKUUpC8iM1AIE7N
 github.com/imkira/go-interpol v1.1.0/go.mod h1:z0h2/2T3XF8kyEPpRgJ3kmNv+C43p+I/CoI+jC3w2iA=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.7.6 h1:rWQc5FwZSPX58r1OQmkuaNicxdmExaEz5A2DO2hUuTk=
+github.com/jackc/pgx/v5 v5.7.6/go.mod h1:aruU7o91Tc2q2cFp5h4uP3f6ztExVpyVv88Xl/8Vl8M=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jarcoal/httpmock v0.0.0-20161210151336-4442edb3db31 h1:Aw95BEvxJ3K6o9GGv5ppCd1P8hkeIeEJ30FO+OhOJpM=
 github.com/jarcoal/httpmock v0.0.0-20161210151336-4442edb3db31/go.mod h1:ks+b9deReOc7jgqp+e7LuFiCBH6Rm5hL32cLcEAArb4=
 github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=

internal/data/accounts.go

@@ -150,13 +150,10 @@ func (m *AccountModel) IsAccountFeeBumpEligible(ctx context.Context, address str
 
 // BatchGetByTxHashes gets the accounts that are associated with the given transaction hashes.
 func (m *AccountModel) BatchGetByTxHashes(ctx context.Context, txHashes []string, columns string) ([]*types.AccountWithTxHash, error) {
-	columns = prepareColumnsWithID(columns, types.Account{}, "accounts", "stellar_address")
-	query := fmt.Sprintf(`
-		SELECT %s, transactions_accounts.tx_hash
+	query := `
+		SELECT account_id AS stellar_address, tx_hash
 		FROM transactions_accounts
-		INNER JOIN accounts
-		ON transactions_accounts.account_id = accounts.stellar_address
-		WHERE transactions_accounts.tx_hash = ANY($1)`, columns)
+		WHERE tx_hash = ANY($1)`
 	var accounts []*types.AccountWithTxHash
 	start := time.Now()
 	err := m.DB.SelectContext(ctx, &accounts, query, pq.Array(txHashes))
@@ -173,13 +170,10 @@ func (m *AccountModel) BatchGetByTxHashes(ctx context.Context, txHashes []string
 
 // BatchGetByOperationIDs gets the accounts that are associated with the given operation IDs.
 func (m *AccountModel) BatchGetByOperationIDs(ctx context.Context, operationIDs []int64, columns string) ([]*types.AccountWithOperationID, error) {
-	columns = prepareColumnsWithID(columns, types.Account{}, "accounts", "stellar_address")
-	query := fmt.Sprintf(`
-		SELECT %s, operations_accounts.operation_id
+	query := `
+		SELECT account_id AS stellar_address, operation_id
 		FROM operations_accounts
-		INNER JOIN accounts
-		ON operations_accounts.account_id = accounts.stellar_address
-		WHERE operations_accounts.operation_id = ANY($1)`, columns)
+		WHERE operation_id = ANY($1)`
 	var accounts []*types.AccountWithOperationID
 	start := time.Now()
 	err := m.DB.SelectContext(ctx, &accounts, query, pq.Array(operationIDs))
@@ -196,8 +190,6 @@ func (m *AccountModel) BatchGetByOperationIDs(ctx context.Context, operationIDs
 
 // BatchGetByStateChangeIDs gets the accounts that are associated with the given state change IDs.
 func (m *AccountModel) BatchGetByStateChangeIDs(ctx context.Context, scToIDs []int64, scOrders []int64, columns string) ([]*types.AccountWithStateChangeID, error) {
-	columns = prepareColumnsWithID(columns, types.Account{}, "accounts", "stellar_address")
-
 	// Build tuples for the IN clause. Since (to_id, state_change_order) is the primary key of state_changes,
 	// it will be faster to search on this tuple.
 	tuples := make([]string, len(scOrders))
@@ -206,12 +198,11 @@ func (m *AccountModel) BatchGetByStateChangeIDs(ctx context.Context, scToIDs []i
 	}
 
 	query := fmt.Sprintf(`
-		SELECT %s, CONCAT(state_changes.to_id, '-', state_changes.state_change_order) AS state_change_id
-		FROM accounts
-		INNER JOIN state_changes ON accounts.stellar_address = state_changes.account_id
-		WHERE (state_changes.to_id, state_changes.state_change_order) IN (%s)
-		ORDER BY accounts.created_at DESC
-	`, columns, strings.Join(tuples, ", "))
+		SELECT account_id AS stellar_address, CONCAT(to_id, '-', state_change_order) AS state_change_id
+		FROM state_changes
+		WHERE (to_id, state_change_order) IN (%s)
+		ORDER BY ledger_created_at DESC
+	`, strings.Join(tuples, ", "))
 
 	var accountsWithStateChanges []*types.AccountWithStateChangeID
 	start := time.Now()

internal/data/ingest_store.go

@@ -12,6 +12,11 @@ import (
 	"github.com/stellar/wallet-backend/internal/utils"
 )
 
+type LedgerRange struct {
+	GapStart uint32 `db:"gap_start"`
+	GapEnd   uint32 `db:"gap_end"`
+}
+
 type IngestStoreModel struct {
 	DB             db.ConnectionPool
 	MetricsService metrics.MetricsService
@@ -22,17 +27,17 @@ func (m *IngestStoreModel) Get(ctx context.Context, cursorName string) (uint32,
 	start := time.Now()
 	err := m.DB.GetContext(ctx, &lastSyncedLedger, `SELECT value FROM ingest_store WHERE key = $1`, cursorName)
 	duration := time.Since(start).Seconds()
-	m.MetricsService.ObserveDBQueryDuration("GetLatestLedgerSynced", "ingest_store", duration)
+	m.MetricsService.ObserveDBQueryDuration("Get", "ingest_store", duration)
 	// First run, key does not exist yet
 	if errors.Is(err, sql.ErrNoRows) {
-		m.MetricsService.IncDBQuery("GetLatestLedgerSynced", "ingest_store")
+		m.MetricsService.IncDBQuery("Get", "ingest_store")
 		return 0, nil
 	}
 	if err != nil {
-		m.MetricsService.IncDBQueryError("GetLatestLedgerSynced", "ingest_store", utils.GetDBErrorType(err))
+		m.MetricsService.IncDBQueryError("Get", "ingest_store", utils.GetDBErrorType(err))
 		return 0, fmt.Errorf("getting latest ledger synced for cursor %s: %w", cursorName, err)
 	}
-	m.MetricsService.IncDBQuery("GetLatestLedgerSynced", "ingest_store")
+	m.MetricsService.IncDBQuery("Get", "ingest_store")
 
 	return lastSyncedLedger, nil
 }
@@ -45,12 +50,48 @@ func (m *IngestStoreModel) Update(ctx context.Context, dbTx db.Transaction, curs
 	start := time.Now()
 	_, err := dbTx.ExecContext(ctx, query, cursorName, ledger)
 	duration := time.Since(start).Seconds()
-	m.MetricsService.ObserveDBQueryDuration("UpdateLatestLedgerSynced", "ingest_store", duration)
+	m.MetricsService.ObserveDBQueryDuration("Update", "ingest_store", duration)
 	if err != nil {
-		m.MetricsService.IncDBQueryError("UpdateLatestLedgerSynced", "ingest_store", utils.GetDBErrorType(err))
+		m.MetricsService.IncDBQueryError("Update", "ingest_store", utils.GetDBErrorType(err))
 		return fmt.Errorf("updating last synced ledger to %d: %w", ledger, err)
 	}
-	m.MetricsService.IncDBQuery("UpdateLatestLedgerSynced", "ingest_store")
+	m.MetricsService.IncDBQuery("Update", "ingest_store")
+	return nil
+}
 
+func (m *IngestStoreModel) UpdateMin(ctx context.Context, dbTx db.Transaction, cursorName string, ledger uint32) error {
+	const query = `
+		UPDATE ingest_store
+		SET value = LEAST(value::integer, $2)::text
+		WHERE key = $1
+	`
+	_, err := dbTx.ExecContext(ctx, query, cursorName, ledger)
+	if err != nil {
+		return fmt.Errorf("updating minimum ledger for cursor %s: %w", cursorName, err)
+	}
 	return nil
 }
+
+func (m *IngestStoreModel) GetLedgerGaps(ctx context.Context) ([]LedgerRange, error) {
+	const query = `
+		SELECT gap_start, gap_end FROM (
+			SELECT
+				ledger_number + 1 AS gap_start,
+				LEAD(ledger_number) OVER (ORDER BY ledger_number) - 1 AS gap_end
+			FROM (SELECT DISTINCT ledger_number FROM transactions) t
+		) gaps
+		WHERE gap_start <= gap_end
+		ORDER BY gap_start
+	`
+	start := time.Now()
+	var ledgerGaps []LedgerRange
+	err := m.DB.SelectContext(ctx, &ledgerGaps, query)
+	duration := time.Since(start).Seconds()
+	m.MetricsService.ObserveDBQueryDuration("GetLedgerGaps", "transactions", duration)
+	if err != nil {
+		m.MetricsService.IncDBQueryError("GetLedgerGaps", "transactions", utils.GetDBErrorType(err))
+		return nil, fmt.Errorf("getting ledger gaps: %w", err)
+	}
+	m.MetricsService.IncDBQuery("GetLedgerGaps", "transactions")
+	return ledgerGaps, nil
+}