Add receipt / log reads to cryptosim

sei-db/ledger_db/receipt/cached_receipt_store.go

@@ -28,9 +28,10 @@ type cachedReceiptStore struct {
 	cacheRotateInterval uint64
 	cacheNextRotate     uint64
 	cacheMu             sync.Mutex
+	readObserver        ReceiptReadObserver
 }
 
-func newCachedReceiptStore(backend ReceiptStore) ReceiptStore {
+func newCachedReceiptStore(backend ReceiptStore, observer ReceiptReadObserver) ReceiptStore {
 	if backend == nil {
 		return nil
 	}
@@ -44,13 +45,34 @@ func newCachedReceiptStore(backend ReceiptStore) ReceiptStore {
 		backend:             backend,
 		cache:               newLedgerCache(),
 		cacheRotateInterval: interval,
+		readObserver:        observer,
 	}
 	if provider, ok := backend.(cacheWarmupProvider); ok {
 		store.cacheReceipts(provider.warmupReceipts())
 	}
 	return store
 }
 
+// StableReceiptCacheWindowBlocks returns the near-tip block window that is
+// guaranteed to stay in the active write chunk until the next rotation.
+func StableReceiptCacheWindowBlocks(store ReceiptStore) uint64 {
+	cached, ok := store.(*cachedReceiptStore)
+	if !ok || cached.cacheRotateInterval == 0 {
+		return 0
+	}
+	return cached.cacheRotateInterval
+}
+
+// EstimatedReceiptCacheWindowBlocks returns the approximate recent block window
+// normally served by the in-memory receipt cache (current chunk + previous one).
+func EstimatedReceiptCacheWindowBlocks(store ReceiptStore) uint64 {
+	hotWindow := StableReceiptCacheWindowBlocks(store)
+	if hotWindow == 0 {
+		return 0
+	}
+	return hotWindow * uint64(numCacheChunks-1)
+}
+
 func (s *cachedReceiptStore) LatestVersion() int64 {
 	return s.backend.LatestVersion()
 }
@@ -65,15 +87,19 @@ func (s *cachedReceiptStore) SetEarliestVersion(version int64) error {
 
 func (s *cachedReceiptStore) GetReceipt(ctx sdk.Context, txHash common.Hash) (*types.Receipt, error) {
 	if receipt, ok := s.cache.GetReceipt(txHash); ok {
+		s.reportCacheHit()
 		return receipt, nil
 	}
+	s.reportCacheMiss()
 	return s.backend.GetReceipt(ctx, txHash)
 }
 
 func (s *cachedReceiptStore) GetReceiptFromStore(ctx sdk.Context, txHash common.Hash) (*types.Receipt, error) {
 	if receipt, ok := s.cache.GetReceipt(txHash); ok {
+		s.reportCacheHit()
 		return receipt, nil
 	}
+	s.reportCacheMiss()
 	return s.backend.GetReceiptFromStore(ctx, txHash)
 }
 
@@ -99,8 +125,10 @@ func (s *cachedReceiptStore) FilterLogs(ctx sdk.Context, fromBlock, toBlock uint
 
 	// Merge results, avoiding duplicates by tracking seen (blockNum, txIndex, logIndex)
 	if len(cacheLogs) == 0 {
+		s.reportLogFilterCacheMiss()
 		return backendLogs, nil
 	}
+	s.reportLogFilterCacheHit()
 	if len(backendLogs) == 0 {
 		sortLogs(cacheLogs)
 		return cacheLogs, nil
@@ -220,3 +248,27 @@ func (s *cachedReceiptStore) maybeRotateCacheLocked(blockNumber uint64) {
 		s.cacheNextRotate += s.cacheRotateInterval
 	}
 }
+
+func (s *cachedReceiptStore) reportCacheHit() {
+	if s.readObserver != nil {
+		s.readObserver.ReportReceiptCacheHit()
+	}
+}
+
+func (s *cachedReceiptStore) reportCacheMiss() {
+	if s.readObserver != nil {
+		s.readObserver.ReportReceiptCacheMiss()
+	}
+}
+
+func (s *cachedReceiptStore) reportLogFilterCacheHit() {
+	if s.readObserver != nil {
+		s.readObserver.ReportLogFilterCacheHit()
+	}
+}
+
+func (s *cachedReceiptStore) reportLogFilterCacheMiss() {
+	if s.readObserver != nil {
+		s.readObserver.ReportLogFilterCacheMiss()
+	}
+}

sei-db/ledger_db/receipt/cached_receipt_store_test.go

@@ -18,6 +18,29 @@ type fakeReceiptBackend struct {
 	filterLogCalls  int
 }
 
+type fakeReceiptReadObserver struct {
+	cacheHits            int
+	cacheMisses          int
+	logFilterCacheHits   int
+	logFilterCacheMisses int
+}
+
+func (f *fakeReceiptReadObserver) ReportReceiptCacheHit() {
+	f.cacheHits++
+}
+
+func (f *fakeReceiptReadObserver) ReportReceiptCacheMiss() {
+	f.cacheMisses++
+}
+
+func (f *fakeReceiptReadObserver) ReportLogFilterCacheHit() {
+	f.logFilterCacheHits++
+}
+
+func (f *fakeReceiptReadObserver) ReportLogFilterCacheMiss() {
+	f.logFilterCacheMisses++
+}
+
 func newFakeReceiptBackend() *fakeReceiptBackend {
 	return &fakeReceiptBackend{
 		receipts: make(map[common.Hash]*types.Receipt),
@@ -70,7 +93,7 @@ func (f *fakeReceiptBackend) Close() error {
 func TestCachedReceiptStoreUsesCacheForReceipt(t *testing.T) {
 	ctx, _ := newTestContext()
 	backend := newFakeReceiptBackend()
-	store := newCachedReceiptStore(backend)
+	store := newCachedReceiptStore(backend, nil)
 
 	txHash := common.HexToHash("0x1")
 	addr := common.HexToAddress("0x100")
@@ -89,7 +112,7 @@ func TestCachedReceiptStoreUsesCacheForReceipt(t *testing.T) {
 func TestCachedReceiptStoreFilterLogsDelegates(t *testing.T) {
 	ctx, _ := newTestContext()
 	backend := newFakeReceiptBackend()
-	store := newCachedReceiptStore(backend)
+	store := newCachedReceiptStore(backend, nil)
 
 	txHash := common.HexToHash("0x2")
 	addr := common.HexToAddress("0x200")
@@ -124,7 +147,7 @@ func TestCachedReceiptStoreFilterLogsReturnsSortedLogs(t *testing.T) {
 			Index:       0,
 		},
 	}
-	store := newCachedReceiptStore(backend)
+	store := newCachedReceiptStore(backend, nil)
 
 	receiptA := makeTestReceipt(common.HexToHash("0xa"), 11, 1, common.HexToAddress("0x210"), []common.Hash{common.HexToHash("0x1")})
 	receiptB := makeTestReceipt(common.HexToHash("0xb"), 11, 0, common.HexToAddress("0x220"), []common.Hash{common.HexToHash("0x2")})
@@ -143,3 +166,36 @@ func TestCachedReceiptStoreFilterLogsReturnsSortedLogs(t *testing.T) {
 	require.Equal(t, uint(1), logs[2].TxIndex)
 	require.Equal(t, uint64(12), logs[3].BlockNumber)
 }
+
+func TestCachedReceiptStoreReportsCacheHit(t *testing.T) {
+	ctx, _ := newTestContext()
+	backend := newFakeReceiptBackend()
+	observer := &fakeReceiptReadObserver{}
+	store := newCachedReceiptStore(backend, observer)
+
+	txHash := common.HexToHash("0x10")
+	receipt := makeTestReceipt(txHash, 7, 1, common.HexToAddress("0x100"), nil)
+
+	require.NoError(t, store.SetReceipts(ctx, []ReceiptRecord{{TxHash: txHash, Receipt: receipt}}))
+
+	backend.getReceiptCalls = 0
+	got, err := store.GetReceipt(ctx, txHash)
+	require.NoError(t, err)
+	require.Equal(t, receipt.TxHashHex, got.TxHashHex)
+	require.Equal(t, 0, backend.getReceiptCalls)
+	require.Equal(t, 1, observer.cacheHits)
+	require.Equal(t, 0, observer.cacheMisses)
+}
+
+func TestCachedReceiptStoreReportsCacheMiss(t *testing.T) {
+	ctx, _ := newTestContext()
+	backend := newFakeReceiptBackend()
+	observer := &fakeReceiptReadObserver{}
+	store := newCachedReceiptStore(backend, observer)
+
+	_, err := store.GetReceipt(ctx, common.HexToHash("0x404"))
+	require.ErrorIs(t, err, ErrNotFound)
+	require.Equal(t, 1, backend.getReceiptCalls)
+	require.Equal(t, 0, observer.cacheHits)
+	require.Equal(t, 1, observer.cacheMisses)
+}

sei-db/ledger_db/receipt/parquet_store.go

@@ -90,6 +90,9 @@ func (s *parquetReceiptStore) GetReceipt(ctx sdk.Context, txHash common.Hash) (*
 		return receipt, nil
 	}
 
+	if s.storeKey == nil {
+		return nil, ErrNotFound
+	}
 	store := ctx.KVStore(s.storeKey)
 	bz := store.Get(types.ReceiptKey(txHash))
 	if bz == nil {

sei-db/ledger_db/receipt/receipt_store.go

@@ -53,6 +53,15 @@ type ReceiptRecord struct {
 	ReceiptBytes []byte // Optional pre-marshaled receipt (must match Receipt if set)
 }
 
+// ReceiptReadObserver receives callbacks when cached receipt lookups either hit
+// the in-memory ledger cache or fall through to the backend store.
+type ReceiptReadObserver interface {
+	ReportReceiptCacheHit()
+	ReportReceiptCacheMiss()
+	ReportLogFilterCacheHit()
+	ReportLogFilterCacheMiss()
+}
+
 type receiptStore struct {
 	db          seidbtypes.StateStore
 	storeKey    sdk.StoreKey
@@ -78,11 +87,21 @@ func normalizeReceiptBackend(backend string) string {
 }
 
 func NewReceiptStore(config dbconfig.ReceiptStoreConfig, storeKey sdk.StoreKey) (ReceiptStore, error) {
+	return NewReceiptStoreWithReadObserver(config, storeKey, nil)
+}
+
+// NewReceiptStoreWithReadObserver constructs a receipt store and optionally
+// reports cache hits/misses for receipt-by-hash reads via observer callbacks.
+func NewReceiptStoreWithReadObserver(
+	config dbconfig.ReceiptStoreConfig,
+	storeKey sdk.StoreKey,
+	observer ReceiptReadObserver,
+) (ReceiptStore, error) {
 	backend, err := newReceiptBackend(config, storeKey)
 	if err != nil {
 		return nil, err
 	}
-	return newCachedReceiptStore(backend), nil
+	return newCachedReceiptStore(backend, observer), nil
 }
 
 // BackendTypeName returns the backend implementation name ("parquet" or "pebble") for testing.

sei-db/state_db/bench/cryptosim/canned_random.go

@@ -91,6 +91,9 @@ func (cr *CannedRandom) Bytes(count int) []byte {
 
 // Returns a slice of random bytes from a given seed. Bytes are deterministic given the same seed.
 //
+// Unlike most CannedRandom methods, SeededBytes is safe for concurrent use: it only reads
+// from the immutable buffer and does not advance the internal index.
+//
 // Returned slice is NOT safe to modify. If modification is required, the caller should make a copy of the slice.
 func (cr *CannedRandom) SeededBytes(count int, seed int64) []byte {
 	if count < 0 {

sei-db/state_db/bench/cryptosim/config/reciept-store.json

@@ -1,8 +1,15 @@
 {
-    "Comment": "For testing with the state store and reciept store both enabled.",
+    "Comment": "For testing with the state store and receipt store both enabled, with concurrent reads, pruning, and cache.",
     "DataDir": "data",
+    "LogDir": "logs",
+    "LogLevel": "info",
     "MinimumNumberOfColdAccounts": 1000000,
     "MinimumNumberOfDormantAccounts": 1000000,
-    "GenerateReceipts": true
+    "GenerateReceipts": true,
+    "ReceiptReadConcurrency": 4,
+    "ReceiptReadsPerSecond": 1000,
+    "ReceiptColdReadRatio": 0.1,
+    "LogFilterReadConcurrency": 4,
+    "LogFilterReadsPerSecond": 200,
+    "LogFilterColdReadRatio": 0.1
 }
-

sei-db/state_db/bench/cryptosim/cryptosim_config.go

@@ -158,6 +158,41 @@ type CryptoSimConfig struct {
 	// If greater than 0, the benchmark will throttle the transaction rate to this value, in hertz.
 	MaxTPS float64
 
+	// Number of concurrent reader goroutines issuing receipt lookups. 0 disables reads.
+	ReceiptReadConcurrency int
+
+	// Target total receipt reads per second across all reader goroutines.
+	// Reads are distributed evenly across readers.
+	ReceiptReadsPerSecond int
+
+	// Fraction of single-receipt reads that intentionally target blocks older
+	// than the in-memory receipt-cache window (0.0-1.0).
+	// These reads should mostly miss cache and fall through to DuckDB.
+	ReceiptColdReadRatio float64
+
+	// Deprecated: ignored when LogFilterReadConcurrency > 0. Previously controlled
+	// the fraction of shared reads that were log filters vs receipt lookups.
+	ReceiptLogFilterRatio float64
+
+	// Number of concurrent goroutines issuing log filter (eth_getLogs) queries. 0 disables log filter reads.
+	// These goroutines are independent from the receipt reader goroutines.
+	LogFilterReadConcurrency int
+
+	// Target total log filter reads per second across all log filter goroutines.
+	LogFilterReadsPerSecond int
+
+	// Fraction of log filter reads that intentionally target blocks older than
+	// the in-memory cache window (0.0-1.0). These cold reads miss the cache
+	// and fall through to DuckDB on closed parquet files. The remaining
+	// (1 - LogFilterColdReadRatio) reads target recent blocks likely in cache.
+	// Default 0.1 yields ~90% cache hit ratio.
+	LogFilterColdReadRatio float64
+
+	// Exponent controlling the recency bias within the chosen hot/cold read range.
+	// 1.0 = uniform within that range; higher values skew reads toward the newest
+	// blocks in the selected bucket.
+	ReceiptReadRecencyExponent float64
+
 	// Number of recent blocks to keep before pruning parquet files. 0 disables pruning.
 	ReceiptKeepRecent int64
 
@@ -215,6 +250,14 @@ func DefaultCryptoSimConfig() *CryptoSimConfig {
 		RecieptChannelCapacity:            32,
 		DisableTransactionExecution:       false,
 		MaxTPS:                            0,
+		ReceiptReadConcurrency:            0,
+		ReceiptReadsPerSecond:             100,
+		ReceiptColdReadRatio:              0.1,
+		ReceiptLogFilterRatio:             0,
+		LogFilterReadConcurrency:          0,
+		LogFilterReadsPerSecond:           100,
+		LogFilterColdReadRatio:            0.1,
+		ReceiptReadRecencyExponent:        3.0,
 		ReceiptKeepRecent:                 100_000,
 		ReceiptPruneIntervalSeconds:       600,
 		LogLevel:                          "info",
@@ -302,12 +345,29 @@ func (c *CryptoSimConfig) Validate() error {
 	if c.MaxTPS < 0 {
 		return fmt.Errorf("MaxTPS must be non-negative (got %f)", c.MaxTPS)
 	}
+	if c.ReceiptColdReadRatio < 0 || c.ReceiptColdReadRatio > 1 {
+		return fmt.Errorf("ReceiptColdReadRatio must be in [0, 1] (got %f)", c.ReceiptColdReadRatio)
+	}
+	if c.ReceiptLogFilterRatio < 0 || c.ReceiptLogFilterRatio > 1 {
+		return fmt.Errorf("ReceiptLogFilterRatio must be in [0, 1] (got %f)", c.ReceiptLogFilterRatio)
+	}
+	if c.LogFilterReadConcurrency < 0 {
+		return fmt.Errorf("LogFilterReadConcurrency must be non-negative (got %d)", c.LogFilterReadConcurrency)
+	}
+	if c.LogFilterColdReadRatio < 0 || c.LogFilterColdReadRatio > 1 {
+		return fmt.Errorf("LogFilterColdReadRatio must be in [0, 1] (got %f)", c.LogFilterColdReadRatio)
+	}
+	if c.ReceiptReadConcurrency < 0 {
+		return fmt.Errorf("ReceiptReadConcurrency must be non-negative (got %d)", c.ReceiptReadConcurrency)
+	}
+	if c.ReceiptReadRecencyExponent < 1 {
+		return fmt.Errorf("ReceiptReadRecencyExponent must be >= 1.0 (got %f)", c.ReceiptReadRecencyExponent)
+	}
 	switch strings.ToLower(c.LogLevel) {
 	case "debug", "info", "warn", "error":
 	default:
 		return fmt.Errorf("LogLevel must be one of debug, info, warn, error (got %q)", c.LogLevel)
 	}
-
 	return nil
 }