Tx index for Parquet receipt store (#3222)

## Describe your changes and provide context

Create a tx index to store (tx hash -> block number) to allow parquet to
quickly serve getReceiptByHash queries.

## Testing performed to validate your change
unit tests

Author: jewei1997

sei-db/config/receipt_config.go

@@ -20,8 +20,21 @@ const (
 	flagRSAsyncWriteBuffer     = "receipt-store.async-write-buffer"
 	flagRSKeepRecent           = "receipt-store.keep-recent"
 	flagRSPruneIntervalSeconds = "receipt-store.prune-interval-seconds"
+	flagRSTxIndexBackend       = "receipt-store.tx-index-backend"
+
+	ReceiptTxIndexBackendNone   = ""
+	ReceiptTxIndexBackendPebble = "pebbledb"
 )
 
+func NormalizeReceiptTxIndexBackend(backend string) string {
+	switch strings.ToLower(strings.TrimSpace(backend)) {
+	case "pebbledb":
+		return ReceiptTxIndexBackendPebble
+	default:
+		return ReceiptTxIndexBackendNone
+	}
+}
+
 // ReceiptStoreConfig defines configuration for the receipt store database.
 type ReceiptStoreConfig struct {
 	// DBDirectory defines the directory to store the receipt store db files
@@ -49,10 +62,13 @@ type ReceiptStoreConfig struct {
 	// default to every 600 seconds
 	PruneIntervalSeconds int `mapstructure:"prune-interval-seconds"`
 
-	// DisableTxIndexLookup must remain true. The tx_hash -> block_number lookup
-	// implementation is intentionally unsupported; setting this to false will
-	// panic during parquet store initialization.
-	DisableTxIndexLookup bool `mapstructure:"disable-tx-index-lookup"`
+	// TxIndexBackend selects the tx-hash index implementation used by the
+	// parquet receipt store. Set to "pebbledb" (the default) to maintain a
+	// Pebble-backed tx_hash -> block_number index alongside parquet files so
+	// receipt-by-hash lookups can target a single file instead of scanning all
+	// files. Set to "" to disable the index and fall back to a full DuckDB scan.
+	// Ignored when the receipt backend is not parquet.
+	TxIndexBackend string `mapstructure:"tx-index-backend"`
 }
 
 // DefaultReceiptStoreConfig returns the default ReceiptStoreConfig
@@ -62,7 +78,7 @@ func DefaultReceiptStoreConfig() ReceiptStoreConfig {
 		AsyncWriteBuffer:     DefaultSSAsyncBuffer,
 		KeepRecent:           DefaultSSKeepRecent,
 		PruneIntervalSeconds: DefaultSSPruneInterval,
-		DisableTxIndexLookup: true,
+		TxIndexBackend:       ReceiptTxIndexBackendPebble,
 	}
 }
 
@@ -113,5 +129,12 @@ func ReadReceiptConfig(opts AppOptions) (ReceiptStoreConfig, error) {
 		}
 		cfg.PruneIntervalSeconds = pruneIntervalSeconds
 	}
+	if v := opts.Get(flagRSTxIndexBackend); v != nil {
+		txIndexBackend, err := cast.ToStringE(v)
+		if err != nil {
+			return cfg, err
+		}
+		cfg.TxIndexBackend = NormalizeReceiptTxIndexBackend(txIndexBackend)
+	}
 	return cfg, nil
 }

sei-db/config/receipt_config_test.go

@@ -21,3 +21,30 @@ func TestReadReceiptConfigRejectsMisnamedBackendKey(t *testing.T) {
 	require.ErrorContains(t, err, "receipt-store.backend")
 	require.ErrorContains(t, err, "receipt-store.rs-backend")
 }
+
+func TestReadReceiptConfigTxIndexBackendOverride(t *testing.T) {
+	cfg, err := ReadReceiptConfig(mapAppOpts{
+		"receipt-store.tx-index-backend": "",
+	})
+
+	require.NoError(t, err)
+	require.Equal(t, "", cfg.TxIndexBackend)
+}
+
+func TestReadReceiptConfigAcceptsPebbleDBTxIndexBackend(t *testing.T) {
+	cfg, err := ReadReceiptConfig(mapAppOpts{
+		"receipt-store.tx-index-backend": "pebbledb",
+	})
+
+	require.NoError(t, err)
+	require.Equal(t, ReceiptTxIndexBackendPebble, cfg.TxIndexBackend)
+}
+
+func TestReadReceiptConfigUnknownTxIndexBackendDefaultsToNone(t *testing.T) {
+	cfg, err := ReadReceiptConfig(mapAppOpts{
+		"receipt-store.tx-index-backend": "rocksdb",
+	})
+
+	require.NoError(t, err)
+	require.Equal(t, ReceiptTxIndexBackendNone, cfg.TxIndexBackend)
+}

