Add receipt / log reads to cryptosim

sei-db/config/receipt_config.go

@@ -48,6 +48,11 @@ type ReceiptStoreConfig struct {
 	// PruneIntervalSeconds defines the interval in seconds to trigger pruning
 	// 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"`
 }
 
 // DefaultReceiptStoreConfig returns the default ReceiptStoreConfig
@@ -57,6 +62,7 @@ func DefaultReceiptStoreConfig() ReceiptStoreConfig {
 		AsyncWriteBuffer:     DefaultSSAsyncBuffer,
 		KeepRecent:           DefaultSSKeepRecent,
 		PruneIntervalSeconds: DefaultSSPruneInterval,
+		DisableTxIndexLookup: true,
 	}
 }
 

sei-db/ledger_db/parquet/reader.go

@@ -384,6 +384,9 @@ func (r *Reader) GetLogs(ctx context.Context, filter LogFilter) ([]LogResult, er
 		if filter.ToBlock != nil && startBlock > *filter.ToBlock {
 			continue
 		}
+		if filter.FromBlock != nil && startBlock+r.maxBlocksPerFile <= *filter.FromBlock {
+			continue
+		}
 		files = append(files, f)
 	}
 	if len(files) == 0 {

sei-db/ledger_db/parquet/reader_filter_test.go

@@ -0,0 +1,119 @@
+package parquet
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+	"os"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	pqgo "github.com/parquet-go/parquet-go"
+	"github.com/stretchr/testify/require"
+)
+
+func createTestLogFile(dir string, startBlock, count uint64) error {
+	path := fmt.Sprintf("%s/logs_%d.parquet", dir, startBlock)
+	f, err := os.Create(path)
+	if err != nil {
+		return err
+	}
+	w := pqgo.NewGenericWriter[LogRecord](f)
+	for i := uint64(0); i < count; i++ {
+		block := startBlock + i
+		txHash := common.BigToHash(new(big.Int).SetUint64(block))
+		if _, err := w.Write([]LogRecord{{
+			BlockNumber: block,
+			TxHash:      txHash[:],
+			Address:     common.HexToAddress("0xdead").Bytes(),
+		}}); err != nil {
+			return err
+		}
+	}
+	if err := w.Close(); err != nil {
+		return err
+	}
+	return f.Close()
+}
+
+func uint64Ptr(v uint64) *uint64 { return &v }
+
+func TestGetLogsPrunesFilesBelowFromBlock(t *testing.T) {
+	dir := t.TempDir()
+
+	for _, start := range []uint64{0, 500, 1000, 1500} {
+		require.NoError(t, createTestReceiptFile(dir, start, 500))
+		require.NoError(t, createTestLogFile(dir, start, 500))
+	}
+
+	reader, err := NewReaderWithMaxBlocksPerFile(dir, 500)
+	require.NoError(t, err)
+	defer func() { _ = reader.Close() }()
+
+	ctx := context.Background()
+
+	results, err := reader.GetLogs(ctx, LogFilter{
+		FromBlock: uint64Ptr(1200),
+		ToBlock:   uint64Ptr(1300),
+	})
+	require.NoError(t, err)
+
+	for _, r := range results {
+		require.GreaterOrEqual(t, r.BlockNumber, uint64(1200))
+		require.LessOrEqual(t, r.BlockNumber, uint64(1300))
+	}
+	require.Equal(t, 101, len(results), "should have blocks 1200-1300 inclusive")
+}
+
+func TestGetLogsPrunesBothEnds(t *testing.T) {
+	dir := t.TempDir()
+
+	for _, start := range []uint64{0, 500, 1000, 1500, 2000} {
+		require.NoError(t, createTestReceiptFile(dir, start, 500))
+		require.NoError(t, createTestLogFile(dir, start, 500))
+	}
+
+	reader, err := NewReaderWithMaxBlocksPerFile(dir, 500)
+	require.NoError(t, err)
+	defer func() { _ = reader.Close() }()
+
+	ctx := context.Background()
+
+	// Query blocks 1400-1600: should need overlapping files 1000 and 1500,
+	// but still prune non-overlapping files 0, 500, and 2000.
+	results, err := reader.GetLogs(ctx, LogFilter{
+		FromBlock: uint64Ptr(1400),
+		ToBlock:   uint64Ptr(1600),
+	})
+	require.NoError(t, err)
+
+	for _, r := range results {
+		require.GreaterOrEqual(t, r.BlockNumber, uint64(1400))
+		require.LessOrEqual(t, r.BlockNumber, uint64(1600))
+	}
+	require.Equal(t, 201, len(results), "should have blocks 1400-1600 inclusive")
+}
+
+func TestStoreGetReceiptByTxHashWithoutIndex(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,
+		DisableTxIndexLookup: true,
+	})
+	require.NoError(t, err)
+	defer func() { _ = store.Close() }()
+
+	txHash := common.BigToHash(new(big.Int).SetUint64(750))
+
+	ctx := context.Background()
+	result, err := store.GetReceiptByTxHash(ctx, txHash)
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.Equal(t, uint64(750), result.BlockNumber)
+}

sei-db/ledger_db/parquet/reader_race_test.go

@@ -82,9 +82,10 @@ func TestConcurrentReadsAndPrune(t *testing.T) {
 	}
 
 	store, err := NewStore(StoreConfig{
-		DBDirectory:      dir,
-		MaxBlocksPerFile: 500,
-		KeepRecent:       600,
+		DBDirectory:          dir,
+		MaxBlocksPerFile:     500,
+		KeepRecent:           600,
+		DisableTxIndexLookup: true,
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })
@@ -134,8 +135,9 @@ func TestOnFileRotationNotBlockedByPruneMu(t *testing.T) {
 	require.NoError(t, createTestReceiptFile(dir, 0, 1))
 
 	store, err := NewStore(StoreConfig{
-		DBDirectory:      dir,
-		MaxBlocksPerFile: 500,
+		DBDirectory:          dir,
+		MaxBlocksPerFile:     500,
+		DisableTxIndexLookup: true,
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })
@@ -168,9 +170,10 @@ func TestConcurrentReadsPruneAndRotation(t *testing.T) {
 	}
 
 	store, err := NewStore(StoreConfig{
-		DBDirectory:      dir,
-		MaxBlocksPerFile: 500,
-		KeepRecent:       1000,
+		DBDirectory:          dir,
+		MaxBlocksPerFile:     500,
+		KeepRecent:           1000,
+		DisableTxIndexLookup: true,
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })

sei-db/ledger_db/parquet/store.go

@@ -36,13 +36,15 @@ type StoreConfig struct {
 	PruneIntervalSeconds int64
 	BlockFlushInterval   uint64
 	MaxBlocksPerFile     uint64
+	DisableTxIndexLookup bool
 }
 
 // DefaultStoreConfig returns the default store configuration.
 func DefaultStoreConfig() StoreConfig {
 	return StoreConfig{
-		BlockFlushInterval: defaultBlockFlushInterval,
-		MaxBlocksPerFile:   defaultMaxBlocksPerFile,
+		BlockFlushInterval:   defaultBlockFlushInterval,
+		MaxBlocksPerFile:     defaultMaxBlocksPerFile,
+		DisableTxIndexLookup: true,
 	}
 }
 
@@ -102,6 +104,9 @@ 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)
@@ -151,6 +156,7 @@ func resolveStoreConfig(cfg StoreConfig) StoreConfig {
 	resolved.DBDirectory = cfg.DBDirectory
 	resolved.KeepRecent = cfg.KeepRecent
 	resolved.PruneIntervalSeconds = cfg.PruneIntervalSeconds
+	resolved.DisableTxIndexLookup = cfg.DisableTxIndexLookup
 	if cfg.BlockFlushInterval > 0 {
 		resolved.BlockFlushInterval = cfg.BlockFlushInterval
 	}
@@ -188,7 +194,8 @@ func (s *Store) SetBlockFlushInterval(interval uint64) {
 	s.config.BlockFlushInterval = interval
 }
 
-// GetReceiptByTxHash retrieves a receipt by transaction hash.
+// GetReceiptByTxHash retrieves a receipt by transaction hash via a full scan of
+// the closed parquet files tracked by the reader.
 func (s *Store) GetReceiptByTxHash(ctx context.Context, txHash common.Hash) (*ReceiptResult, error) {
 	return s.Reader.GetReceiptByTxHash(ctx, txHash)
 }
@@ -320,6 +327,7 @@ func (s *Store) Close() error {
 		}
 		if closeErr := s.Reader.Close(); closeErr != nil {
 			err = closeErr
+			return
 		}
 	})
 