sei-db/config/toml.go

@@ -160,6 +160,11 @@ keep-recent = {{ .ReceiptStore.KeepRecent }}
 # PruneIntervalSeconds defines the interval in seconds to trigger pruning.
 # defaults to 600 seconds
 prune-interval-seconds = {{ .ReceiptStore.PruneIntervalSeconds }}
+
+# TxIndexBackend selects the tx-hash index implementation for parquet receipts.
+# Set to "pebbledb" to enable the index, or "" to disable it.
+# Ignored unless rs-backend = "parquet".
+tx-index-backend = "{{ .ReceiptStore.TxIndexBackend }}"
 `
 
 // DefaultConfigTemplate combines both templates for backward compatibility

sei-db/config/toml_test.go

@@ -117,6 +117,7 @@ func TestReceiptStoreConfigTemplate(t *testing.T) {
 	require.Contains(t, output, "async-write-buffer =", "Missing async-write-buffer")
 	require.Contains(t, output, "keep-recent =", "Missing keep-recent")
 	require.Contains(t, output, "prune-interval-seconds =", "Missing prune-interval-seconds")
+	require.Contains(t, output, `tx-index-backend = "pebbledb"`, "Missing or incorrect tx-index-backend")
 	require.Contains(t, output, `Applies only when rs-backend = "pebbledb"`, "Missing pebble-only async-write-buffer note")
 	require.NotContains(t, output, "use-default-comparer", "use-default-comparer should not be in receipt-store template")
 }
@@ -151,6 +152,7 @@ func TestDefaultConfigTemplate(t *testing.T) {
 	require.Contains(t, output, "async-write-buffer =")
 	require.Contains(t, output, "keep-recent =")
 	require.Contains(t, output, "prune-interval-seconds =")
+	require.Contains(t, output, `tx-index-backend = "pebbledb"`)
 }
 
 // TestWriteModeValues verifies WriteMode enum values match template output

sei-db/ledger_db/parquet/reader.go

@@ -321,26 +321,81 @@ func (r *Reader) MaxReceiptBlockNumber(ctx context.Context) (uint64, bool, error
 	return uint64(max.Int64), true, nil
 }
 
-// GetReceiptByTxHash queries for a receipt by transaction hash.
+// GetReceiptByTxHash queries for a receipt by transaction hash across all
+// closed parquet files (full scan).
 func (r *Reader) GetReceiptByTxHash(ctx context.Context, txHash common.Hash) (*ReceiptResult, error) {
-	// Hold pruneMu first to prevent file deletion, then snapshot the list.
+	return r.getReceiptByTxHashFromFiles(ctx, txHash, nil)
+}
+
+// GetReceiptByTxHashInBlock narrows the search to the parquet file that
+// should contain blockNumber, falling back to a full scan on miss.
+func (r *Reader) GetReceiptByTxHashInBlock(ctx context.Context, txHash common.Hash, blockNumber uint64) (*ReceiptResult, error) {
+	// Hold pruneMu across candidate selection and the targeted query so a
+	// concurrent prune cannot delete the file between fileForBlock and the
+	// DuckDB read (see getReceiptByTxHashFromFilesLocked).
 	r.pruneMu.RLock()
-	defer r.pruneMu.RUnlock()
+	candidateFile := r.fileForBlock(blockNumber)
+
+	var result *ReceiptResult
+	var err error
+	if candidateFile != "" {
+		result, err = r.getReceiptByTxHashFromFilesLocked(ctx, txHash, []string{candidateFile})
+	}
+	r.pruneMu.RUnlock()
 
+	if err != nil {
+		return nil, err
+	}
+	if result != nil {
+		return result, nil
+	}
+	return r.getReceiptByTxHashFromFiles(ctx, txHash, nil)
+}
+
+// fileForBlock returns the receipt parquet file path that should contain
+// blockNumber, using the sorted tracked file list. Returns "" if no
+// candidate is found.
+func (r *Reader) fileForBlock(blockNumber uint64) string {
 	r.mu.RLock()
-	closedFiles := make([]string, len(r.closedReceiptFiles))
-	copy(closedFiles, r.closedReceiptFiles)
-	r.mu.RUnlock()
+	defer r.mu.RUnlock()
 
-	if len(closedFiles) == 0 {
+	var best string
+	for _, f := range r.closedReceiptFiles {
+		start := ExtractBlockNumber(f)
+		if start <= blockNumber {
+			best = f
+		} else {
+			break
+		}
+	}
+	return best
+}
+
+func (r *Reader) getReceiptByTxHashFromFiles(ctx context.Context, txHash common.Hash, files []string) (*ReceiptResult, error) {
+	r.pruneMu.RLock()
+	defer r.pruneMu.RUnlock()
+	return r.getReceiptByTxHashFromFilesLocked(ctx, txHash, files)
+}
+
+// getReceiptByTxHashFromFilesLocked runs the receipt query. The caller must
+// hold r.pruneMu.RLock (or otherwise ensure listed files are not deleted).
+func (r *Reader) getReceiptByTxHashFromFilesLocked(ctx context.Context, txHash common.Hash, files []string) (*ReceiptResult, error) {
+	if files == nil {
+		r.mu.RLock()
+		files = make([]string, len(r.closedReceiptFiles))
+		copy(files, r.closedReceiptFiles)
+		r.mu.RUnlock()
+	}
+
+	if len(files) == 0 {
 		return nil, nil
 	}
 
 	var parquetFiles string
-	if len(closedFiles) == 1 {
-		parquetFiles = quoteSQLString(closedFiles[0])
+	if len(files) == 1 {
+		parquetFiles = quoteSQLString(files[0])
 	} else {
-		parquetFiles = fmt.Sprintf("[%s]", joinQuoted(closedFiles))
+		parquetFiles = fmt.Sprintf("[%s]", joinQuoted(files))
 	}
 
 	// #nosec G201 -- parquetFiles derived from local file paths

sei-db/ledger_db/parquet/reader_filter_test.go

@@ -102,9 +102,9 @@ func TestStoreGetReceiptByTxHashWithoutIndex(t *testing.T) {
 	}
 
 	store, err := NewStore(StoreConfig{
-		DBDirectory:          dir,
-		MaxBlocksPerFile:     500,
-		DisableTxIndexLookup: true,
+		DBDirectory:      dir,
+		MaxBlocksPerFile: 500,
+		TxIndexBackend:   "none",
 	})
 	require.NoError(t, err)
 	defer func() { _ = store.Close() }()

sei-db/ledger_db/parquet/reader_race_test.go

@@ -82,10 +82,10 @@ func TestConcurrentReadsAndPrune(t *testing.T) {
 	}
 
 	store, err := NewStore(StoreConfig{
-		DBDirectory:          dir,
-		MaxBlocksPerFile:     500,
-		KeepRecent:           600,
-		DisableTxIndexLookup: true,
+		DBDirectory:      dir,
+		MaxBlocksPerFile: 500,
+		KeepRecent:       600,
+		TxIndexBackend:   "none",
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })
@@ -126,6 +126,56 @@ func TestConcurrentReadsAndPrune(t *testing.T) {
 	require.NoError(t, g.Wait())
 }
 
+// TestConcurrentTargetedReadsAndPrune exercises GetReceiptByTxHashInBlock
+// (single-file candidate path) concurrently with pruning. Without holding
+// pruneMu across fileForBlock and the targeted query, a prune could delete
+// the candidate file between those steps and DuckDB would error.
+func TestConcurrentTargetedReadsAndPrune(t *testing.T) {
+	dir := t.TempDir()
+
+	for _, start := range []uint64{0, 500, 1000} {
+		require.NoError(t, createTestReceiptFile(dir, start, 500))
+	}
+
+	store, err := NewStore(StoreConfig{
+		DBDirectory:      dir,
+		MaxBlocksPerFile: 500,
+		KeepRecent:       600,
+		TxIndexBackend:   "none",
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = store.Close() })
+
+	ctx := context.Background()
+	txHash := common.BigToHash(new(big.Int).SetUint64(250))
+
+	result, err := store.GetReceiptByTxHashInBlock(ctx, txHash, 250)
+	require.NoError(t, err)
+	require.NotNil(t, result)
+
+	const numReaders = 32
+	const readsPerReader = 100
+
+	g, _ := errgroup.WithContext(ctx)
+	for i := 0; i < numReaders; i++ {
+		g.Go(func() error {
+			for j := 0; j < readsPerReader; j++ {
+				if _, err := store.GetReceiptByTxHashInBlock(ctx, txHash, 250); err != nil {
+					return fmt.Errorf("GetReceiptByTxHashInBlock(250): %w", err)
+				}
+			}
+			return nil
+		})
+	}
+
+	g.Go(func() error {
+		store.PruneOldFiles(600)
+		return nil
+	})
+
+	require.NoError(t, g.Wait())
+}
+
 // TestOnFileRotationNotBlockedByPruneMu verifies the structural property
 // that OnFileRotation only acquires mu (the file-list lock), never pruneMu
 // (the file-lifetime lock). We hold pruneMu.RLock to simulate in-flight
@@ -135,9 +185,9 @@ func TestOnFileRotationNotBlockedByPruneMu(t *testing.T) {
 	require.NoError(t, createTestReceiptFile(dir, 0, 1))
 
 	store, err := NewStore(StoreConfig{
-		DBDirectory:          dir,
-		MaxBlocksPerFile:     500,
-		DisableTxIndexLookup: true,
+		DBDirectory:      dir,
+		MaxBlocksPerFile: 500,
+		TxIndexBackend:   "none",
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })
@@ -170,10 +220,10 @@ func TestConcurrentReadsPruneAndRotation(t *testing.T) {
 	}
 
 	store, err := NewStore(StoreConfig{
-		DBDirectory:          dir,
-		MaxBlocksPerFile:     500,
-		KeepRecent:           1000,
-		DisableTxIndexLookup: true,
+		DBDirectory:      dir,
+		MaxBlocksPerFile: 500,
+		KeepRecent:       1000,
+		TxIndexBackend:   "none",
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })

sei-db/ledger_db/parquet/store.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/parquet-go/parquet-go"
+	dbconfig "github.com/sei-protocol/sei-chain/sei-db/config"
 	dbwal "github.com/sei-protocol/sei-chain/sei-db/wal"
 	"github.com/sei-protocol/seilog"
 )
@@ -36,15 +37,15 @@ type StoreConfig struct {
 	PruneIntervalSeconds int64
 	BlockFlushInterval   uint64
 	MaxBlocksPerFile     uint64
-	DisableTxIndexLookup bool
+	TxIndexBackend       string
 }
 
 // DefaultStoreConfig returns the default store configuration.
 func DefaultStoreConfig() StoreConfig {
 	return StoreConfig{
-		BlockFlushInterval:   defaultBlockFlushInterval,
-		MaxBlocksPerFile:     defaultMaxBlocksPerFile,
-		DisableTxIndexLookup: true,
+		BlockFlushInterval: defaultBlockFlushInterval,
+		MaxBlocksPerFile:   defaultMaxBlocksPerFile,
+		TxIndexBackend:     dbconfig.ReceiptTxIndexBackendPebble,
 	}
 }
 
@@ -104,9 +105,6 @@ type Store struct {
 // NewStore creates a new parquet store.
 func NewStore(cfg StoreConfig) (*Store, error) {
 	storeCfg := resolveStoreConfig(cfg)
-	if !storeCfg.DisableTxIndexLookup {
-		panic("not implemented")
-	}
 
 	if err := os.MkdirAll(cfg.DBDirectory, 0o750); err != nil {
 		return nil, fmt.Errorf("failed to create parquet base directory: %w", err)
@@ -156,7 +154,9 @@ func resolveStoreConfig(cfg StoreConfig) StoreConfig {
 	resolved.DBDirectory = cfg.DBDirectory
 	resolved.KeepRecent = cfg.KeepRecent
 	resolved.PruneIntervalSeconds = cfg.PruneIntervalSeconds
-	resolved.DisableTxIndexLookup = cfg.DisableTxIndexLookup
+	if cfg.TxIndexBackend != "" {
+		resolved.TxIndexBackend = cfg.TxIndexBackend
+	}
 	if cfg.BlockFlushInterval > 0 {
 		resolved.BlockFlushInterval = cfg.BlockFlushInterval
 	}
@@ -200,6 +200,12 @@ func (s *Store) GetReceiptByTxHash(ctx context.Context, txHash common.Hash) (*Re
 	return s.Reader.GetReceiptByTxHash(ctx, txHash)
 }
 
+// GetReceiptByTxHashInBlock narrows the parquet search to the file containing
+// blockNumber, falling back to a full scan on miss.
+func (s *Store) GetReceiptByTxHashInBlock(ctx context.Context, txHash common.Hash, blockNumber uint64) (*ReceiptResult, error) {
+	return s.Reader.GetReceiptByTxHashInBlock(ctx, txHash, blockNumber)
+}
+
 // GetLogs retrieves logs matching the filter.
 func (s *Store) GetLogs(ctx context.Context, filter LogFilter) ([]LogResult, error) {
 	return s.Reader.GetLogs(ctx, filter)