sei-db/ledger_db/parquet/store_config_test.go

@@ -37,9 +37,10 @@ func (m *mockParquetWAL) Close() error { return nil }
 
 func TestNewStoreAppliesConfiguredIntervals(t *testing.T) {
 	store, err := NewStore(StoreConfig{
-		DBDirectory:        t.TempDir(),
-		BlockFlushInterval: 7,
-		MaxBlocksPerFile:   11,
+		DBDirectory:          t.TempDir(),
+		BlockFlushInterval:   7,
+		MaxBlocksPerFile:     11,
+		DisableTxIndexLookup: true,
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })
@@ -54,32 +55,46 @@ func TestNewStoreAppliesConfiguredIntervals(t *testing.T) {
 
 func TestNewStoreUsesDefaultIntervalsWhenUnset(t *testing.T) {
 	store, err := NewStore(StoreConfig{
-		DBDirectory: t.TempDir(),
+		DBDirectory:          t.TempDir(),
+		DisableTxIndexLookup: true,
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })
 
 	require.Equal(t, defaultBlockFlushInterval, store.config.BlockFlushInterval)
 	require.Equal(t, defaultMaxBlocksPerFile, store.config.MaxBlocksPerFile)
 	require.Equal(t, defaultMaxBlocksPerFile, store.CacheRotateInterval())
+	require.True(t, store.config.DisableTxIndexLookup)
 }
 
 func TestNewStorePreservesKeepRecentAndPruneIntervalSettings(t *testing.T) {
 	store, err := NewStore(StoreConfig{
 		DBDirectory:          t.TempDir(),
 		KeepRecent:           123,
 		PruneIntervalSeconds: 9,
+		DisableTxIndexLookup: true,
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })
 
 	require.Equal(t, int64(123), store.config.KeepRecent)
 	require.Equal(t, int64(9), store.config.PruneIntervalSeconds)
+	require.True(t, store.config.DisableTxIndexLookup)
+}
+
+func TestNewStorePanicsWhenTxIndexLookupEnabled(t *testing.T) {
+	require.PanicsWithValue(t, "not implemented", func() {
+		_, _ = NewStore(StoreConfig{
+			DBDirectory:          t.TempDir(),
+			DisableTxIndexLookup: false,
+		})
+	})
 }
 
 func TestPruneOldFilesKeepsTrackingOnDeleteFailure(t *testing.T) {
 	store, err := NewStore(StoreConfig{
-		DBDirectory: t.TempDir(),
+		DBDirectory:          t.TempDir(),
+		DisableTxIndexLookup: true,
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })
@@ -133,7 +148,8 @@ func TestCorruptLastFileDeletedOnStartup(t *testing.T) {
 	require.NoError(t, os.WriteFile(corruptLog, []byte("not a parquet file"), 0o644))
 
 	store, err := NewStore(StoreConfig{
-		DBDirectory: dir,
+		DBDirectory:          dir,
+		DisableTxIndexLookup: true,
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })
@@ -165,7 +181,8 @@ func TestCorruptLogFileUntracksReceiptCounterpart(t *testing.T) {
 	require.NoError(t, os.WriteFile(corruptLog, []byte("not a parquet file"), 0o644))
 
 	store, err := NewStore(StoreConfig{
-		DBDirectory: dir,
+		DBDirectory:          dir,
+		DisableTxIndexLookup: true,
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })
@@ -184,7 +201,8 @@ func TestLazyInitCreatesFileOnFirstWrite(t *testing.T) {
 	dir := t.TempDir()
 
 	store, err := NewStore(StoreConfig{
-		DBDirectory: dir,
+		DBDirectory:          dir,
+		DisableTxIndexLookup: true,
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = store.Close() })

sei-db/ledger_db/parquet/wal.go

@@ -113,6 +113,9 @@ func NewWAL(dir string) (dbwal.GenericWAL[WALEntry], error) {
 		encodeWALEntry,
 		decodeWALEntry,
 		dir,
-		dbwal.Config{},
+		dbwal.Config{
+			// Allow the WAL to be fully emptied after rotation/truncation.
+			AllowEmpty: true,
+		},
 	)
 }