From 235dda51aaf951dd20dcdd0088c1bcdfc0038f42 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 13:52:56 -0700 Subject: [PATCH 01/11] Make receipt write fully async --- .../dashboards/gigasim-dashboard.json | 33 ++++- evmrpc/setup_test.go | 28 ++++- evmrpc/simulate_test.go | 3 +- evmrpc/tests/utils.go | 16 ++- .../cryptosim/reciept_store_simulator.go | 4 - sei-db/bootstrap/recovery_test.go | 8 +- sei-db/config/receipt_config.go | 5 +- .../receipt/litt_ctx_internal_test.go | 6 + .../ledger_db/receipt/litt_receipt_store.go | 114 ++++++++++++++++-- sei-db/ledger_db/receipt/littidx_test.go | 71 +++++++++++ .../receipt/offline_internal_test.go | 1 + .../receipt/receipt_bench_read_test.go | 5 + sei-db/ledger_db/receipt/receipt_store.go | 7 +- .../ledger_db/receipt/receipt_store_test.go | 12 +- sei-db/ledger_db/receipt/test_helpers_test.go | 12 ++ 15 files changed, 290 insertions(+), 35 deletions(-) diff --git a/docker/monitornode/dashboards/gigasim-dashboard.json b/docker/monitornode/dashboards/gigasim-dashboard.json index b3f01c2688..57817bfffa 100644 --- a/docker/monitornode/dashboards/gigasim-dashboard.json +++ b/docker/monitornode/dashboards/gigasim-dashboard.json @@ -313,8 +313,8 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "litt_control_queue_depth{table=\"receipts\"}", - "legendFormat": "control loop", + "expr": "receipt_write_queue_depth", + "legendFormat": "receipt write (whole write)", "range": true }, "version": "v0" @@ -334,24 +334,45 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "litt_flush_queue_depth{table=\"receipts\"}", - "legendFormat": "flush loop", + "expr": "litt_control_queue_depth{table=\"receipts\"}", + "legendFormat": "litt control loop", "range": true }, "version": "v0" }, "refId": "B" } + }, + { + "kind": "PanelQuery", + "spec": { + "hidden": false, + "query": { + "datasource": { + "name": "PBFA97CFB590B2093" + }, + "group": "prometheus", + "kind": "DataQuery", + "spec": { + "editorMode": "code", + "expr": "litt_flush_queue_depth{table=\"receipts\"}", + "legendFormat": "litt flush loop", + "range": true + }, + "version": "v0" + }, + "refId": "C" + } } ], "queryOptions": {}, "transformations": [] } }, - "description": "LittDB's write path for receipt bodies. Receipt writes block on the control loop when it fills.", + "description": "Blocks waiting for the receipt writer. SetReceipts queues the whole write — bodies, log index and version marker — and returns, so this depth is the depth of the receipt write itself. The litt series are the table queues downstream of it.", "id": 23, "links": [], - "title": "ReceiptDB Queue Depth", + "title": "Receipt Write Queue Depth", "vizConfig": { "group": "timeseries", "kind": "VizConfig", diff --git a/evmrpc/setup_test.go b/evmrpc/setup_test.go index 28dc5a279c..57c4ad89d8 100644 --- a/evmrpc/setup_test.go +++ b/evmrpc/setup_test.go @@ -31,6 +31,7 @@ import ( "github.com/sei-protocol/sei-chain/sei-cosmos/crypto/hd" sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" sdkerrors "github.com/sei-protocol/sei-chain/sei-cosmos/types/errors" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/bytes" tmutils "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" @@ -62,6 +63,26 @@ const MockHeight103 = 103 const MockHeight101 = 101 const MockHeight100 = 100 +// receiptVersionPinner is implemented by receipt stores whose version markers can be written +// directly. SetReceipts carries those markers, so the store's interface does not expose them. +type receiptVersionPinner interface { + SetLatestVersion(version int64) error + SetEarliestVersion(version int64) error +} + +// pinReceiptVersions widens a store's queryable window to [1, latest]. These tests seed receipts +// by other means, so nothing has advanced the markers a read is gated on. +func pinReceiptVersions(store receipt.ReceiptStore, latest int64) error { + pinner, ok := store.(receiptVersionPinner) + if !ok { + return fmt.Errorf("receipt store %T cannot pin versions", store) + } + if err := pinner.SetLatestVersion(latest); err != nil { + return err + } + return pinner.SetEarliestVersion(1) +} + // LatestCtxUpgradeName makes the test ctx look like a real chain that has // applied a post-v5.8.0 upgrade. The default Ctx has empty // ClosestUpgradeName and semver.Compare("", "v5.8.0") returns -1 (treated @@ -660,11 +681,9 @@ func init() { } testApp.Commit(context.Background()) if store := EVMKeeper.ReceiptStore(); store != nil { - latest := int64(math.MaxInt64) - if err := store.SetLatestVersion(latest); err != nil { + if err := pinReceiptVersions(store, math.MaxInt64); err != nil { panic(err) } - _ = store.SetEarliestVersion(1) } ctxProvider := func(height int64) sdk.Context { if height == MockHeight2 { @@ -1263,10 +1282,9 @@ func setupLogs() { EVMKeeper.SetEvmOnlyBlockBloom(Ctx, []ethtypes.Bloom{bloom4, bloomTx1}) if store := EVMKeeper.ReceiptStore(); store != nil { - if err := store.SetLatestVersion(MockHeight103); err != nil { + if err := pinReceiptVersions(store, MockHeight103); err != nil { panic(err) } - _ = store.SetEarliestVersion(1) } } diff --git a/evmrpc/simulate_test.go b/evmrpc/simulate_test.go index a89bffbce3..c42e722b76 100644 --- a/evmrpc/simulate_test.go +++ b/evmrpc/simulate_test.go @@ -47,8 +47,7 @@ import ( func primeReceiptStore(t *testing.T, store receipt.ReceiptStore, latest int64) { t.Helper() - require.NoError(t, store.SetLatestVersion(latest)) - require.NoError(t, store.SetEarliestVersion(1)) + require.NoError(t, pinReceiptVersions(store, latest)) } // bcAlwaysFailClient fails every Block call (header resolution uses a single block fetch). diff --git a/evmrpc/tests/utils.go b/evmrpc/tests/utils.go index a367663ea5..fb8ced197d 100644 --- a/evmrpc/tests/utils.go +++ b/evmrpc/tests/utils.go @@ -187,11 +187,21 @@ func setupTestServer( } pinStateStoreLatestVersion(a, ctxProvider) if store := a.EvmKeeper.ReceiptStore(); store != nil { - latest := int64(math.MaxInt64) - if err := store.SetLatestVersion(latest); err != nil { + // SetReceipts carries the version markers, so they are off the store's interface. These + // tests seed receipts by other means and would otherwise read against an unset window. + pinner, ok := store.(interface { + SetLatestVersion(version int64) error + SetEarliestVersion(version int64) error + }) + if !ok { + panic(fmt.Sprintf("receipt store %T cannot pin versions", store)) + } + if err := pinner.SetLatestVersion(math.MaxInt64); err != nil { + panic(err) + } + if err := pinner.SetEarliestVersion(1); err != nil { panic(err) } - _ = store.SetEarliestVersion(1) } return TestServer{EVMServer: s, port: port, mockClient: mockClient, app: a, ctxProvider: ctxProvider} } diff --git a/sei-db/bench/cryptosim/reciept_store_simulator.go b/sei-db/bench/cryptosim/reciept_store_simulator.go index 3a0c5aa117..b1bfdc8d7e 100644 --- a/sei-db/bench/cryptosim/reciept_store_simulator.go +++ b/sei-db/bench/cryptosim/reciept_store_simulator.go @@ -251,10 +251,6 @@ func (r *RecieptStoreSimulator) processBlock(blk *block) { for _, entry := range ringEntries { r.txRing.Push(entry.txHash, blockNumber, entry.contractAddress) } - - if err := r.store.SetLatestVersion(int64(blockNumber)); err != nil { //nolint:gosec - fmt.Printf("failed to update latest version for block %d: %v\n", blockNumber, err) - } } // startReceiptReaders launches dedicated goroutines for receipt-by-hash lookups. diff --git a/sei-db/bootstrap/recovery_test.go b/sei-db/bootstrap/recovery_test.go index 4c10cc72e5..fa652c6718 100644 --- a/sei-db/bootstrap/recovery_test.go +++ b/sei-db/bootstrap/recovery_test.go @@ -235,7 +235,13 @@ func TestRecoverStoresAtAZeroTargetLeavesReceiptsAlone(t *testing.T) { func TestFindTargetRecoveryHeightIsZeroWithoutABlockLedger(t *testing.T) { manager, _ := openManager(t, nil) commitBlocks(t, manager, 3) - require.NoError(t, manager.ReceiptDB().SetLatestVersion(3)) + // The version marker rides SetReceipts, so it is off the store's interface; this test stamps a + // head without bodies on purpose. + pinner, ok := manager.ReceiptDB().(interface { + SetLatestVersion(version int64) error + }) + require.True(t, ok) + require.NoError(t, pinner.SetLatestVersion(3)) // findTargetRecoveryHeight reads the state and receipt directories offline, so both stores have // to be closed for it. closeStateDB(t, manager) diff --git a/sei-db/config/receipt_config.go b/sei-db/config/receipt_config.go index f8ae9df29b..38cfa63ea2 100644 --- a/sei-db/config/receipt_config.go +++ b/sei-db/config/receipt_config.go @@ -46,8 +46,9 @@ type ReceiptStoreConfig struct { // defaults to pebbledb Backend string `mapstructure:"rs-backend"` - // AsyncWriteBuffer defines the async queue length for commits to be applied to receipt store - // Applies only to the pebbledb backend. + // AsyncWriteBuffer defines the async queue length for commits to be applied to receipt store. + // It bounds how many blocks the store may fall behind the chain before a write blocks, and so + // how far LatestVersion may trail the height just written. // Set <= 0 for synchronous writes. // defaults to 100 AsyncWriteBuffer int `mapstructure:"async-write-buffer"` diff --git a/sei-db/ledger_db/receipt/litt_ctx_internal_test.go b/sei-db/ledger_db/receipt/litt_ctx_internal_test.go index 1f10a1341c..6e4cdbd467 100644 --- a/sei-db/ledger_db/receipt/litt_ctx_internal_test.go +++ b/sei-db/ledger_db/receipt/litt_ctx_internal_test.go @@ -77,6 +77,7 @@ func TestBlockLogsReturnsCanceledContextBeforeScanning(t *testing.T) { topic := common.HexToHash("0xdef1") txHash, rcpt := littCtxTestReceipt(1, 0, addr, topic, 1) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(1), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, 1) ctx, cancel := context.WithCancel(context.Background()) cancel() @@ -95,6 +96,7 @@ func TestCandidateBlockLogsReturnsCanceledContextBeforeTx(t *testing.T) { topic := common.HexToHash("0xdef2") txHash, rcpt := littCtxTestReceipt(2, 0, addr, topic, 1) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(2), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, 2) candidates, err := s.blockTagCandidates(2, criteriaTagGroups(filters.FilterCriteria{})) require.NoError(t, err) @@ -121,6 +123,7 @@ func TestCandidateBlockLogsCancelsMidLoopOverLogs(t *testing.T) { topic := common.HexToHash("0xdef3") txHash, rcpt := littCtxTestReceipt(3, 0, addr, topic, 5) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(3), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, 3) candidates, err := s.blockTagCandidates(3, criteriaTagGroups(filters.FilterCriteria{})) require.NoError(t, err) @@ -147,6 +150,7 @@ func TestCandidateBlockLogsTripsBudgetMidLoopOverLogs(t *testing.T) { topic := common.HexToHash("0xdef4") txHash, rcpt := littCtxTestReceipt(4, 0, addr, topic, 2) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(4), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, 4) candidates, err := s.blockTagCandidates(4, criteriaTagGroups(filters.FilterCriteria{})) require.NoError(t, err) @@ -173,6 +177,7 @@ func TestFilterLogsByTagsPreCanceledContextReturnsEmptyFast(t *testing.T) { for block := uint64(1); block <= 5; block++ { txHash, rcpt := littCtxTestReceipt(block, 0, addr, topic, 1) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(block), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, int64(block)) } ctx, cancel := context.WithCancel(context.Background()) @@ -196,6 +201,7 @@ func TestFilterLogsThreadsSDKContext(t *testing.T) { topic := common.HexToHash("0xdef6") txHash, rcpt := littCtxTestReceipt(6, 0, addr, topic, 1) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(6), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, 6) crit := filters.FilterCriteria{Addresses: []common.Address{addr}} diff --git a/sei-db/ledger_db/receipt/litt_receipt_store.go b/sei-db/ledger_db/receipt/litt_receipt_store.go index a645587369..c82642b453 100644 --- a/sei-db/ledger_db/receipt/litt_receipt_store.go +++ b/sei-db/ledger_db/receipt/litt_receipt_store.go @@ -62,6 +62,9 @@ import ( // - unset: the background pruner below keeps the last KeepRecent blocks. // - set: the StorageGarbageCollector prunes through the gc.PrunableStore // implementation in litt_receipt_gc.go, and startPruning stands down. +// +// Writes are applied in the background: SetReceipts queues a block and returns, so a receipt is not +// necessarily readable the moment it returns. Sync waits for the queue to empty. type littReceiptStore struct { values litt.DB receipts litt.Table @@ -79,12 +82,30 @@ type littReceiptStore struct { backgroundWg sync.WaitGroup closeOnce sync.Once - // Breaks a write into its stages. The receipt bodies go to litt asynchronously while the log index - // is committed inline, so which of the two a slow write is in is not otherwise visible. Only the - // commit path writes, so one timer serves the store. + // Breaks a write into its stages, so which of them a slow write is in is visible. Only the writer + // goroutine records, so one timer serves the store. writePhases *seidbmetrics.PhaseTimer + + // Receipt writes waiting to be applied, and the meter reporting what waiting for room on this + // queue costs the caller. A whole write is queued — bodies, log index and version marker — so + // the queue's depth is the depth of the receipt write itself. Its capacity bounds how far the + // store may fall behind the chain; a nil channel means writes are applied on the caller. + writes chan receiptWrite + writeQueue *seidbmetrics.QueueMeter + writeErr atomic.Pointer[error] + stopSampling context.CancelFunc } +// receiptWrite is one block's receipts, waiting to be applied. +type receiptWrite struct { + height int64 + receipts []ReceiptRecord +} + +// writeQueueSampleIntervalSeconds is how often the write queue's depth is read. Sampling on a timer +// rather than at each send keeps the reading unbiased by the send rate. +const writeQueueSampleIntervalSeconds = 1 + var _ ReceiptStore = (*littReceiptStore)(nil) var ( @@ -213,7 +234,19 @@ func newLittReceiptStore(cfg dbconfig.ReceiptStoreConfig, storeKey sdk.StoreKey) return nil, fmt.Errorf("failed to open receipt log index: %w", err) } s.index = index - s.writePhases = seidbmetrics.NewPhaseTimer(otel.Meter("seidb_receipt"), "receipt_store_write") + + receiptMeter := otel.Meter("seidb_receipt") + s.writePhases = seidbmetrics.NewPhaseTimer(receiptMeter, "receipt_store_write") + if cfg.AsyncWriteBuffer > 0 { + s.writes = make(chan receiptWrite, cfg.AsyncWriteBuffer) + s.writeQueue = seidbmetrics.NewQueueMeter(receiptMeter, "receipt_write") + s.startWriter() + + samplingCtx, stopSampling := context.WithCancel(context.Background()) + s.stopSampling = stopSampling + s.writeQueue.SampleDepth(samplingCtx, writeQueueSampleIntervalSeconds, + func() int { return len(s.writes) }) + } s.latestVersion.Store(s.readMeta(receiptLatestVersionKey)) s.earliestVersion.Store(s.readMeta(receiptEarliestVersionKey)) @@ -301,17 +334,31 @@ func (s *littReceiptStore) belowRetentionFloor(blockNumber uint64) bool { return earliest > 0 && blockNumber < uint64(earliest) //nolint:gosec // earliest is non-negative } +// SetReceipts hands the block's receipts to the writer and returns without waiting for them to be +// applied, blocking only once the queue is full. It reports the failure of an earlier write, there +// being no other caller to report it to. With AsyncWriteBuffer off, the write is applied here. func (s *littReceiptStore) SetReceipts(ctx sdk.Context, receipts []ReceiptRecord) error { + if s.writes == nil { + return s.applyReceipts(ctx.BlockHeight(), receipts) + } + if err := s.takeWriteErr(); err != nil { + return err + } + seidbmetrics.Send(s.writeQueue, s.writes, receiptWrite{height: ctx.BlockHeight(), receipts: receipts}) + return nil +} + +// applyReceipts writes a block's receipt bodies, log index and version marker. The bodies go to +// litt first, so an indexed block always has its values written. +func (s *littReceiptStore) applyReceipts(height int64, receipts []ReceiptRecord) error { blockNumbers, receiptsByBlock := groupReceiptRecordsByBlock(receipts) if len(blockNumbers) == 0 { - return s.SetLatestVersion(ctx.BlockHeight()) + return s.SetLatestVersion(height) } // Closes the stage in flight, so the gap until the next write is charged to neither. defer s.writePhases.Reset() - // Receipt values go to litt first; the index batch (tag keys + version - // meta) commits after, so an indexed block always has its values written. batch := s.index.NewBatch() defer func() { _ = batch.Close() }() @@ -421,6 +468,50 @@ func (s *littReceiptStore) FilterLogs(ctx sdk.Context, fromBlock, toBlock uint64 return s.filterLogsByTags(reqCtx, fromBlock, toBlock, crit, budget) } +// startWriter applies queued receipt writes, in the order they were enqueued, until the store +// closes. +// +// It drains what is queued before returning, so a clean shutdown persists every write that was +// accepted. An unclean exit does not: up to writeQueueSize blocks of receipts are lost, the same +// direction litt's own write already takes, being flushed on a timer rather than at the call. +func (s *littReceiptStore) startWriter() { + s.backgroundWg.Add(1) + go func() { + defer s.backgroundWg.Done() + for { + select { + case write := <-s.writes: + s.applyWrite(write) + case <-s.stopBackground: + for { + select { + case write := <-s.writes: + s.applyWrite(write) + default: + return + } + } + } + } + }() +} + +// applyWrite performs one queued write, keeping the first failure for the next caller to collect. +func (s *littReceiptStore) applyWrite(write receiptWrite) { + if err := s.applyReceipts(write.height, write.receipts); err != nil { + logger.Error("failed to write receipts", "height", write.height, "err", err) + s.writeErr.CompareAndSwap(nil, &err) + } +} + +// takeWriteErr returns the first failure a queued write hit, clearing it so it is reported once. +func (s *littReceiptStore) takeWriteErr() error { + if err := s.writeErr.Swap(nil); err != nil { + return *err + } + return nil +} + // startFlusher bounds litt durability lag to littFlushInterval from a // background goroutine so block commit never waits on an fsync. func (s *littReceiptStore) startFlusher() { @@ -445,10 +536,17 @@ func (s *littReceiptStore) startFlusher() { func (s *littReceiptStore) Close() error { var err error s.closeOnce.Do(func() { + if s.stopSampling != nil { + s.stopSampling() + } close(s.stopBackground) + // The writer drains what it holds before returning, so this is where queued writes land. s.backgroundWg.Wait() + err = s.takeWriteErr() // litt's Close flushes, so the last sub-interval of writes is durable. - err = s.values.Close() + if valuesErr := s.values.Close(); err == nil { + err = valuesErr + } if indexErr := s.index.Close(); err == nil { err = indexErr } diff --git a/sei-db/ledger_db/receipt/littidx_test.go b/sei-db/ledger_db/receipt/littidx_test.go index d4f5fa23ad..1a4fa3132a 100644 --- a/sei-db/ledger_db/receipt/littidx_test.go +++ b/sei-db/ledger_db/receipt/littidx_test.go @@ -3,6 +3,7 @@ package receipt_test import ( "fmt" "testing" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/eth/filters" @@ -15,6 +16,62 @@ import ( "github.com/stretchr/testify/require" ) +// TestLittIdxSynchronousWriteBuffer pins the AsyncWriteBuffer <= 0 case: the write is applied on the +// caller, so the block is queryable the moment SetReceipts returns. +func TestLittIdxSynchronousWriteBuffer(t *testing.T) { + storeKey := storetypes.NewKVStoreKey("evm") + tkey := storetypes.NewTransientStoreKey("evm_transient") + ctx := testutil.DefaultContext(storeKey, tkey).WithBlockHeight(1) + cfg := dbconfig.DefaultReceiptStoreConfig() + cfg.Backend = "littidx" + cfg.DBDirectory = t.TempDir() + cfg.KeepRecent = 0 + cfg.AsyncWriteBuffer = 0 + + store, err := receipt.NewReceiptStore(cfg, storeKey) + require.NoError(t, err) + t.Cleanup(func() { _ = store.Close() }) + + addr := common.HexToAddress("0xabc") + record := litReceipt(1, 0, addr, common.HexToHash("0xdead")) + require.NoError(t, store.SetReceipts(ctx, []receipt.ReceiptRecord{record})) + + require.Equal(t, int64(1), store.LatestVersion()) + got, err := store.GetReceipt(ctx, record.TxHash) + require.NoError(t, err) + require.Equal(t, record.Receipt.TxHashHex, got.TxHashHex) +} + +// TestLittIdxWriteBufferBoundsLag pins that the buffer is the back-pressure point: with room for one +// block, a writer cannot get further than the buffer ahead of what has been applied. +func TestLittIdxWriteBufferBoundsLag(t *testing.T) { + storeKey := storetypes.NewKVStoreKey("evm") + tkey := storetypes.NewTransientStoreKey("evm_transient") + ctx := testutil.DefaultContext(storeKey, tkey).WithBlockHeight(1) + cfg := dbconfig.DefaultReceiptStoreConfig() + cfg.Backend = "littidx" + cfg.DBDirectory = t.TempDir() + cfg.KeepRecent = 0 + cfg.AsyncWriteBuffer = 1 + + store, err := receipt.NewReceiptStore(cfg, storeKey) + require.NoError(t, err) + t.Cleanup(func() { _ = store.Close() }) + + addr := common.HexToAddress("0xabc") + const blocks = 8 + for block := uint64(1); block <= blocks; block++ { + record := litReceipt(block, 0, addr, common.HexToHash("0xdead")) + require.NoError(t, store.SetReceipts(ctx.WithBlockHeight(int64(block)), //nolint:gosec // small test heights + []receipt.ReceiptRecord{record})) + // One queued block plus the one in flight is as far as the store may trail. + require.GreaterOrEqual(t, store.LatestVersion(), int64(block)-2) //nolint:gosec // small test heights + } + + require.Eventually(t, func() bool { return store.LatestVersion() == blocks }, + 5*time.Second, time.Millisecond) +} + func setupLittIdx(t *testing.T, dir string) (receipt.ReceiptStore, sdk.Context) { t.Helper() return setupLittIdxPar(t, dir, dbconfig.DefaultReceiptLogFilterParallelism) @@ -64,6 +121,20 @@ func litReceipt(block uint64, txIndex uint32, addr common.Address, topics ...com func writeLitBlock(t *testing.T, store receipt.ReceiptStore, ctx sdk.Context, block uint64, records ...receipt.ReceiptRecord) { t.Helper() require.NoError(t, store.SetReceipts(ctx.WithBlockHeight(int64(block)), records)) //nolint:gosec // small test heights + if len(records) == 0 { + return + } + // The write may be applied after SetReceipts returns, and neither signal alone marks the end of + // it: the bodies land before the version marker, and a block written in parts advances the + // marker on its first part. Wait for both. + last := records[len(records)-1].TxHash + require.Eventually(t, func() bool { + if store.LatestVersion() < int64(block) { //nolint:gosec // small test heights + return false + } + _, err := store.GetReceiptFromStore(ctx, last) + return err == nil + }, 5*time.Second, time.Millisecond) } func TestLittIdxReadWrite(t *testing.T) { diff --git a/sei-db/ledger_db/receipt/offline_internal_test.go b/sei-db/ledger_db/receipt/offline_internal_test.go index ac030667ca..2fd51092e9 100644 --- a/sei-db/ledger_db/receipt/offline_internal_test.go +++ b/sei-db/ledger_db/receipt/offline_internal_test.go @@ -36,6 +36,7 @@ func writeLittIdxReceipts(t *testing.T, dir string, blocks uint64) { []common.Hash{topic})} //nolint:gosec // small test heights require.NoError(t, store.SetReceipts(ctx.WithBlockHeight(int64(block)), []ReceiptRecord{record})) + requireReceiptVersion(t, store, int64(block)) //nolint:gosec // small test heights } require.NoError(t, store.Close()) } diff --git a/sei-db/ledger_db/receipt/receipt_bench_read_test.go b/sei-db/ledger_db/receipt/receipt_bench_read_test.go index 0605370167..4d9e002011 100644 --- a/sei-db/ledger_db/receipt/receipt_bench_read_test.go +++ b/sei-db/ledger_db/receipt/receipt_bench_read_test.go @@ -6,6 +6,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/ethereum/go-ethereum/common" ethtypes "github.com/ethereum/go-ethereum/core/types" @@ -271,6 +272,10 @@ func setupReadBenchmark(b *testing.B, backend string, blocks, receiptsPerBlock, if err := store.SetReceipts(ctx.WithBlockHeight(int64(blockNumber)), batch); err != nil { b.Fatalf("failed to write block %d: %v", blockNumber, err) } + // Seeding outruns the writer, so wait for the block to be published before the next one. + for store.LatestVersion() < int64(blockNumber) { //nolint:gosec // small test heights + time.Sleep(time.Millisecond) + } seed += uint64(receiptsPerBlock) if (block+1)%logInterval == 0 { diff --git a/sei-db/ledger_db/receipt/receipt_store.go b/sei-db/ledger_db/receipt/receipt_store.go index eea5219070..d8e61d5a68 100644 --- a/sei-db/ledger_db/receipt/receipt_store.go +++ b/sei-db/ledger_db/receipt/receipt_store.go @@ -51,12 +51,15 @@ func NewTooManyLogBytesError(maxBytes int64) error { type ReceiptStore interface { controller.PrunableStore + // LatestVersion is the highest block whose receipts are queryable. A write may be applied + // after SetReceipts returns, so this is the watermark a reader follows rather than the + // height it last wrote. LatestVersion() int64 EarliestVersion() int64 - SetLatestVersion(version int64) error - SetEarliestVersion(version int64) error GetReceipt(ctx sdk.Context, txHash common.Hash) (*types.Receipt, error) GetReceiptFromStore(ctx sdk.Context, txHash common.Hash) (*types.Receipt, error) + // SetReceipts writes the block's receipts, carrying the version markers with them. An + // implementation may apply the write in the background; LatestVersion reports when it lands. SetReceipts(ctx sdk.Context, receipts []ReceiptRecord) error // FilterLogs queries logs across a range of blocks. // For single-block queries, set fromBlock == toBlock. diff --git a/sei-db/ledger_db/receipt/receipt_store_test.go b/sei-db/ledger_db/receipt/receipt_store_test.go index 28461dbed8..95dd4db79b 100644 --- a/sei-db/ledger_db/receipt/receipt_store_test.go +++ b/sei-db/ledger_db/receipt/receipt_store_test.go @@ -93,6 +93,7 @@ func TestSetReceiptsAndGet(t *testing.T) { {TxHash: txHash}, }) require.NoError(t, err) + require.Eventually(t, func() bool { return store.LatestVersion() >= 1 }, 5*time.Second, time.Millisecond) got, err := store.GetReceipt(ctx, txHash) require.NoError(t, err) @@ -106,9 +107,16 @@ func TestSetReceiptsAndGet(t *testing.T) { require.Error(t, err) require.GreaterOrEqual(t, store.LatestVersion(), int64(1)) - require.NoError(t, store.SetLatestVersion(10)) + + // The version markers ride SetReceipts, so they are off the store's interface. + pinner, ok := store.(interface { + SetLatestVersion(version int64) error + SetEarliestVersion(version int64) error + }) + require.True(t, ok) + require.NoError(t, pinner.SetLatestVersion(10)) require.Equal(t, int64(10), store.LatestVersion()) - require.NoError(t, store.SetEarliestVersion(1)) + require.NoError(t, pinner.SetEarliestVersion(1)) require.Equal(t, int64(1), store.EarliestVersion()) } diff --git a/sei-db/ledger_db/receipt/test_helpers_test.go b/sei-db/ledger_db/receipt/test_helpers_test.go index bbb8df60c8..09c1c89e18 100644 --- a/sei-db/ledger_db/receipt/test_helpers_test.go +++ b/sei-db/ledger_db/receipt/test_helpers_test.go @@ -1,13 +1,25 @@ package receipt import ( + "testing" + "time" + "github.com/ethereum/go-ethereum/common" storetypes "github.com/sei-protocol/sei-chain/sei-cosmos/store/types" "github.com/sei-protocol/sei-chain/sei-cosmos/testutil" sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" "github.com/sei-protocol/sei-chain/x/evm/types" + "github.com/stretchr/testify/require" ) +// requireReceiptVersion waits for the store to publish height. A write may be applied after +// SetReceipts returns, and LatestVersion is how a reader learns that it landed. +func requireReceiptVersion(t *testing.T, store ReceiptStore, height int64) { + t.Helper() + require.Eventually(t, func() bool { return store.LatestVersion() >= height }, + 5*time.Second, time.Millisecond) +} + func newTestContext() (sdk.Context, storetypes.StoreKey) { storeKey := storetypes.NewKVStoreKey("evm") tkey := storetypes.NewTransientStoreKey("evm_transient") From d07bc7edb5c8ea23d8ca5a1c9e7cd8d2be0e1c4b Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 14:22:16 -0700 Subject: [PATCH 02/11] Disable internal wal for SS --- sei-db/config/ss_config.go | 10 +++++ sei-db/db_engine/pebbledb/mvcc/db.go | 47 ++++++++++++-------- sei-db/state_db/giga/state_db.go | 10 ++++- sei-db/state_db/giga/state_db_replay_test.go | 20 +++++++++ 4 files changed, 67 insertions(+), 20 deletions(-) diff --git a/sei-db/config/ss_config.go b/sei-db/config/ss_config.go index 0e371816f8..cde25c190f 100644 --- a/sei-db/config/ss_config.go +++ b/sei-db/config/ss_config.go @@ -38,6 +38,16 @@ type StateStoreConfig struct { // defaults to 100 AsyncWriteBuffer int `mapstructure:"async-write-buffer"` + // DisableInternalWAL stops the backend from keeping a changelog WAL of its own, so a commit is not + // held up by a log write. It is for an owner that already logs every block and replays that log + // into this store: giga's StateDB writes its state WAL before the store and catches the store up + // from it on open, which makes a second log here written and never read. + // + // Like KeepRecent this is not read from the state-store config, since it is only correct when the + // owner supplies the log. Rollback through ss/composite replays the changelog and has no other + // source for the versions above a snapshot, so that path must leave it on. + DisableInternalWAL bool `mapstructure:"-"` + // KeepRecent defines the number of versions to keep in state store (shared by Cosmos and EVM). // Setting it to 0 means keep everything. // Default to keep the last 100,000 blocks diff --git a/sei-db/db_engine/pebbledb/mvcc/db.go b/sei-db/db_engine/pebbledb/mvcc/db.go index 7e079dc465..4793f6d305 100644 --- a/sei-db/db_engine/pebbledb/mvcc/db.go +++ b/sei-db/db_engine/pebbledb/mvcc/db.go @@ -103,6 +103,9 @@ type Database struct { // Pending changes to be written to the DB pendingChanges chan VersionedChangesets + // Guards the one close of pendingChanges, so Close stays idempotent. + drainOnce sync.Once + // Reports pendingChanges from the writer's side: how full it was when a write needed room, and how // long writes waited when it had none. pendingChangesQueue *seidbmetrics.QueueMeter @@ -239,23 +242,27 @@ func OpenDB(dataDir string, config config.StateStoreConfig) (types.StateStore, e _ = db.Close() return nil, errors.New("KeepRecent must be non-negative") } - walKeepRecent := changelogKeepRecent(config) - // Snapshot rollback replays the changelog forward from the oldest retained - // snapshot, so count-based pruning must not cut inside that span. The - // snapshot manager prunes this changelog by snapshot version after every - // retention pass and is what actually holds it down; the count below is the - // ceiling for the states that pass does not cover — external snapshot - // pruning, and the stretch before enough snapshots exist to prune. Raising - // the ceiling is what a rollback window costs on disk: roughly one snapshot - // interval of changelog per retained snapshot. - streamHandler, err := wal.NewChangelogWAL(utils.GetChangelogPath(dataDir), wal.Config{ - KeepRecent: walKeepRecent, - PruneInterval: time.Duration(config.PruneIntervalSeconds) * time.Second, - }) - if err != nil { - return nil, err + // An owner that logs every block itself replays that log into this store, so the changelog here + // would be written and never read; DisableInternalWAL drops it and the commit-path write with it. + if !config.DisableInternalWAL { + walKeepRecent := changelogKeepRecent(config) + // Snapshot rollback replays the changelog forward from the oldest retained + // snapshot, so count-based pruning must not cut inside that span. The + // snapshot manager prunes this changelog by snapshot version after every + // retention pass and is what actually holds it down; the count below is the + // ceiling for the states that pass does not cover — external snapshot + // pruning, and the stretch before enough snapshots exist to prune. Raising + // the ceiling is what a rollback window costs on disk: roughly one snapshot + // interval of changelog per retained snapshot. + streamHandler, err := wal.NewChangelogWAL(utils.GetChangelogPath(dataDir), wal.Config{ + KeepRecent: walKeepRecent, + PruneInterval: time.Duration(config.PruneIntervalSeconds) * time.Second, + }) + if err != nil { + return nil, err + } + database.streamHandler = streamHandler } - database.streamHandler = streamHandler database.asyncWriteWG.Add(1) go database.writeAsyncInBackground() @@ -395,12 +402,16 @@ func (db *Database) Close() error { db.metricsCancel() } - if db.streamHandler != nil { + // Draining is owed whether or not a changelog is kept: the queued blocks are only in memory, and + // with no changelog there is nothing to replay them from either. The channel is left in place so + // that a send after close still panics rather than blocking forever on a nil one. + db.drainOnce.Do(func() { // First, stop accepting new pending changes and drain the worker close(db.pendingChanges) // Wait for the async writes to finish db.asyncWriteWG.Wait() - // Now close the WAL stream + }) + if db.streamHandler != nil { _ = db.streamHandler.Close() db.streamHandler = nil } diff --git a/sei-db/state_db/giga/state_db.go b/sei-db/state_db/giga/state_db.go index 8d7c224e35..56083ef5be 100644 --- a/sei-db/state_db/giga/state_db.go +++ b/sei-db/state_db/giga/state_db.go @@ -192,12 +192,18 @@ func (s *StateDB) openSS() error { if !s.ssCfg.Enable { return nil } - ss, err := evm.NewEVMStateStore(s.ssCfg.EVMDBDirectory, s.ssCfg) + // The state WAL this StateDB writes before every commit is what catchUpTo replays into SS, and + // rollback rewinds SS from its snapshots against that same WAL. A changelog inside SS would be + // a second log of every block that nothing here reads, paid for on the commit path. + ssCfg := s.ssCfg + ssCfg.DisableInternalWAL = true + + ss, err := evm.NewEVMStateStore(ssCfg.EVMDBDirectory, ssCfg) if err != nil { return fmt.Errorf("open EVM state store: %w", err) } s.ss = ss - if err := s.ss.StartSnapshots(s.ssSnapshotRoot(), s.ssCfg, nil); err != nil { + if err := s.ss.StartSnapshots(s.ssSnapshotRoot(), ssCfg, nil); err != nil { return fmt.Errorf("start EVM state store snapshot manager: %w", err) } return nil diff --git a/sei-db/state_db/giga/state_db_replay_test.go b/sei-db/state_db/giga/state_db_replay_test.go index f1861346c9..8946ade3da 100644 --- a/sei-db/state_db/giga/state_db_replay_test.go +++ b/sei-db/state_db/giga/state_db_replay_test.go @@ -13,6 +13,26 @@ import ( "github.com/stretchr/testify/require" ) +// SS keeps no changelog of its own under giga: the state WAL written before every commit is what +// catchUpTo replays into it, so a second log would be written on the commit path and never read. +// Recovery rests on that, which is why the absence is pinned rather than left to the config. +func TestOpenSSKeepsNoChangelogOfItsOwn(t *testing.T) { + s := &StateDB{ + flatkvCfg: flatkvconfig.DefaultTestConfig(t), + ssCfg: config.DefaultStateStoreConfig(), + } + s.ssCfg.Enable = true + s.ssCfg.EVMDBDirectory = filepath.Join(t.TempDir(), "ss") + + require.NoError(t, s.openSS()) + t.Cleanup(func() { _ = s.ss.Close() }) + + changelog := utils.GetChangelogPath(s.ssCfg.EVMDBDirectory) + _, err := os.Stat(changelog) + require.True(t, os.IsNotExist(err), + "SS must keep no changelog under giga; found one at %s", changelog) +} + // A node that keeps no EVM state store never reaches it, so nothing probes a store it does not have. // The directory is one an earlier run with SS on could have left, and the WAL reaches block 1, so a // rollback that read it would come back with a rewind to run. From 0e84e10706c59c500a5b93b82319e3014195be9d Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 14:25:28 -0700 Subject: [PATCH 03/11] Disable internal wal for GigaStorageManager --- sei-db/config/giga_config.go | 1 + 1 file changed, 1 insertion(+) diff --git a/sei-db/config/giga_config.go b/sei-db/config/giga_config.go index dca0f6d107..b97a6bc3bf 100644 --- a/sei-db/config/giga_config.go +++ b/sei-db/config/giga_config.go @@ -44,6 +44,7 @@ func DefaultGigaStorageConfig(homePath string) (*GigaStorageConfig, error) { ssConfig := DefaultStateStoreConfig() ssConfig.EVMDBDirectory = utils.GetEVMStateStorePath(homePath, ssConfig.Backend) ssConfig.ExternalPruning = true + ssConfig.DisableInternalWAL = true receiptConfig := DefaultReceiptStoreConfig() receiptConfig.Backend = gigaReceiptBackend From b665177fb1c9b83527753d470f7615ddf5bca860 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 15:03:35 -0700 Subject: [PATCH 04/11] Address comments --- .../dashboards/gigasim-dashboard.json | 125 ++++++++++++++ evmrpc/setup_test.go | 16 +- evmrpc/tests/utils.go | 16 +- sei-db/bootstrap/recovery_test.go | 4 +- sei-db/config/giga_config.go | 1 + sei-db/config/receipt_config.go | 21 ++- .../ledger_db/receipt/litt_receipt_store.go | 79 +++++++-- .../litt_write_failure_internal_test.go | 157 ++++++++++++++++++ sei-db/ledger_db/receipt/receipt_store.go | 21 +++ .../ledger_db/receipt/receipt_store_test.go | 8 +- 10 files changed, 394 insertions(+), 54 deletions(-) create mode 100644 sei-db/ledger_db/receipt/litt_write_failure_internal_test.go diff --git a/docker/monitornode/dashboards/gigasim-dashboard.json b/docker/monitornode/dashboards/gigasim-dashboard.json index 57817bfffa..ad7d5ece2c 100644 --- a/docker/monitornode/dashboards/gigasim-dashboard.json +++ b/docker/monitornode/dashboards/gigasim-dashboard.json @@ -10796,6 +10796,118 @@ "version": "13.2.1" } } + }, + "panel-134": { + "kind": "Panel", + "spec": { + "data": { + "kind": "QueryGroup", + "spec": { + "queries": [ + { + "kind": "PanelQuery", + "spec": { + "hidden": false, + "query": { + "datasource": { + "name": "PBFA97CFB590B2093" + }, + "group": "prometheus", + "kind": "DataQuery", + "spec": { + "editorMode": "code", + "expr": "sum(rate(pebble_pending_changes_queue_blocked_seconds_total{db=~\".*state_store.*\"}[$__rate_interval])) / sum(rate(gigasim_blocks_processed_total[$__rate_interval]))", + "legendFormat": "blocked per block", + "range": true + }, + "version": "v0" + }, + "refId": "A" + } + } + ], + "queryOptions": {}, + "transformations": [] + } + }, + "description": "Seconds the commit path spent waiting for room on the EVM state store's apply queue, per block. Depth is sampled and can miss a queue that fills and drains between samples; this counter integrates every wait. When it accounts for most of enqueue_ss, the store is applying slower than blocks arrive, so the commit path is waiting on the queue rather than on work of its own — and no further pipelining helps, because the bottleneck is downstream of it.", + "id": 134, + "links": [], + "title": "SS Commit Queue — Blocked Time per Block", + "vizConfig": { + "group": "timeseries", + "kind": "VizConfig", + "spec": { + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 15, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "showValues": false, + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "min": 0, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + } + }, + "version": "13.2.1" + } + } } }, "layout": { @@ -11188,6 +11300,19 @@ "x": 16, "y": 8 } + }, + { + "kind": "GridLayoutItem", + "spec": { + "element": { + "kind": "ElementReference", + "name": "panel-134" + }, + "height": 8, + "width": 8, + "x": 0, + "y": 16 + } } ] } diff --git a/evmrpc/setup_test.go b/evmrpc/setup_test.go index 57c4ad89d8..bc7a4261d6 100644 --- a/evmrpc/setup_test.go +++ b/evmrpc/setup_test.go @@ -63,24 +63,10 @@ const MockHeight103 = 103 const MockHeight101 = 101 const MockHeight100 = 100 -// receiptVersionPinner is implemented by receipt stores whose version markers can be written -// directly. SetReceipts carries those markers, so the store's interface does not expose them. -type receiptVersionPinner interface { - SetLatestVersion(version int64) error - SetEarliestVersion(version int64) error -} - // pinReceiptVersions widens a store's queryable window to [1, latest]. These tests seed receipts // by other means, so nothing has advanced the markers a read is gated on. func pinReceiptVersions(store receipt.ReceiptStore, latest int64) error { - pinner, ok := store.(receiptVersionPinner) - if !ok { - return fmt.Errorf("receipt store %T cannot pin versions", store) - } - if err := pinner.SetLatestVersion(latest); err != nil { - return err - } - return pinner.SetEarliestVersion(1) + return receipt.PinVersions(store, 1, latest) } // LatestCtxUpgradeName makes the test ctx look like a real chain that has diff --git a/evmrpc/tests/utils.go b/evmrpc/tests/utils.go index fb8ced197d..54592ac5d9 100644 --- a/evmrpc/tests/utils.go +++ b/evmrpc/tests/utils.go @@ -20,6 +20,7 @@ import ( evmrpcconfig "github.com/sei-protocol/sei-chain/evmrpc/config" "github.com/sei-protocol/sei-chain/sei-cosmos/client" sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" tmproto "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/types" testkeeper "github.com/sei-protocol/sei-chain/testutil/keeper" @@ -187,19 +188,8 @@ func setupTestServer( } pinStateStoreLatestVersion(a, ctxProvider) if store := a.EvmKeeper.ReceiptStore(); store != nil { - // SetReceipts carries the version markers, so they are off the store's interface. These - // tests seed receipts by other means and would otherwise read against an unset window. - pinner, ok := store.(interface { - SetLatestVersion(version int64) error - SetEarliestVersion(version int64) error - }) - if !ok { - panic(fmt.Sprintf("receipt store %T cannot pin versions", store)) - } - if err := pinner.SetLatestVersion(math.MaxInt64); err != nil { - panic(err) - } - if err := pinner.SetEarliestVersion(1); err != nil { + // These tests seed receipts by other means and would otherwise read against an unset window. + if err := receipt.PinVersions(store, 1, math.MaxInt64); err != nil { panic(err) } } diff --git a/sei-db/bootstrap/recovery_test.go b/sei-db/bootstrap/recovery_test.go index fa652c6718..293b4b92d8 100644 --- a/sei-db/bootstrap/recovery_test.go +++ b/sei-db/bootstrap/recovery_test.go @@ -237,9 +237,7 @@ func TestFindTargetRecoveryHeightIsZeroWithoutABlockLedger(t *testing.T) { commitBlocks(t, manager, 3) // The version marker rides SetReceipts, so it is off the store's interface; this test stamps a // head without bodies on purpose. - pinner, ok := manager.ReceiptDB().(interface { - SetLatestVersion(version int64) error - }) + pinner, ok := manager.ReceiptDB().(receipt.VersionPinner) require.True(t, ok) require.NoError(t, pinner.SetLatestVersion(3)) // findTargetRecoveryHeight reads the state and receipt directories offline, so both stores have diff --git a/sei-db/config/giga_config.go b/sei-db/config/giga_config.go index b97a6bc3bf..478361f014 100644 --- a/sei-db/config/giga_config.go +++ b/sei-db/config/giga_config.go @@ -45,6 +45,7 @@ func DefaultGigaStorageConfig(homePath string) (*GigaStorageConfig, error) { ssConfig.EVMDBDirectory = utils.GetEVMStateStorePath(homePath, ssConfig.Backend) ssConfig.ExternalPruning = true ssConfig.DisableInternalWAL = true + ssConfig.SeparateEVMSubDBs = true receiptConfig := DefaultReceiptStoreConfig() receiptConfig.Backend = gigaReceiptBackend diff --git a/sei-db/config/receipt_config.go b/sei-db/config/receipt_config.go index 38cfa63ea2..c9de072f2c 100644 --- a/sei-db/config/receipt_config.go +++ b/sei-db/config/receipt_config.go @@ -28,6 +28,12 @@ const ( // littidx eth_getLogs (see ReceiptStoreConfig.LogFilterParallelism). const DefaultReceiptLogFilterParallelism = 16 +// DefaultReceiptAsyncWriteBuffer is the default queue depth for receipt writes +// (see ReceiptStoreConfig.AsyncWriteBuffer). It is small because the queue's +// depth is how far an unclean exit sets recovery back, not only how much burst +// the writer can absorb. +const DefaultReceiptAsyncWriteBuffer = 10 + // ReceiptStoreConfig defines configuration for the receipt store database. type ReceiptStoreConfig struct { // Enable reports whether the receipt store is opened. A node with it off keeps no receipt @@ -49,8 +55,19 @@ type ReceiptStoreConfig struct { // AsyncWriteBuffer defines the async queue length for commits to be applied to receipt store. // It bounds how many blocks the store may fall behind the chain before a write blocks, and so // how far LatestVersion may trail the height just written. + // + // Raising it costs more than the memory it holds. The queue is not on disk, so an unclean exit + // loses it, and recovery converges every store on the lowest head: a receipt store that comes + // back this many blocks behind rolls the state DB and block store back with it, and that + // rollback refuses outright if the state snapshots and WAL cannot span the distance. Size it + // for the burst the writer must absorb, not larger. + // + // It also bounds how stale the EVM RPC head can be. The watermark those queries are served + // against takes the lowest height every store can answer for, this one included, so a receipt + // store behind by a queue's depth holds eth_blockNumber and "latest" that far back. + // // Set <= 0 for synchronous writes. - // defaults to 100 + // defaults to 10 AsyncWriteBuffer int `mapstructure:"async-write-buffer"` // KeepRecent defines the number of versions to keep in receipt store. @@ -99,7 +116,7 @@ func DefaultReceiptStoreConfig() ReceiptStoreConfig { return ReceiptStoreConfig{ Enable: true, Backend: "pebbledb", - AsyncWriteBuffer: DefaultSSAsyncBuffer, + AsyncWriteBuffer: DefaultReceiptAsyncWriteBuffer, KeepRecent: 0, PruneIntervalSeconds: DefaultSSPruneInterval, LogFilterParallelism: DefaultReceiptLogFilterParallelism, diff --git a/sei-db/ledger_db/receipt/litt_receipt_store.go b/sei-db/ledger_db/receipt/litt_receipt_store.go index c82642b453..779917e6b1 100644 --- a/sei-db/ledger_db/receipt/litt_receipt_store.go +++ b/sei-db/ledger_db/receipt/litt_receipt_store.go @@ -64,7 +64,8 @@ import ( // implementation in litt_receipt_gc.go, and startPruning stands down. // // Writes are applied in the background: SetReceipts queues a block and returns, so a receipt is not -// necessarily readable the moment it returns. Sync waits for the queue to empty. +// necessarily readable the moment it returns. LatestVersion is the watermark that says how far the +// applied writes have reached, and Close is what waits for the queue to empty. type littReceiptStore struct { values litt.DB receipts litt.Table @@ -335,17 +336,55 @@ func (s *littReceiptStore) belowRetentionFloor(blockNumber uint64) bool { } // SetReceipts hands the block's receipts to the writer and returns without waiting for them to be -// applied, blocking only once the queue is full. It reports the failure of an earlier write, there -// being no other caller to report it to. With AsyncWriteBuffer off, the write is applied here. +// applied, blocking only once the queue is full. With AsyncWriteBuffer off, the write is applied +// here instead. +// +// It refuses once a queued write has failed, reporting that failure rather than taking the block: +// the store applies nothing after a failure, so accepting one would drop it silently. func (s *littReceiptStore) SetReceipts(ctx sdk.Context, receipts []ReceiptRecord) error { if s.writes == nil { return s.applyReceipts(ctx.BlockHeight(), receipts) } - if err := s.takeWriteErr(); err != nil { + if err := s.writeFailure(); err != nil { return err } - seidbmetrics.Send(s.writeQueue, s.writes, receiptWrite{height: ctx.BlockHeight(), receipts: receipts}) - return nil + return s.queueWrite(receiptWrite{height: ctx.BlockHeight(), receipts: receipts}) +} + +// ErrStoreClosed is returned by a write the store can no longer apply, the writer having stopped. +var ErrStoreClosed = errors.New("receipt store is closed") + +// queueWrite hands a write to the writer, waiting for room when the queue is full. +// +// A closed store is refused rather than accepted. The writer drains and exits during Close, so a +// send after that point would sit in a channel nobody reads, and a send once the queue is full +// would never return — inside a commit, which hangs the node instead of failing it. +func (s *littReceiptStore) queueWrite(write receiptWrite) error { + // Checked before the send rather than only alongside it: select picks uniformly among ready + // cases, so a send with room available would win half the races against an already-closed store. + select { + case <-s.stopBackground: + return ErrStoreClosed + default: + } + return s.writeQueue.SendVia( + func() bool { + select { + case s.writes <- write: + return true + default: + return false + } + }, + func() error { + select { + case s.writes <- write: + return nil + case <-s.stopBackground: + return ErrStoreClosed + } + }, + ) } // applyReceipts writes a block's receipt bodies, log index and version marker. The bodies go to @@ -471,9 +510,11 @@ func (s *littReceiptStore) FilterLogs(ctx sdk.Context, fromBlock, toBlock uint64 // startWriter applies queued receipt writes, in the order they were enqueued, until the store // closes. // -// It drains what is queued before returning, so a clean shutdown persists every write that was -// accepted. An unclean exit does not: up to writeQueueSize blocks of receipts are lost, the same -// direction litt's own write already takes, being flushed on a timer rather than at the call. +// It drains what is queued before returning, so a clean shutdown applies the writes it holds. An +// unclean exit does not: everything queued is lost, up to the AsyncWriteBuffer blocks the queue +// holds, and that includes each block's log index and version marker rather than only its bodies. +// The store therefore comes back that far behind, which recovery resolves by rolling every other +// store down to it, so the buffer's size is a recovery cost and not only a memory one. func (s *littReceiptStore) startWriter() { s.backgroundWg.Add(1) go func() { @@ -496,17 +537,27 @@ func (s *littReceiptStore) startWriter() { }() } -// applyWrite performs one queued write, keeping the first failure for the next caller to collect. +// applyWrite performs one queued write, keeping the first failure for its callers to collect. +// +// Nothing is applied after a failure. A later block would carry its own version marker, publishing a +// head above one whose receipts were never written: reads of the missing block would report no logs +// and no such transaction, and recovery converges on the published head, so the gap never refills. +// The queue is drained rather than left to fill, so a writer blocked on it is released to see the +// error instead of waiting on a consumer that will never take its block. func (s *littReceiptStore) applyWrite(write receiptWrite) { + if s.writeFailure() != nil { + return + } if err := s.applyReceipts(write.height, write.receipts); err != nil { logger.Error("failed to write receipts", "height", write.height, "err", err) s.writeErr.CompareAndSwap(nil, &err) } } -// takeWriteErr returns the first failure a queued write hit, clearing it so it is reported once. -func (s *littReceiptStore) takeWriteErr() error { - if err := s.writeErr.Swap(nil); err != nil { +// writeFailure returns the first failure a queued write hit. It latches rather than clearing, so +// every later SetReceipts and Close reports it and no single caller can consume it from the rest. +func (s *littReceiptStore) writeFailure() error { + if err := s.writeErr.Load(); err != nil { return *err } return nil @@ -542,7 +593,7 @@ func (s *littReceiptStore) Close() error { close(s.stopBackground) // The writer drains what it holds before returning, so this is where queued writes land. s.backgroundWg.Wait() - err = s.takeWriteErr() + err = s.writeFailure() // litt's Close flushes, so the last sub-interval of writes is durable. if valuesErr := s.values.Close(); err == nil { err = valuesErr diff --git a/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go b/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go new file mode 100644 index 0000000000..4c178ba436 --- /dev/null +++ b/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go @@ -0,0 +1,157 @@ +package receipt + +import ( + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + dbtypes "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" + "github.com/stretchr/testify/require" +) + +var errIndexCommit = errors.New("injected index commit failure") + +// failingIndex is the store's log index with its batch commits made to fail on demand. Committing +// the index is the one step of a receipt write with no other way to fail in a test, and holding the +// commit is what lets a test queue a block behind the one that is failing. +type failingIndex struct { + dbtypes.KeyValueDB + failing atomic.Bool + entered chan struct{} // closed once a failing commit has been reached + enteredOnce sync.Once // more than one commit may fail, and entered closes for the first + release chan struct{} // closed to let that commit return its error +} + +func (f *failingIndex) NewBatch() dbtypes.Batch { + return &failingBatch{Batch: f.KeyValueDB.NewBatch(), index: f} +} + +type failingBatch struct { + dbtypes.Batch + index *failingIndex +} + +func (b *failingBatch) Commit(opts dbtypes.WriteOptions) error { + if b.index.failing.Load() { + b.index.enteredOnce.Do(func() { close(b.index.entered) }) + <-b.index.release + return errIndexCommit + } + return b.Batch.Commit(opts) +} + +// TestWriteFailureHoldsTheHeadAgainstAQueuedBlock covers what a failed background write owes the +// blocks already queued behind it. Applying one would commit its own version marker and publish a +// head above the block that never landed: reads of the missing block report no logs and no such +// transaction, and recovery converges on the published head, so the gap never refills. +// +// The block behind the failure is queued while the failing commit is held, which is the ordering +// that makes this reachable — SetReceipts refuses new blocks once the failure is visible. +func TestWriteFailureHoldsTheHeadAgainstAQueuedBlock(t *testing.T) { + s, closeStore := setupLittCtxStore(t) + defer closeStore() + + addr := common.HexToAddress("0xfa11") + topic := common.HexToHash("0xfa12") + + index := &failingIndex{ + KeyValueDB: s.index, + entered: make(chan struct{}), + release: make(chan struct{}), + } + s.index = index + + // Block 1 lands, so there is a real head for the failure to hold. + writeOneReceipt(t, s, 1, addr, topic) + requireReceiptVersion(t, s, 1) + + // Block 2 reaches its commit and stops there, still holding the writer. + index.failing.Store(true) + writeOneReceipt(t, s, 2, addr, topic) + <-index.entered + + // Block 3 would commit cleanly and carry a marker naming it the head. Queued now, while block 2 + // is mid-commit, it is past the refusal in SetReceipts and only the writer can hold it back. + index.failing.Store(false) + writeOneReceipt(t, s, 3, addr, topic) + + close(index.release) + + // Close drains, so the writer has decided about block 3 by the time this returns. + require.ErrorIs(t, s.Close(), errIndexCommit) + require.Equal(t, int64(1), s.LatestVersion(), + "the head must not move past a block whose receipts were never written") +} + +// TestWriteFailureLatches covers the failure reaching every later caller rather than the first one +// to ask, which is what lets both a commit and Close act on it. +func TestWriteFailureLatches(t *testing.T) { + s, closeStore := setupLittCtxStore(t) + defer closeStore() + + addr := common.HexToAddress("0xfa21") + topic := common.HexToHash("0xfa22") + + index := &failingIndex{ + KeyValueDB: s.index, + entered: make(chan struct{}), + release: make(chan struct{}), + } + s.index = index + close(index.release) + + index.failing.Store(true) + writeOneReceipt(t, s, 1, addr, topic) + require.Eventually(t, func() bool { return s.writeFailure() != nil }, 5*time.Second, time.Millisecond) + + txHash, rcpt := littCtxTestReceipt(2, 0, addr, topic, 1) + require.ErrorIs(t, s.SetReceipts(newTestCtxAtHeight(2), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}}), + errIndexCommit, "a commit after a failed write must be refused rather than queued") + require.ErrorIs(t, s.writeFailure(), errIndexCommit, "reading the failure must not consume it") + require.ErrorIs(t, s.Close(), errIndexCommit, "Close must report it too") +} + +// TestWriteAfterCloseIsRefused covers a commit that races shutdown. The writer has drained and gone +// by then, so a write it accepted would sit in a channel nobody reads, and one arriving on a full +// queue would never return — inside a commit, which hangs the node rather than failing it. +func TestWriteAfterCloseIsRefused(t *testing.T) { + s, _ := setupLittCtxStore(t) + require.NoError(t, s.Close()) + + txHash, rcpt := littCtxTestReceipt(1, 0, common.HexToAddress("0xfa31"), common.HexToHash("0xfa32"), 1) + err := s.SetReceipts(newTestCtxAtHeight(1), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}}) + require.ErrorIs(t, err, ErrStoreClosed) +} + +// TestWriteAfterCloseIsRefusedWithAFullQueue is the same refusal with no room left to send into, +// which is the case that would otherwise block forever rather than return. +func TestWriteAfterCloseIsRefusedWithAFullQueue(t *testing.T) { + s, _ := setupLittCtxStore(t) + require.NoError(t, s.Close()) + + // Leftovers with no writer behind them: the send has nowhere to go and nobody to take it. + for len(s.writes) < cap(s.writes) { + s.writes <- receiptWrite{height: 1} + } + + txHash, rcpt := littCtxTestReceipt(1, 0, common.HexToAddress("0xfa41"), common.HexToHash("0xfa42"), 1) + done := make(chan error, 1) + go func() { + done <- s.SetReceipts(newTestCtxAtHeight(1), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}}) + }() + select { + case err := <-done: + require.ErrorIs(t, err, ErrStoreClosed) + case <-time.After(5 * time.Second): + t.Fatal("a write into a full queue on a closed store never returned") + } +} + +func writeOneReceipt(t *testing.T, s *littReceiptStore, block uint64, addr common.Address, topic common.Hash) { + t.Helper() + txHash, rcpt := littCtxTestReceipt(block, 0, addr, topic, 1) + require.NoError(t, s.SetReceipts(newTestCtxAtHeight(block), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) +} diff --git a/sei-db/ledger_db/receipt/receipt_store.go b/sei-db/ledger_db/receipt/receipt_store.go index d8e61d5a68..c9f9b3f6d7 100644 --- a/sei-db/ledger_db/receipt/receipt_store.go +++ b/sei-db/ledger_db/receipt/receipt_store.go @@ -71,6 +71,27 @@ type ReceiptStore interface { Close() error } +// VersionPinner is implemented by receipt stores whose version markers can be written directly. +// SetReceipts carries those markers, so they are not on ReceiptStore; this is for a caller that has +// put receipts in place by other means and has to state the window they cover. +type VersionPinner interface { + SetLatestVersion(version int64) error + SetEarliestVersion(version int64) error +} + +// PinVersions widens store's queryable window to [earliest, latest]. It reports a store that does +// not support being pinned rather than leaving the window silently unset. +func PinVersions(store ReceiptStore, earliest, latest int64) error { + pinner, ok := store.(VersionPinner) + if !ok { + return fmt.Errorf("receipt store %T cannot pin versions", store) + } + if err := pinner.SetLatestVersion(latest); err != nil { + return err + } + return pinner.SetEarliestVersion(earliest) +} + type ReceiptRecord struct { TxHash common.Hash Receipt *types.Receipt diff --git a/sei-db/ledger_db/receipt/receipt_store_test.go b/sei-db/ledger_db/receipt/receipt_store_test.go index 95dd4db79b..7d51966d53 100644 --- a/sei-db/ledger_db/receipt/receipt_store_test.go +++ b/sei-db/ledger_db/receipt/receipt_store_test.go @@ -109,14 +109,8 @@ func TestSetReceiptsAndGet(t *testing.T) { require.GreaterOrEqual(t, store.LatestVersion(), int64(1)) // The version markers ride SetReceipts, so they are off the store's interface. - pinner, ok := store.(interface { - SetLatestVersion(version int64) error - SetEarliestVersion(version int64) error - }) - require.True(t, ok) - require.NoError(t, pinner.SetLatestVersion(10)) + require.NoError(t, receipt.PinVersions(store, 1, 10)) require.Equal(t, int64(10), store.LatestVersion()) - require.NoError(t, pinner.SetEarliestVersion(1)) require.Equal(t, int64(1), store.EarliestVersion()) } From 72e3345c46ef5922638f1a34415016f999092893 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 15:12:41 -0700 Subject: [PATCH 05/11] Tune pebbledb configs --- sei-db/db_engine/pebbledb/mvcc/db.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/sei-db/db_engine/pebbledb/mvcc/db.go b/sei-db/db_engine/pebbledb/mvcc/db.go index 4793f6d305..e0b02ec1ba 100644 --- a/sei-db/db_engine/pebbledb/mvcc/db.go +++ b/sei-db/db_engine/pebbledb/mvcc/db.go @@ -59,7 +59,7 @@ const ( // compactor cannot keep up with the tombstone churn that pruning generates, // so deleted data accumulates and slows every subsequent prune scan. Allowing // Pebble to burst up to a few compactions clears that backlog. - maxConcurrentCompactions = 4 + maxConcurrentCompactions = 16 ) var ( @@ -152,12 +152,12 @@ func newPebbleOptions(config config.StateStoreConfig, cache *pebble.Cache) *pebb FormatMajorVersion: pebble.FormatVirtualSSTables, L0CompactionThreshold: 2, L0StopWritesThreshold: 1000, - LBaseMaxBytes: 64 << 20, // 64 MB - MemTableSize: 64 << 20, + LBaseMaxBytes: 256 << 20, // 64 MB + MemTableSize: 256 << 20, MemTableStopWritesThreshold: 4, // Let Pebble run several compactions in parallel so it can keep up with // the tombstone churn produced by pruning. See maxConcurrentCompactions. - CompactionConcurrencyRange: func() (int, int) { return 1, maxConcurrentCompactions }, + CompactionConcurrencyRange: func() (int, int) { return 2, maxConcurrentCompactions }, } // Configure L0 with explicit settings From 9339a2c7f2af0eb6f22bd552aae777bfbbbed4bc Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 15:35:57 -0700 Subject: [PATCH 06/11] Optimize receipt generation --- sei-db/bench/gigasim/block_generator.go | 8 +- sei-db/bench/gigasim/receipt.go | 123 +++++++++++++++------ sei-db/bench/gigasim/receipt_bloom_test.go | 99 +++++++++++++++++ 3 files changed, 191 insertions(+), 39 deletions(-) create mode 100644 sei-db/bench/gigasim/receipt_bloom_test.go diff --git a/sei-db/bench/gigasim/block_generator.go b/sei-db/bench/gigasim/block_generator.go index a073b32ac3..60e55668ce 100644 --- a/sei-db/bench/gigasim/block_generator.go +++ b/sei-db/bench/gigasim/block_generator.go @@ -3,9 +3,7 @@ package gigasim import ( "context" "fmt" - "hash" - "golang.org/x/crypto/sha3" "golang.org/x/time/rate" "github.com/sei-protocol/sei-chain/sei-db/common/metrics" @@ -77,7 +75,7 @@ type blockGenerator struct { // The keccak hasher every receipt's bloom is built with, held here because only this goroutine // builds receipts. - bloomHasher hash.Hash + receiptCache *receiptCache // This goroutine's share of a block's critical path: building it and storing it. lifecycle *metrics.PhaseTimer @@ -113,7 +111,7 @@ func newBlockGenerator( blocks: blocks, rateLimiter: rateLimiter, blocksChan: make(chan *simulatedBlock, config.MaxPendingExecutionQueueSize), - bloomHasher: sha3.NewLegacyKeccak256(), + receiptCache: newReceiptCache(), lifecycle: gigasimMetrics.NewBlockProducingTimer(), blockStoreWrite: blockStoreWrite, metrics: gigasimMetrics, @@ -201,7 +199,7 @@ func (g *blockGenerator) buildBlock() (*simulatedBlock, error) { } var receipts *receiptBuffer if g.config.EnableReceiptStore { - receipts = newReceiptBuffer(count, g.bloomHasher) + receipts = newReceiptBuffer(count, g.receiptCache) block.receipts = receipts.receipts } diff --git a/sei-db/bench/gigasim/receipt.go b/sei-db/bench/gigasim/receipt.go index 1dd6e75d3c..0177e317ec 100644 --- a/sei-db/bench/gigasim/receipt.go +++ b/sei-db/bench/gigasim/receipt.go @@ -3,7 +3,6 @@ package gigasim import ( "encoding/binary" "encoding/hex" - "hash" ethtypes "github.com/ethereum/go-ethereum/core/types" "github.com/sei-protocol/sei-chain/sei-db/common/keys" @@ -64,6 +63,80 @@ func writeSyntheticTxHash(dst []byte, rand *crand.CannedRandom, blockNumber int6 // topicsPerTransferLog is the Transfer event signature plus its two indexed address topics. const topicsPerTransferLog = 3 +// bloomBits are the three bit positions a value contributes to a log bloom. +type bloomBits [3]uint + +// bloomBitsFor derives the three bits a value sets in a log bloom. +// +// A real bloom takes them from the value's keccak digest. This one mixes the bytes instead, which +// is not a bloom any filter could match against. Nothing in the benchmark reads a log back, and the +// store is measured on what a receipt occupies rather than on what its bloom would answer, so the +// properties kept are the ones that reach the store: the same three bits per value, spread over the +// same 2048 positions, and the same bits for the same value on a rerun of the same seed. +// +// Keccak over four values per transaction was most of the cost of building a block's receipts. +func bloomBitsFor(value []byte) bloomBits { + // FNV-1a, for a spread across the bloom's positions that costs a multiply per byte. + const ( + fnvOffset uint64 = 14695981039346656037 + fnvPrime uint64 = 1099511628211 + ) + mixed := fnvOffset + for _, b := range value { + mixed ^= uint64(b) + mixed *= fnvPrime + } + var bits bloomBits + for i := range bits { + bits[i] = uint(mixed & 2047) + mixed >>= 11 + } + return bits +} + +// receiptCache holds what a receipt repeats rather than derives anew: the constant event +// signature's bloom bits, and the values that follow from a contract address. The contract pool is +// fixed, so it is worth keeping across the blocks a run produces. +// +// It is not safe for concurrent use; only the generator builds receipts. +type receiptCache struct { + signature bloomBits + contracts map[[keys.AddressLen]byte]contractFields +} + +// contractFields are the per-contract values a receipt repeats and none of its transactions change. +type contractFields struct { + bits bloomBits + hex string +} + +// newReceiptCache returns a cache with the constant inputs already resolved. +func newReceiptCache() *receiptCache { + return &receiptCache{ + signature: bloomBitsFor(erc20TransferEventSignatureBytes[:]), + contracts: make(map[[keys.AddressLen]byte]contractFields), + } +} + +// contract returns an ERC20 contract's bloom bits and hex address, resolving one it has not seen. +func (c *receiptCache) contract(address []byte) contractFields { + var key [keys.AddressLen]byte + copy(key[:], address) + if fields, ok := c.contracts[key]; ok { + return fields + } + fields := contractFields{bits: bloomBitsFor(address), hex: bytesToHex(address)} + c.contracts[key] = fields + return fields +} + +// setBits marks bits in a bloom. +func setBits(bloom *ethtypes.Bloom, bits bloomBits) { + for _, bit := range bits { + bloom[ethtypes.BloomByteLength-1-bit/8] |= byte(1 << (bit % 8)) + } +} + // receiptBuffer holds one block's receipts in a fixed number of allocations: every array a receipt // points into is carved out of a slice the buffer owns. type receiptBuffer struct { @@ -76,13 +149,13 @@ type receiptBuffer struct { blooms []ethtypes.Bloom data []byte - // The bloom hasher, which belongs to the generator rather than to any one block: it is reset - // before each use, and building one per receipt costs more than the hashing does. - hasher hash.Hash + // What the generator has already resolved about the contract pool, which is worth keeping + // across blocks rather than rebuilding per block. + cache *receiptCache } // newReceiptBuffer allocates the backing storage for one block of receipts. -func newReceiptBuffer(count int, hasher hash.Hash) *receiptBuffer { +func newReceiptBuffer(count int, cache *receiptCache) *receiptBuffer { return &receiptBuffer{ receipts: make([]*evmtypes.Receipt, count), storage: make([]evmtypes.Receipt, count), @@ -91,7 +164,7 @@ func newReceiptBuffer(count int, hasher hash.Hash) *receiptBuffer { topics: make([]string, count*topicsPerTransferLog), blooms: make([]ethtypes.Bloom, count), data: make([]byte, count*hashLen), - hasher: hasher, + cache: cache, } } @@ -99,7 +172,7 @@ func newReceiptBuffer(count int, hasher hash.Hash) *receiptBuffer { // address topics, and a bloom covering them. The values are synthetic, since the receipt store is // measured on the volume and shape of what it stores rather than on the arithmetic behind it. func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transaction, blockNumber int64) { - contractAddress := addressFromKey(txn.erc20Contract) + contract := b.cache.contract(addressFromKey(txn.erc20Contract)) senderTopic := indexedAddressTopic(addressFromKey(txn.srcAccount)) receiverTopic := indexedAddressTopic(addressFromKey(txn.dstAccount)) @@ -112,11 +185,9 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact effectiveGasPrice := receiptGasPriceBase + rand.Int64Range(0, receiptGasPriceSpan) transferAmount := receiptTransferBase + rand.Int64Range(0, receiptTransferSpan) - contractAddressHex := bytesToHex(contractAddress) - bloom := &b.blooms[index] *bloom = ethtypes.Bloom{} - b.addTransferLogToBloom(bloom, contractAddress, senderTopic[:], receiverTopic[:]) + b.addTransferLogToBloom(bloom, contract.bits, senderTopic[:], receiverTopic[:]) topics := b.topics[index*topicsPerTransferLog : (index+1)*topicsPerTransferLog] topics[0] = erc20TransferEventSignatureHex @@ -130,7 +201,7 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact log := &b.logs[index] b.logRefs[index] = log *log = evmtypes.Log{ - Address: contractAddressHex, + Address: contract.hex, Topics: topics, Data: amount, Index: 0, @@ -145,7 +216,7 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact *receipt = evmtypes.Receipt{ TxType: txType, CumulativeGasUsed: uint64(gasUsed + int64(index)*previousGas), - ContractAddress: contractAddressHex, + ContractAddress: contract.hex, TxHashHex: bytesToHex(txHash[:]), GasUsed: uint64(gasUsed), EffectiveGasPrice: uint64(effectiveGasPrice), @@ -153,7 +224,7 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact TransactionIndex: uint32(index), Status: uint32(ethtypes.ReceiptStatusSuccessful), From: bytesToHex(addressFromKey(txn.srcAccount)), - To: contractAddressHex, + To: contract.hex, Logs: b.logRefs[index : index+1], LogsBloom: bloom[:], } @@ -163,28 +234,12 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact // signature and both indexed topics. func (b *receiptBuffer) addTransferLogToBloom( bloom *ethtypes.Bloom, - contractAddress, senderTopic, receiverTopic []byte, + contractBits bloomBits, senderTopic, receiverTopic []byte, ) { - var digest [hashLen]byte - for _, value := range [4][]byte{ - contractAddress, - erc20TransferEventSignatureBytes[:], - senderTopic, - receiverTopic, - } { - addToBloom(b.hasher, &digest, bloom, value) - } -} - -// addToBloom sets the three bits a value contributes to a bloom filter. -func addToBloom(hasher hash.Hash, digest *[hashLen]byte, bloom *ethtypes.Bloom, value []byte) { - hasher.Reset() - _, _ = hasher.Write(value) - sum := hasher.Sum(digest[:0]) - for i := 0; i < 6; i += 2 { - bit := (uint(sum[i])<<8)&2047 + uint(sum[i+1]) - bloom[ethtypes.BloomByteLength-1-bit/8] |= byte(1 << (bit % 8)) - } + setBits(bloom, contractBits) + setBits(bloom, b.cache.signature) + setBits(bloom, bloomBitsFor(senderTopic)) + setBits(bloom, bloomBitsFor(receiverTopic)) } // addressFromKey takes the address out of an EVM key, which carries it after a one-byte prefix. A diff --git a/sei-db/bench/gigasim/receipt_bloom_test.go b/sei-db/bench/gigasim/receipt_bloom_test.go new file mode 100644 index 0000000000..81eaf24a28 --- /dev/null +++ b/sei-db/bench/gigasim/receipt_bloom_test.go @@ -0,0 +1,99 @@ +package gigasim + +import ( + "encoding/binary" + "math/bits" + "testing" + + ethtypes "github.com/ethereum/go-ethereum/core/types" + "github.com/sei-protocol/sei-chain/sei-db/common/keys" + "github.com/stretchr/testify/require" + "golang.org/x/crypto/sha3" +) + +// keccakBloomBits is how a real log bloom picks its bits. The benchmark's blooms are not built this +// way, and these tests hold the mixed ones to what the store sees of the difference. +func keccakBloomBits(value []byte) bloomBits { + hasher := sha3.NewLegacyKeccak256() + _, _ = hasher.Write(value) + sum := hasher.Sum(nil) + var picked bloomBits + for i := 0; i < 6; i += 2 { + picked[i/2] = (uint(sum[i])<<8)&2047 + uint(sum[i+1]) + } + return picked +} + +// bloomTestValue is a distinct value per seed. The seed is written in rather than folded into every +// byte, which wraps at 256 and would hand these tests far fewer values than they ask for. +func bloomTestValue(seed int) []byte { + value := make([]byte, hashLen) + binary.BigEndian.PutUint64(value, uint64(seed)) //nolint:gosec // seeds are small and non-negative + for i := 8; i < len(value); i++ { + value[i] = byte(i * 7) + } + return value +} + +// TestBloomBitsForFillsABloomLikeKeccakDoes pins the property the store is measured on. The bits are +// not the ones a filter would look for, but a receipt's bloom has to occupy its 256 bytes the same +// way, so the count of bits set across a corpus has to match what keccak would have set. +func TestBloomBitsForFillsABloomLikeKeccakDoes(t *testing.T) { + const values = 2048 + var mixedSet, keccakSet int + for seed := range values { + value := bloomTestValue(seed) + + var mixed, keccak ethtypes.Bloom + setBits(&mixed, bloomBitsFor(value)) + setBits(&keccak, keccakBloomBits(value)) + + mixedSet += countBloomBits(mixed) + keccakSet += countBloomBits(keccak) + } + // Three bits per value either way, less whatever collides; the collision rates have to agree. + require.InDelta(t, keccakSet, mixedSet, float64(keccakSet)*0.01, + "a mixed bloom must fill to the same density as a keccak one, or the corpus compresses differently") +} + +// TestBloomBitsForIsDeterministic pins that a rerun of the same seed produces the same corpus, which +// is what lets two runs of the benchmark be compared. +func TestBloomBitsForIsDeterministic(t *testing.T) { + for seed := range 64 { + value := bloomTestValue(seed) + require.Equal(t, bloomBitsFor(value), bloomBitsFor(value)) + } +} + +// TestBloomBitsForSeparatesValues pins that the bloom is not degenerate. Blooms that collapsed onto +// a few bit patterns would compress far better than real ones and flatter the store. +func TestBloomBitsForSeparatesValues(t *testing.T) { + const values = 4096 + seen := make(map[bloomBits]struct{}, values) + for seed := range values { + seen[bloomBitsFor(bloomTestValue(seed))] = struct{}{} + } + require.Greater(t, len(seen), values*99/100, "distinct values must land on distinct bits") +} + +// TestReceiptCacheReturnsWhatItCached pins that a contract resolved once reads back the same, since +// a receipt repeats its hex address in two fields. +func TestReceiptCacheReturnsWhatItCached(t *testing.T) { + cache := newReceiptCache() + address := make([]byte, keys.AddressLen) + for i := range address { + address[i] = byte(i) + } + first := cache.contract(address) + require.Equal(t, bytesToHex(address), first.hex) + require.Equal(t, bloomBitsFor(address), first.bits) + require.Equal(t, first, cache.contract(address), "the cached value must match the resolved one") +} + +func countBloomBits(bloom ethtypes.Bloom) int { + total := 0 + for _, b := range bloom { + total += bits.OnesCount8(b) + } + return total +} From 1954c6de0d0582eabe045e2233b578bdb8a0df76 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 15:45:04 -0700 Subject: [PATCH 07/11] Fix receipt generation --- .../dashboards/gigasim-dashboard.json | 222 ------------------ sei-db/bench/gigasim/block_generator.go | 20 +- sei-db/bench/gigasim/gigasim.go | 11 +- sei-db/bench/gigasim/gigasim_metrics.go | 12 - sei-db/bench/gigasim/receipt.go | 48 +++- sei-db/bench/gigasim/receipt_test.go | 27 +++ sei-db/bench/gigasim/receipt_writer.go | 33 +-- 7 files changed, 84 insertions(+), 289 deletions(-) diff --git a/docker/monitornode/dashboards/gigasim-dashboard.json b/docker/monitornode/dashboards/gigasim-dashboard.json index ad7d5ece2c..badc93bd3e 100644 --- a/docker/monitornode/dashboards/gigasim-dashboard.json +++ b/docker/monitornode/dashboards/gigasim-dashboard.json @@ -9071,91 +9071,6 @@ } } }, - "panel-104": { - "kind": "Panel", - "spec": { - "data": { - "kind": "QueryGroup", - "spec": { - "queries": [ - { - "kind": "PanelQuery", - "spec": { - "hidden": false, - "query": { - "datasource": { - "name": "PBFA97CFB590B2093" - }, - "group": "prometheus", - "kind": "DataQuery", - "spec": { - "editorMode": "code", - "expr": "rate(gigasim_receipt_write_phase_duration_seconds_total[$__rate_interval])", - "legendFormat": "{{phase}}", - "range": true - }, - "version": "v0" - }, - "refId": "A" - } - } - ], - "queryOptions": {}, - "transformations": [] - } - }, - "description": "What the main execution loop's write_receipts phase is made of: encoding the receipts, which the benchmark does itself, against handing them to the receipt store.", - "id": 104, - "links": [], - "title": "└ Write Receipts — encode vs store", - "vizConfig": { - "group": "piechart", - "kind": "VizConfig", - "spec": { - "fieldConfig": { - "defaults": { - "color": { - "fixedColor": "#73BF69", - "mode": "palette-classic" - }, - "custom": { - "hideFrom": { - "legend": false, - "tooltip": false, - "viz": false - } - }, - "unit": "percentunit" - }, - "overrides": [] - }, - "options": { - "legend": { - "displayMode": "list", - "placement": "bottom", - "showLegend": true, - "values": [] - }, - "pieType": "pie", - "reduceOptions": { - "calcs": [ - "lastNotNull" - ], - "fields": "", - "values": false - }, - "sort": "desc", - "tooltip": { - "hideZeros": false, - "mode": "single", - "sort": "none" - } - } - }, - "version": "13.2.1" - } - } - }, "panel-106": { "kind": "Panel", "spec": { @@ -9824,117 +9739,6 @@ } } }, - "panel-124": { - "kind": "Panel", - "spec": { - "data": { - "kind": "QueryGroup", - "spec": { - "queries": [ - { - "kind": "PanelQuery", - "spec": { - "hidden": false, - "query": { - "datasource": { - "name": "PBFA97CFB590B2093" - }, - "group": "prometheus", - "kind": "DataQuery", - "spec": { - "editorMode": "code", - "expr": "rate(gigasim_receipt_write_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", - "legendFormat": "{{phase}}", - "range": true - }, - "version": "v0" - }, - "refId": "A" - } - } - ], - "queryOptions": {}, - "transformations": [] - } - }, - "description": "What the main execution loop's write_receipts phase is made of: encoding the receipts, which the benchmark does itself, against handing them to the receipt store. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", - "id": 124, - "links": [], - "title": "└ Write Receipts — encode vs store (per block)", - "vizConfig": { - "group": "timeseries", - "kind": "VizConfig", - "spec": { - "fieldConfig": { - "defaults": { - "color": { - "mode": "palette-classic" - }, - "custom": { - "axisBorderShow": false, - "axisCenteredZero": false, - "axisColorMode": "text", - "axisLabel": "", - "axisPlacement": "auto", - "barAlignment": 0, - "barWidthFactor": 0.6, - "drawStyle": "line", - "fillOpacity": 100, - "gradientMode": "none", - "hideFrom": { - "legend": false, - "tooltip": false, - "viz": false - }, - "insertNulls": false, - "lineInterpolation": "linear", - "lineWidth": 1, - "pointSize": 5, - "scaleDistribution": { - "type": "linear" - }, - "showPoints": "auto", - "showValues": false, - "spanNulls": false, - "stacking": { - "group": "A", - "mode": "normal" - }, - "thresholdsStyle": { - "mode": "off" - } - }, - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green", - "value": 0 - } - ] - }, - "unit": "s" - }, - "overrides": [] - }, - "options": { - "legend": { - "calcs": [], - "displayMode": "list", - "placement": "bottom", - "showLegend": true - }, - "tooltip": { - "hideZeros": false, - "mode": "multi", - "sort": "desc" - } - } - }, - "version": "13.2.1" - } - } - }, "panel-125": { "kind": "Panel", "spec": { @@ -11157,32 +10961,6 @@ "y": 54 } }, - { - "kind": "GridLayoutItem", - "spec": { - "element": { - "kind": "ElementReference", - "name": "panel-104" - }, - "height": 9, - "width": 8, - "x": 0, - "y": 63 - } - }, - { - "kind": "GridLayoutItem", - "spec": { - "element": { - "kind": "ElementReference", - "name": "panel-124" - }, - "height": 9, - "width": 16, - "x": 8, - "y": 63 - } - }, { "kind": "GridLayoutItem", "spec": { diff --git a/sei-db/bench/gigasim/block_generator.go b/sei-db/bench/gigasim/block_generator.go index 60e55668ce..8e0a5b344a 100644 --- a/sei-db/bench/gigasim/block_generator.go +++ b/sei-db/bench/gigasim/block_generator.go @@ -7,7 +7,7 @@ import ( "golang.org/x/time/rate" "github.com/sei-protocol/sei-chain/sei-db/common/metrics" - evmtypes "github.com/sei-protocol/sei-chain/x/evm/types" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" ) // simulatedBlock is one block's worth of work: the transactions the execution phase runs, the payload @@ -20,8 +20,13 @@ type simulatedBlock struct { // executor pool, so a block's transaction count is also its degree of parallelism. transactions []*transaction - // The receipts written to the receipt store, empty when receipts are disabled. - receipts []*evmtypes.Receipt + // The receipts written to the receipt store, in the form it takes them, empty when receipts are + // disabled. They are marshaled here rather than on the execution loop: nothing execution does + // changes them, and the loop that hands them to the store is what paces the run. + receiptRecords []receipt.ReceiptRecord + + // What those records marshaled to, which the run reports as bytes written. + receiptBytes int64 // The transaction bytes the block store persists. These stand in for encoded transactions, which // the block store holds as opaque bytes. @@ -200,7 +205,7 @@ func (g *blockGenerator) buildBlock() (*simulatedBlock, error) { var receipts *receiptBuffer if g.config.EnableReceiptStore { receipts = newReceiptBuffer(count, g.receiptCache) - block.receipts = receipts.receipts + block.receiptRecords = receipts.records } for i := range count { @@ -212,9 +217,14 @@ func (g *blockGenerator) buildBlock() (*simulatedBlock, error) { block.payload[i] = g.accounts.Rand().Bytes(g.config.BytesPerTransaction) if receipts != nil { - receipts.build(i, g.accounts.Rand(), txn, number) + if err := receipts.build(i, g.accounts.Rand(), txn, number); err != nil { + return nil, err + } } } + if receipts != nil { + block.receiptBytes = receipts.encodedBytes + } // Accounts minted for this block become legal read targets once it is complete. g.accounts.ReportEndOfBlock() diff --git a/sei-db/bench/gigasim/gigasim.go b/sei-db/bench/gigasim/gigasim.go index 0cc2796c70..c931977ba2 100644 --- a/sei-db/bench/gigasim/gigasim.go +++ b/sei-db/bench/gigasim/gigasim.go @@ -18,7 +18,7 @@ import ( crand "github.com/sei-protocol/sei-chain/sei-db/common/rand" "github.com/sei-protocol/sei-chain/sei-db/common/utils" dbconfig "github.com/sei-protocol/sei-chain/sei-db/config" - evmtypes "github.com/sei-protocol/sei-chain/x/evm/types" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" ) // GigaSim runs the benchmark, driving generated blocks through the block store, the state DB and the @@ -401,7 +401,7 @@ func (g *GigaSim) finalizeSetupBlock() error { if err := g.blocks.writeBlock(number, payload); err != nil { return err } - if err := g.persistExecutionResults(number, nil, g.accounts.Counters()); err != nil { + if err := g.persistExecutionResults(number, nil, 0, g.accounts.Counters()); err != nil { return err } g.accounts.ReportEndOfBlock() @@ -472,7 +472,7 @@ func (g *GigaSim) halt() { func (g *GigaSim) executeAndRecord(block *simulatedBlock) error { g.executeBlock(block) - if err := g.persistExecutionResults(block.number, block.receipts, block.counters); err != nil { + if err := g.persistExecutionResults(block.number, block.receiptRecords, block.receiptBytes, block.counters); err != nil { return err } @@ -514,12 +514,13 @@ func (g *GigaSim) executeBlock(block *simulatedBlock) { // the reverse leaves committed state whose receipts were dropped. func (g *GigaSim) persistExecutionResults( number int64, - receipts []*evmtypes.Receipt, + records []receipt.ReceiptRecord, + receiptBytes int64, counters identifierCounters, ) error { if g.receipts != nil { g.lifecycle.SetPhase("write_receipts") - if err := g.receipts.writeBlock(number, receipts); err != nil { + if err := g.receipts.writeBlock(number, records, receiptBytes); err != nil { return err } } diff --git a/sei-db/bench/gigasim/gigasim_metrics.go b/sei-db/bench/gigasim/gigasim_metrics.go index c35985c29a..20c6675a17 100644 --- a/sei-db/bench/gigasim/gigasim_metrics.go +++ b/sei-db/bench/gigasim/gigasim_metrics.go @@ -52,7 +52,6 @@ type GigasimMetrics struct { executionLoopPhases *metrics.PhaseTimerFactory blockProducingPhases *metrics.PhaseTimerFactory blockStoreWritePhases *metrics.PhaseTimerFactory - receiptWritePhases *metrics.PhaseTimerFactory pendingExecutionQueue *metrics.QueueMeter } @@ -160,7 +159,6 @@ func NewGigasimMetrics() *GigasimMetrics { executionLoopPhases: metrics.NewPhaseTimerFactory(meter, "gigasim_execution_loop").RecordLatencies(), blockProducingPhases: metrics.NewPhaseTimerFactory(meter, "gigasim_block_producing_loop").RecordLatencies(), blockStoreWritePhases: metrics.NewPhaseTimerFactory(meter, "gigasim_blockstore_write"), - receiptWritePhases: metrics.NewPhaseTimerFactory(meter, "gigasim_receipt_write"), pendingExecutionQueue: metrics.NewQueueMeter(meter, "gigasim_pending_execution"), } } @@ -197,16 +195,6 @@ func (m *GigasimMetrics) NewBlockStoreWriteTimer() *metrics.PhaseTimer { return m.blockStoreWritePhases.Build() } -// NewReceiptWriteTimer returns the timer breaking a receipt write into encoding the receipts and -// handing them to the store. It subdivides the execution loop's write_receipts phase rather than -// adding to it. -func (m *GigasimMetrics) NewReceiptWriteTimer() *metrics.PhaseTimer { - if m == nil || m.receiptWritePhases == nil { - return nil - } - return m.receiptWritePhases.Build() -} - // NewTransactionPhaseTimer returns a phase timer for one executor. Each executor needs its own: a // timer tracks a single thread's current phase. func (m *GigasimMetrics) NewTransactionPhaseTimer() *metrics.PhaseTimer { diff --git a/sei-db/bench/gigasim/receipt.go b/sei-db/bench/gigasim/receipt.go index 0177e317ec..a28a3a4742 100644 --- a/sei-db/bench/gigasim/receipt.go +++ b/sei-db/bench/gigasim/receipt.go @@ -3,10 +3,13 @@ package gigasim import ( "encoding/binary" "encoding/hex" + "fmt" + "github.com/ethereum/go-ethereum/common" ethtypes "github.com/ethereum/go-ethereum/core/types" "github.com/sei-protocol/sei-chain/sei-db/common/keys" crand "github.com/sei-protocol/sei-chain/sei-db/common/rand" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" evmtypes "github.com/sei-protocol/sei-chain/x/evm/types" ) @@ -140,7 +143,11 @@ func setBits(bloom *ethtypes.Bloom, bits bloomBits) { // receiptBuffer holds one block's receipts in a fixed number of allocations: every array a receipt // points into is carved out of a slice the buffer owns. type receiptBuffer struct { - receipts []*evmtypes.Receipt + // The records the store is handed, marshaled as each receipt is built. + records []receipt.ReceiptRecord + + // The total size of what those records marshaled to. + encodedBytes int64 storage []evmtypes.Receipt logs []evmtypes.Log @@ -157,21 +164,21 @@ type receiptBuffer struct { // newReceiptBuffer allocates the backing storage for one block of receipts. func newReceiptBuffer(count int, cache *receiptCache) *receiptBuffer { return &receiptBuffer{ - receipts: make([]*evmtypes.Receipt, count), - storage: make([]evmtypes.Receipt, count), - logs: make([]evmtypes.Log, count), - logRefs: make([]*evmtypes.Log, count), - topics: make([]string, count*topicsPerTransferLog), - blooms: make([]ethtypes.Bloom, count), - data: make([]byte, count*hashLen), - cache: cache, + records: make([]receipt.ReceiptRecord, count), + storage: make([]evmtypes.Receipt, count), + logs: make([]evmtypes.Log, count), + logRefs: make([]*evmtypes.Log, count), + topics: make([]string, count*topicsPerTransferLog), + blooms: make([]ethtypes.Bloom, count), + data: make([]byte, count*hashLen), + cache: cache, } } // build fills in the receipt an ERC20 transfer would leave behind: one Transfer log with two indexed // address topics, and a bloom covering them. The values are synthetic, since the receipt store is // measured on the volume and shape of what it stores rather than on the arithmetic behind it. -func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transaction, blockNumber int64) { +func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transaction, blockNumber int64) error { contract := b.cache.contract(addressFromKey(txn.erc20Contract)) senderTopic := indexedAddressTopic(addressFromKey(txn.srcAccount)) receiverTopic := indexedAddressTopic(addressFromKey(txn.dstAccount)) @@ -210,10 +217,9 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact var txHash [hashLen]byte writeSyntheticTxHash(txHash[:], rand, blockNumber, index) - receipt := &b.storage[index] - b.receipts[index] = receipt + built := &b.storage[index] //nolint:gosec // G115 - benchmark values are bounded well below the conversion limits - *receipt = evmtypes.Receipt{ + *built = evmtypes.Receipt{ TxType: txType, CumulativeGasUsed: uint64(gasUsed + int64(index)*previousGas), ContractAddress: contract.hex, @@ -228,6 +234,22 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact Logs: b.logRefs[index : index+1], LogsBloom: bloom[:], } + + // Marshaled here rather than where the store is called: the receipt is final once built, and the + // loop that calls the store is the one pacing the run. The hash goes over as bytes for the same + // reason, the store's caller having had to parse the hex form back otherwise. + encoded, err := built.Marshal() + if err != nil { + return fmt.Errorf("failed to marshal the receipt for transaction %d of block %d: %w", + index, blockNumber, err) + } + b.encodedBytes += int64(len(encoded)) + b.records[index] = receipt.ReceiptRecord{ + TxHash: common.BytesToHash(txHash[:]), + Receipt: built, + ReceiptBytes: encoded, + } + return nil } // addTransferLogToBloom sets the bits a Transfer log contributes: the emitting contract, the event diff --git a/sei-db/bench/gigasim/receipt_test.go b/sei-db/bench/gigasim/receipt_test.go index 5e89ee5b22..be143ad2fd 100644 --- a/sei-db/bench/gigasim/receipt_test.go +++ b/sei-db/bench/gigasim/receipt_test.go @@ -3,8 +3,10 @@ package gigasim import ( "testing" + "github.com/ethereum/go-ethereum/common" "github.com/stretchr/testify/require" + "github.com/sei-protocol/sei-chain/sei-db/common/keys" crand "github.com/sei-protocol/sei-chain/sei-db/common/rand" ) @@ -34,6 +36,31 @@ func TestSyntheticTxHashesAreUniqueAcrossPositions(t *testing.T) { } } +// TestBuiltRecordKeysOnItsOwnReceiptHash pins the key a record carries to the hash inside the +// receipt it carries. The store keys on the record's TxHash and refuses a block repeating one, so a +// record keyed on anything other than its own receipt would either collide or hide the receipt. +func TestBuiltRecordKeysOnItsOwnReceiptHash(t *testing.T) { + t.Parallel() + + const count = 8 + buffer := newReceiptBuffer(count, newReceiptCache()) + rand := crand.NewCannedRandom(1<<20, 1337) + txn := &transaction{ + erc20Contract: make([]byte, 1+keys.AddressLen+hashLen), + srcAccount: make([]byte, 1+keys.AddressLen+hashLen), + dstAccount: make([]byte, 1+keys.AddressLen+hashLen), + } + + for index := range count { + require.NoError(t, buffer.build(index, rand, txn, 3)) + + record := buffer.records[index] + require.Equal(t, common.HexToHash(record.Receipt.TxHashHex), record.TxHash, + "the record's key must be the hash its own receipt reports") + require.NotEmpty(t, record.ReceiptBytes, "a record reaches the store already marshaled") + } +} + // A hash is recomputable from its position alone, which is what lets a run's transaction hashes be // derived rather than stored. func TestSyntheticTxHashDependsOnlyOnItsPosition(t *testing.T) { diff --git a/sei-db/bench/gigasim/receipt_writer.go b/sei-db/bench/gigasim/receipt_writer.go index 6f65709e0d..5ace22f9a9 100644 --- a/sei-db/bench/gigasim/receipt_writer.go +++ b/sei-db/bench/gigasim/receipt_writer.go @@ -3,12 +3,9 @@ package gigasim import ( "fmt" - "github.com/ethereum/go-ethereum/common" sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" - "github.com/sei-protocol/sei-chain/sei-db/common/metrics" "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" tmproto "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/types" - evmtypes "github.com/sei-protocol/sei-chain/x/evm/types" ) // receiptWriter persists a block's receipts through the production write path. It exists only when @@ -16,10 +13,6 @@ import ( type receiptWriter struct { store receipt.ReceiptStore - // Splits this writer's work into encoding the receipts and handing them to the store, subdividing - // the execution loop's write_receipts phase. Only that loop writes receipts, so one timer serves it. - phases *metrics.PhaseTimer - metrics *GigasimMetrics } @@ -27,7 +20,6 @@ type receiptWriter struct { func newReceiptWriter(store receipt.ReceiptStore, gigasimMetrics *GigasimMetrics) *receiptWriter { return &receiptWriter{ store: store, - phases: gigasimMetrics.NewReceiptWriteTimer(), metrics: gigasimMetrics, } } @@ -38,30 +30,7 @@ func newReceiptWriter(store receipt.ReceiptStore, gigasimMetrics *GigasimMetrics // skipping the call would leave the receipt head behind the ledger and the state for every setup // block — heights recovery takes the minimum of, so a run interrupted during setup over an existing // directory would roll state back to a height the ledger has passed and then refuse to reopen. -func (w *receiptWriter) writeBlock(number int64, receipts []*evmtypes.Receipt) error { - // Closes the phase in flight, so the gap until the next block's receipts is charged to neither. - defer w.phases.Reset() - - w.phases.SetPhase("encode") - var encodedBytes int64 - records := make([]receipt.ReceiptRecord, 0, len(receipts)) - for _, rcpt := range receipts { - // The store accepts pre-marshaled bytes, and marshaling here keeps the cost of producing them - // attributed to the benchmark rather than to the store. - encoded, err := rcpt.Marshal() - if err != nil { - return fmt.Errorf("failed to marshal the receipt for transaction %d of block %d: %w", - rcpt.TransactionIndex, number, err) - } - encodedBytes += int64(len(encoded)) - records = append(records, receipt.ReceiptRecord{ - TxHash: common.HexToHash(rcpt.TxHashHex), - Receipt: rcpt, - ReceiptBytes: encoded, - }) - } - - w.phases.SetPhase("store_write") +func (w *receiptWriter) writeBlock(number int64, records []receipt.ReceiptRecord, encodedBytes int64) error { if err := w.store.SetReceipts(sdk.NewContext(nil, tmproto.Header{Height: number}, false), records); err != nil { return fmt.Errorf("failed to write the receipts for block %d: %w", number, err) } From eb658a9d77690e980ec986ed4c500e7b7311e206 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 15:52:55 -0700 Subject: [PATCH 08/11] Revert separate DB --- sei-db/config/giga_config.go | 1 - 1 file changed, 1 deletion(-) diff --git a/sei-db/config/giga_config.go b/sei-db/config/giga_config.go index 478361f014..b97a6bc3bf 100644 --- a/sei-db/config/giga_config.go +++ b/sei-db/config/giga_config.go @@ -45,7 +45,6 @@ func DefaultGigaStorageConfig(homePath string) (*GigaStorageConfig, error) { ssConfig.EVMDBDirectory = utils.GetEVMStateStorePath(homePath, ssConfig.Backend) ssConfig.ExternalPruning = true ssConfig.DisableInternalWAL = true - ssConfig.SeparateEVMSubDBs = true receiptConfig := DefaultReceiptStoreConfig() receiptConfig.Backend = gigaReceiptBackend From c9060307849563c8830c6fbb306b1613b27d00d1 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 16:02:17 -0700 Subject: [PATCH 09/11] Address comments --- sei-db/config/ss_config.go | 7 +-- .../ledger_db/receipt/litt_receipt_store.go | 49 +++++++++--------- .../litt_write_failure_internal_test.go | 31 ++++++++++++ sei-db/state_db/giga/state_db.go | 26 ++++++---- sei-db/state_db/giga/state_db_replay_test.go | 50 +++++++++++++++---- 5 files changed, 117 insertions(+), 46 deletions(-) diff --git a/sei-db/config/ss_config.go b/sei-db/config/ss_config.go index cde25c190f..f2af85fd2e 100644 --- a/sei-db/config/ss_config.go +++ b/sei-db/config/ss_config.go @@ -43,9 +43,10 @@ type StateStoreConfig struct { // into this store: giga's StateDB writes its state WAL before the store and catches the store up // from it on open, which makes a second log here written and never read. // - // Like KeepRecent this is not read from the state-store config, since it is only correct when the - // owner supplies the log. Rollback through ss/composite replays the changelog and has no other - // source for the versions above a snapshot, so that path must leave it on. + // Like ExternalPruning this is not read from the state-store config but set by the code wiring + // the store into its owner, since it is only correct when that owner supplies the log. Rollback + // through ss/composite replays the changelog and has no other source for the versions above a + // snapshot, so that path must leave it on. DisableInternalWAL bool `mapstructure:"-"` // KeepRecent defines the number of versions to keep in state store (shared by Cosmos and EVM). diff --git a/sei-db/ledger_db/receipt/litt_receipt_store.go b/sei-db/ledger_db/receipt/litt_receipt_store.go index 779917e6b1..4ff3597da2 100644 --- a/sei-db/ledger_db/receipt/litt_receipt_store.go +++ b/sei-db/ledger_db/receipt/litt_receipt_store.go @@ -91,7 +91,14 @@ type littReceiptStore struct { // queue costs the caller. A whole write is queued — bodies, log index and version marker — so // the queue's depth is the depth of the receipt write itself. Its capacity bounds how far the // store may fall behind the chain; a nil channel means writes are applied on the caller. - writes chan receiptWrite + writes chan receiptWrite + + // Orders admitting a write against shutting the writer down. queueWrite holds it shared for the + // length of a send; Close takes it exclusively to refuse further writes before stopping the + // writer, so no write is ever accepted into a queue that will not be drained. + admission sync.RWMutex + closing bool + writeQueue *seidbmetrics.QueueMeter writeErr atomic.Pointer[error] stopSampling context.CancelFunc @@ -360,31 +367,19 @@ var ErrStoreClosed = errors.New("receipt store is closed") // send after that point would sit in a channel nobody reads, and a send once the queue is full // would never return — inside a commit, which hangs the node instead of failing it. func (s *littReceiptStore) queueWrite(write receiptWrite) error { - // Checked before the send rather than only alongside it: select picks uniformly among ready - // cases, so a send with room available would win half the races against an already-closed store. - select { - case <-s.stopBackground: + // Held across the send, not merely to read the flag. Close takes the same lock exclusively before + // it stops the writer, so a write admitted here reaches a writer that is still running, and one + // arriving after Close has begun is refused instead of landing in a queue nobody will drain. + // + // A send blocked on a full queue holds the lock and delays Close, which is the intended order: the + // writer is still draining, so the send completes and Close proceeds behind it. + s.admission.RLock() + defer s.admission.RUnlock() + if s.closing { return ErrStoreClosed - default: } - return s.writeQueue.SendVia( - func() bool { - select { - case s.writes <- write: - return true - default: - return false - } - }, - func() error { - select { - case s.writes <- write: - return nil - case <-s.stopBackground: - return ErrStoreClosed - } - }, - ) + seidbmetrics.Send(s.writeQueue, s.writes, write) + return nil } // applyReceipts writes a block's receipt bodies, log index and version marker. The bodies go to @@ -587,6 +582,12 @@ func (s *littReceiptStore) startFlusher() { func (s *littReceiptStore) Close() error { var err error s.closeOnce.Do(func() { + // Before the writer is stopped, and exclusive, so it takes effect only once the writes already + // admitted have been handed over. + s.admission.Lock() + s.closing = true + s.admission.Unlock() + if s.stopSampling != nil { s.stopSampling() } diff --git a/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go b/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go index 4c178ba436..eff942b910 100644 --- a/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go +++ b/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go @@ -150,6 +150,37 @@ func TestWriteAfterCloseIsRefusedWithAFullQueue(t *testing.T) { } } +// TestWriteRacingCloseIsEitherAppliedOrRefused covers the ordering the two tests above cannot reach, +// where a write is admitted while Close is running rather than after it has returned. Admission and +// shutdown have to be mutually exclusive: a write that returns success must have reached a writer +// that was still running, so the height it reports is one the store actually holds. +func TestWriteRacingCloseIsEitherAppliedOrRefused(t *testing.T) { + for attempt := range 50 { + s, _ := setupLittCtxStore(t) + + addr := common.HexToAddress("0xfa51") + topic := common.HexToHash("0xfa52") + txHash, rcpt := littCtxTestReceipt(1, 0, addr, topic, 1) + + started := make(chan struct{}) + result := make(chan error, 1) + go func() { + close(started) + result <- s.SetReceipts(newTestCtxAtHeight(1), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}}) + }() + <-started + require.NoError(t, s.Close()) + + if err := <-result; err != nil { + require.ErrorIs(t, err, ErrStoreClosed, "attempt %d", attempt) + continue + } + // Accepted, so the writer must have applied it before Close let the writer go. + require.Equal(t, int64(1), s.LatestVersion(), + "attempt %d: a write that reported success must have been applied", attempt) + } +} + func writeOneReceipt(t *testing.T, s *littReceiptStore, block uint64, addr common.Address, topic common.Hash) { t.Helper() txHash, rcpt := littCtxTestReceipt(block, 0, addr, topic, 1) diff --git a/sei-db/state_db/giga/state_db.go b/sei-db/state_db/giga/state_db.go index 56083ef5be..a714e86e2a 100644 --- a/sei-db/state_db/giga/state_db.go +++ b/sei-db/state_db/giga/state_db.go @@ -69,6 +69,18 @@ const gigaMeterName = "seidb_giga" // // The returned StateDB owns all three stores and closes them on Close. A failed call closes whatever it // had already opened. +// stateStoreConfigFor is the state store config a StateDB opens SS with, settled once here rather +// than at each open so that every path reaches the same databases. The rollback path opens them too, +// through DiscardStateAbove, and a config differing there would leave it writing a changelog beside +// stores the commit path keeps none for. +// +// The changelog is off because this StateDB already logs every block: its state WAL is written +// before SS and is what catchUpTo replays into it, so a second log inside SS is never read. +func stateStoreConfigFor(cfg config.StateStoreConfig) config.StateStoreConfig { + cfg.DisableInternalWAL = true + return cfg +} + func NewStateDB( ctx context.Context, flatkvCfg *flatkvconfig.Config, @@ -77,7 +89,7 @@ func NewStateDB( ) (db *StateDB, retErr error) { s := &StateDB{ flatkvCfg: flatkvCfg, - ssCfg: ssCfg, + ssCfg: stateStoreConfigFor(ssCfg), commitPhases: metrics.NewPhaseTimerFactory(otel.Meter(gigaMeterName), commitPhaseTimerName). RecordLatencies().Build(), } @@ -131,7 +143,7 @@ func NewStateDBWithRollback( } // rewindTo only moves files, so it needs no store open, only where they live. - offline := &StateDB{flatkvCfg: flatkvCfg, ssCfg: ssCfg} + offline := &StateDB{flatkvCfg: flatkvCfg, ssCfg: stateStoreConfigFor(ssCfg)} if err := offline.rewindTo(target); err != nil { return nil, err } @@ -192,18 +204,12 @@ func (s *StateDB) openSS() error { if !s.ssCfg.Enable { return nil } - // The state WAL this StateDB writes before every commit is what catchUpTo replays into SS, and - // rollback rewinds SS from its snapshots against that same WAL. A changelog inside SS would be - // a second log of every block that nothing here reads, paid for on the commit path. - ssCfg := s.ssCfg - ssCfg.DisableInternalWAL = true - - ss, err := evm.NewEVMStateStore(ssCfg.EVMDBDirectory, ssCfg) + ss, err := evm.NewEVMStateStore(s.ssCfg.EVMDBDirectory, s.ssCfg) if err != nil { return fmt.Errorf("open EVM state store: %w", err) } s.ss = ss - if err := s.ss.StartSnapshots(s.ssSnapshotRoot(), ssCfg, nil); err != nil { + if err := s.ss.StartSnapshots(s.ssSnapshotRoot(), s.ssCfg, nil); err != nil { return fmt.Errorf("start EVM state store snapshot manager: %w", err) } return nil diff --git a/sei-db/state_db/giga/state_db_replay_test.go b/sei-db/state_db/giga/state_db_replay_test.go index 8946ade3da..792e162cda 100644 --- a/sei-db/state_db/giga/state_db_replay_test.go +++ b/sei-db/state_db/giga/state_db_replay_test.go @@ -16,18 +16,50 @@ import ( // SS keeps no changelog of its own under giga: the state WAL written before every commit is what // catchUpTo replays into it, so a second log would be written on the commit path and never read. // Recovery rests on that, which is why the absence is pinned rather than left to the config. -func TestOpenSSKeepsNoChangelogOfItsOwn(t *testing.T) { - s := &StateDB{ - flatkvCfg: flatkvconfig.DefaultTestConfig(t), - ssCfg: config.DefaultStateStoreConfig(), +func TestGigaOpensSSWithoutAChangelog(t *testing.T) { + newStateDB := func(t *testing.T) *StateDB { + t.Helper() + ssCfg := config.DefaultStateStoreConfig() + ssCfg.Enable = true + ssCfg.EVMDBDirectory = filepath.Join(t.TempDir(), "ss") + return &StateDB{ + flatkvCfg: flatkvconfig.DefaultTestConfig(t), + // As the constructors settle it, which is what makes both paths below agree. + ssCfg: stateStoreConfigFor(ssCfg), + } } - s.ssCfg.Enable = true - s.ssCfg.EVMDBDirectory = filepath.Join(t.TempDir(), "ss") - require.NoError(t, s.openSS()) - t.Cleanup(func() { _ = s.ss.Close() }) + t.Run("opened to commit", func(t *testing.T) { + s := newStateDB(t) + require.NoError(t, s.openSS()) + t.Cleanup(func() { _ = s.ss.Close() }) + requireNoSSChangelog(t, s.ssCfg.EVMDBDirectory) + }) + + // The rollback path opens the same databases through DiscardStateAbove rather than openSS, so it + // is the one a config settled per-open would miss. It reaches them via StoredVersions, which + // returns without opening anything when the directory is absent, so the store has to exist first. + t.Run("opened to roll back", func(t *testing.T) { + s := newStateDB(t) + require.NoError(t, s.openSS()) + require.NoError(t, s.ss.Close()) + + require.NoError(t, s.discardStateAbove(storedWALRange{first: 1, last: 9}, 7)) + requireNoSSChangelog(t, s.ssCfg.EVMDBDirectory) + }) +} - changelog := utils.GetChangelogPath(s.ssCfg.EVMDBDirectory) +// TestStateStoreConfigForDisablesTheInternalWAL pins what the constructors apply, since every path +// that opens SS reads the config they settled rather than disabling the log for itself. +func TestStateStoreConfigForDisablesTheInternalWAL(t *testing.T) { + handedIn := config.DefaultStateStoreConfig() + require.False(t, handedIn.DisableInternalWAL, "a caller is not expected to have set it") + require.True(t, stateStoreConfigFor(handedIn).DisableInternalWAL) +} + +func requireNoSSChangelog(t *testing.T, evmDBDirectory string) { + t.Helper() + changelog := utils.GetChangelogPath(evmDBDirectory) _, err := os.Stat(changelog) require.True(t, os.IsNotExist(err), "SS must keep no changelog under giga; found one at %s", changelog) From a06e3b77d0f5e65609b5a08e75508a97063058f7 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 16:07:57 -0700 Subject: [PATCH 10/11] Simplify godoc --- sei-db/bench/gigasim/block_generator.go | 4 +- sei-db/bench/gigasim/receipt.go | 28 +++----- sei-db/bench/gigasim/receipt_bloom_test.go | 19 +++--- sei-db/bench/gigasim/receipt_test.go | 5 +- sei-db/config/receipt_config.go | 21 ++---- sei-db/config/ss_config.go | 13 ++-- sei-db/db_engine/pebbledb/mvcc/db.go | 11 ++- .../ledger_db/receipt/litt_receipt_store.go | 68 ++++++------------- .../litt_write_failure_internal_test.go | 32 ++++----- sei-db/ledger_db/receipt/littidx_test.go | 5 +- sei-db/ledger_db/receipt/receipt_store.go | 14 ++-- sei-db/state_db/giga/state_db.go | 10 +-- sei-db/state_db/giga/state_db_replay_test.go | 15 ++-- 13 files changed, 89 insertions(+), 156 deletions(-) diff --git a/sei-db/bench/gigasim/block_generator.go b/sei-db/bench/gigasim/block_generator.go index 8e0a5b344a..597a151b85 100644 --- a/sei-db/bench/gigasim/block_generator.go +++ b/sei-db/bench/gigasim/block_generator.go @@ -21,8 +21,8 @@ type simulatedBlock struct { transactions []*transaction // The receipts written to the receipt store, in the form it takes them, empty when receipts are - // disabled. They are marshaled here rather than on the execution loop: nothing execution does - // changes them, and the loop that hands them to the store is what paces the run. + // disabled. They are marshaled here because execution does not change them and its loop paces + // the run. receiptRecords []receipt.ReceiptRecord // What those records marshaled to, which the run reports as bytes written. diff --git a/sei-db/bench/gigasim/receipt.go b/sei-db/bench/gigasim/receipt.go index a28a3a4742..b3824558af 100644 --- a/sei-db/bench/gigasim/receipt.go +++ b/sei-db/bench/gigasim/receipt.go @@ -69,15 +69,10 @@ const topicsPerTransferLog = 3 // bloomBits are the three bit positions a value contributes to a log bloom. type bloomBits [3]uint -// bloomBitsFor derives the three bits a value sets in a log bloom. -// -// A real bloom takes them from the value's keccak digest. This one mixes the bytes instead, which -// is not a bloom any filter could match against. Nothing in the benchmark reads a log back, and the -// store is measured on what a receipt occupies rather than on what its bloom would answer, so the -// properties kept are the ones that reach the store: the same three bits per value, spread over the -// same 2048 positions, and the same bits for the same value on a rerun of the same seed. -// -// Keccak over four values per transaction was most of the cost of building a block's receipts. +// bloomBitsFor derives the three bits a value sets in a log bloom by mixing its bytes rather than +// hashing them, which no filter could match against. Nothing here reads a log back, so what is kept +// is what reaches the store: the same bits per value, the same spread, and the same bits on a +// rerun of the seed. func bloomBitsFor(value []byte) bloomBits { // FNV-1a, for a spread across the bloom's positions that costs a multiply per byte. const ( @@ -97,11 +92,8 @@ func bloomBitsFor(value []byte) bloomBits { return bits } -// receiptCache holds what a receipt repeats rather than derives anew: the constant event -// signature's bloom bits, and the values that follow from a contract address. The contract pool is -// fixed, so it is worth keeping across the blocks a run produces. -// -// It is not safe for concurrent use; only the generator builds receipts. +// receiptCache holds what a receipt repeats rather than derives anew: the event signature's bloom +// bits, and the values that follow from a contract address. It is not safe for concurrent use. type receiptCache struct { signature bloomBits contracts map[[keys.AddressLen]byte]contractFields @@ -156,8 +148,7 @@ type receiptBuffer struct { blooms []ethtypes.Bloom data []byte - // What the generator has already resolved about the contract pool, which is worth keeping - // across blocks rather than rebuilding per block. + // What the generator has already resolved about the contract pool, kept across blocks. cache *receiptCache } @@ -235,9 +226,8 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact LogsBloom: bloom[:], } - // Marshaled here rather than where the store is called: the receipt is final once built, and the - // loop that calls the store is the one pacing the run. The hash goes over as bytes for the same - // reason, the store's caller having had to parse the hex form back otherwise. + // Marshaled here rather than on the execution loop, which is what paces the run. The receipt is + // final once built, so nothing downstream changes what this encodes. encoded, err := built.Marshal() if err != nil { return fmt.Errorf("failed to marshal the receipt for transaction %d of block %d: %w", diff --git a/sei-db/bench/gigasim/receipt_bloom_test.go b/sei-db/bench/gigasim/receipt_bloom_test.go index 81eaf24a28..19d6d5c8e1 100644 --- a/sei-db/bench/gigasim/receipt_bloom_test.go +++ b/sei-db/bench/gigasim/receipt_bloom_test.go @@ -11,8 +11,7 @@ import ( "golang.org/x/crypto/sha3" ) -// keccakBloomBits is how a real log bloom picks its bits. The benchmark's blooms are not built this -// way, and these tests hold the mixed ones to what the store sees of the difference. +// keccakBloomBits is how a real log bloom picks its bits, which these tests compare against. func keccakBloomBits(value []byte) bloomBits { hasher := sha3.NewLegacyKeccak256() _, _ = hasher.Write(value) @@ -25,7 +24,7 @@ func keccakBloomBits(value []byte) bloomBits { } // bloomTestValue is a distinct value per seed. The seed is written in rather than folded into every -// byte, which wraps at 256 and would hand these tests far fewer values than they ask for. +// byte, which wraps at 256 and would yield far fewer values than asked for. func bloomTestValue(seed int) []byte { value := make([]byte, hashLen) binary.BigEndian.PutUint64(value, uint64(seed)) //nolint:gosec // seeds are small and non-negative @@ -35,9 +34,8 @@ func bloomTestValue(seed int) []byte { return value } -// TestBloomBitsForFillsABloomLikeKeccakDoes pins the property the store is measured on. The bits are -// not the ones a filter would look for, but a receipt's bloom has to occupy its 256 bytes the same -// way, so the count of bits set across a corpus has to match what keccak would have set. +// TestBloomBitsForFillsABloomLikeKeccakDoes pins the property the store is measured on: the bits +// differ from a real bloom's, but a corpus has to set as many of them as keccak would. func TestBloomBitsForFillsABloomLikeKeccakDoes(t *testing.T) { const values = 2048 var mixedSet, keccakSet int @@ -57,7 +55,7 @@ func TestBloomBitsForFillsABloomLikeKeccakDoes(t *testing.T) { } // TestBloomBitsForIsDeterministic pins that a rerun of the same seed produces the same corpus, which -// is what lets two runs of the benchmark be compared. +// is what lets two runs be compared. func TestBloomBitsForIsDeterministic(t *testing.T) { for seed := range 64 { value := bloomTestValue(seed) @@ -65,8 +63,8 @@ func TestBloomBitsForIsDeterministic(t *testing.T) { } } -// TestBloomBitsForSeparatesValues pins that the bloom is not degenerate. Blooms that collapsed onto -// a few bit patterns would compress far better than real ones and flatter the store. +// TestBloomBitsForSeparatesValues pins that blooms do not collapse onto a few bit patterns, which +// would compress better than real ones and flatter the store. func TestBloomBitsForSeparatesValues(t *testing.T) { const values = 4096 seen := make(map[bloomBits]struct{}, values) @@ -76,8 +74,7 @@ func TestBloomBitsForSeparatesValues(t *testing.T) { require.Greater(t, len(seen), values*99/100, "distinct values must land on distinct bits") } -// TestReceiptCacheReturnsWhatItCached pins that a contract resolved once reads back the same, since -// a receipt repeats its hex address in two fields. +// TestReceiptCacheReturnsWhatItCached pins that a contract resolved once reads back the same. func TestReceiptCacheReturnsWhatItCached(t *testing.T) { cache := newReceiptCache() address := make([]byte, keys.AddressLen) diff --git a/sei-db/bench/gigasim/receipt_test.go b/sei-db/bench/gigasim/receipt_test.go index be143ad2fd..d8c9c11fff 100644 --- a/sei-db/bench/gigasim/receipt_test.go +++ b/sei-db/bench/gigasim/receipt_test.go @@ -36,9 +36,8 @@ func TestSyntheticTxHashesAreUniqueAcrossPositions(t *testing.T) { } } -// TestBuiltRecordKeysOnItsOwnReceiptHash pins the key a record carries to the hash inside the -// receipt it carries. The store keys on the record's TxHash and refuses a block repeating one, so a -// record keyed on anything other than its own receipt would either collide or hide the receipt. +// TestBuiltRecordKeysOnItsOwnReceiptHash pins a record's key to the hash inside the receipt it +// carries. The store keys on TxHash, so a record keyed on anything else would hide its receipt. func TestBuiltRecordKeysOnItsOwnReceiptHash(t *testing.T) { t.Parallel() diff --git a/sei-db/config/receipt_config.go b/sei-db/config/receipt_config.go index c9de072f2c..b0a7cf1d04 100644 --- a/sei-db/config/receipt_config.go +++ b/sei-db/config/receipt_config.go @@ -28,10 +28,8 @@ const ( // littidx eth_getLogs (see ReceiptStoreConfig.LogFilterParallelism). const DefaultReceiptLogFilterParallelism = 16 -// DefaultReceiptAsyncWriteBuffer is the default queue depth for receipt writes -// (see ReceiptStoreConfig.AsyncWriteBuffer). It is small because the queue's -// depth is how far an unclean exit sets recovery back, not only how much burst -// the writer can absorb. +// DefaultReceiptAsyncWriteBuffer is the default queue depth for receipt writes. It is small because +// the depth is also how far an unclean exit sets recovery back. const DefaultReceiptAsyncWriteBuffer = 10 // ReceiptStoreConfig defines configuration for the receipt store database. @@ -53,18 +51,11 @@ type ReceiptStoreConfig struct { Backend string `mapstructure:"rs-backend"` // AsyncWriteBuffer defines the async queue length for commits to be applied to receipt store. - // It bounds how many blocks the store may fall behind the chain before a write blocks, and so - // how far LatestVersion may trail the height just written. + // It bounds how many blocks the store may fall behind the chain before a write blocks. // - // Raising it costs more than the memory it holds. The queue is not on disk, so an unclean exit - // loses it, and recovery converges every store on the lowest head: a receipt store that comes - // back this many blocks behind rolls the state DB and block store back with it, and that - // rollback refuses outright if the state snapshots and WAL cannot span the distance. Size it - // for the burst the writer must absorb, not larger. - // - // It also bounds how stale the EVM RPC head can be. The watermark those queries are served - // against takes the lowest height every store can answer for, this one included, so a receipt - // store behind by a queue's depth holds eth_blockNumber and "latest" that far back. + // Raising it costs more than memory. The queue is not on disk, so an unclean exit loses it and + // the store comes back that far behind, dragging recovery of every other store down with it; + // the EVM RPC head also trails by the queue's depth. Size it for the burst the writer absorbs. // // Set <= 0 for synchronous writes. // defaults to 10 diff --git a/sei-db/config/ss_config.go b/sei-db/config/ss_config.go index f2af85fd2e..c7ab6cae64 100644 --- a/sei-db/config/ss_config.go +++ b/sei-db/config/ss_config.go @@ -38,15 +38,12 @@ type StateStoreConfig struct { // defaults to 100 AsyncWriteBuffer int `mapstructure:"async-write-buffer"` - // DisableInternalWAL stops the backend from keeping a changelog WAL of its own, so a commit is not - // held up by a log write. It is for an owner that already logs every block and replays that log - // into this store: giga's StateDB writes its state WAL before the store and catches the store up - // from it on open, which makes a second log here written and never read. + // DisableInternalWAL stops the backend from keeping a changelog WAL of its own, so a commit is + // not held up by a log write. It is for an owner that already logs every block and replays that + // log into this store, as giga's StateDB does with its state WAL. // - // Like ExternalPruning this is not read from the state-store config but set by the code wiring - // the store into its owner, since it is only correct when that owner supplies the log. Rollback - // through ss/composite replays the changelog and has no other source for the versions above a - // snapshot, so that path must leave it on. + // Like ExternalPruning it is set by the code wiring the store into its owner rather than read + // from app.toml. Rollback through ss/composite replays the changelog, so that path must keep it. DisableInternalWAL bool `mapstructure:"-"` // KeepRecent defines the number of versions to keep in state store (shared by Cosmos and EVM). diff --git a/sei-db/db_engine/pebbledb/mvcc/db.go b/sei-db/db_engine/pebbledb/mvcc/db.go index e0b02ec1ba..47b613126a 100644 --- a/sei-db/db_engine/pebbledb/mvcc/db.go +++ b/sei-db/db_engine/pebbledb/mvcc/db.go @@ -152,7 +152,7 @@ func newPebbleOptions(config config.StateStoreConfig, cache *pebble.Cache) *pebb FormatMajorVersion: pebble.FormatVirtualSSTables, L0CompactionThreshold: 2, L0StopWritesThreshold: 1000, - LBaseMaxBytes: 256 << 20, // 64 MB + LBaseMaxBytes: 256 << 20, // 256 MiB MemTableSize: 256 << 20, MemTableStopWritesThreshold: 4, // Let Pebble run several compactions in parallel so it can keep up with @@ -242,8 +242,8 @@ func OpenDB(dataDir string, config config.StateStoreConfig) (types.StateStore, e _ = db.Close() return nil, errors.New("KeepRecent must be non-negative") } - // An owner that logs every block itself replays that log into this store, so the changelog here - // would be written and never read; DisableInternalWAL drops it and the commit-path write with it. + // An owner that logs every block replays it into this store, leaving the changelog here written + // and never read. if !config.DisableInternalWAL { walKeepRecent := changelogKeepRecent(config) // Snapshot rollback replays the changelog forward from the oldest retained @@ -402,9 +402,8 @@ func (db *Database) Close() error { db.metricsCancel() } - // Draining is owed whether or not a changelog is kept: the queued blocks are only in memory, and - // with no changelog there is nothing to replay them from either. The channel is left in place so - // that a send after close still panics rather than blocking forever on a nil one. + // Owed whether or not a changelog is kept, the queued blocks being only in memory. The channel + // is left in place so a send after close still panics rather than blocking on a nil one. db.drainOnce.Do(func() { // First, stop accepting new pending changes and drain the worker close(db.pendingChanges) diff --git a/sei-db/ledger_db/receipt/litt_receipt_store.go b/sei-db/ledger_db/receipt/litt_receipt_store.go index 4ff3597da2..9155b47e58 100644 --- a/sei-db/ledger_db/receipt/litt_receipt_store.go +++ b/sei-db/ledger_db/receipt/litt_receipt_store.go @@ -63,9 +63,8 @@ import ( // - set: the StorageGarbageCollector prunes through the gc.PrunableStore // implementation in litt_receipt_gc.go, and startPruning stands down. // -// Writes are applied in the background: SetReceipts queues a block and returns, so a receipt is not -// necessarily readable the moment it returns. LatestVersion is the watermark that says how far the -// applied writes have reached, and Close is what waits for the queue to empty. +// Writes are applied in the background, so a receipt is not necessarily readable when SetReceipts +// returns. LatestVersion is the watermark of what has been applied; Close waits for the queue. type littReceiptStore struct { values litt.DB receipts litt.Table @@ -83,19 +82,15 @@ type littReceiptStore struct { backgroundWg sync.WaitGroup closeOnce sync.Once - // Breaks a write into its stages, so which of them a slow write is in is visible. Only the writer - // goroutine records, so one timer serves the store. + // Breaks a write into its stages. Only the writer goroutine records, so one timer serves the store. writePhases *seidbmetrics.PhaseTimer - // Receipt writes waiting to be applied, and the meter reporting what waiting for room on this - // queue costs the caller. A whole write is queued — bodies, log index and version marker — so - // the queue's depth is the depth of the receipt write itself. Its capacity bounds how far the - // store may fall behind the chain; a nil channel means writes are applied on the caller. + // Receipt writes waiting to be applied, and the meter for time spent waiting on a full queue. A + // whole write is queued, so the depth is the receipt write's own. Nil means writes apply inline. writes chan receiptWrite - // Orders admitting a write against shutting the writer down. queueWrite holds it shared for the - // length of a send; Close takes it exclusively to refuse further writes before stopping the - // writer, so no write is ever accepted into a queue that will not be drained. + // Orders admitting a write against shutting the writer down, so none is accepted into a queue + // that will not be drained. queueWrite holds it shared; Close takes it exclusively. admission sync.RWMutex closing bool @@ -342,12 +337,9 @@ func (s *littReceiptStore) belowRetentionFloor(blockNumber uint64) bool { return earliest > 0 && blockNumber < uint64(earliest) //nolint:gosec // earliest is non-negative } -// SetReceipts hands the block's receipts to the writer and returns without waiting for them to be -// applied, blocking only once the queue is full. With AsyncWriteBuffer off, the write is applied -// here instead. -// -// It refuses once a queued write has failed, reporting that failure rather than taking the block: -// the store applies nothing after a failure, so accepting one would drop it silently. +// SetReceipts hands the block's receipts to the writer, blocking only when the queue is full, or +// applies them inline when AsyncWriteBuffer is off. Once a queued write has failed it takes no +// further block and returns that failure. func (s *littReceiptStore) SetReceipts(ctx sdk.Context, receipts []ReceiptRecord) error { if s.writes == nil { return s.applyReceipts(ctx.BlockHeight(), receipts) @@ -361,18 +353,11 @@ func (s *littReceiptStore) SetReceipts(ctx sdk.Context, receipts []ReceiptRecord // ErrStoreClosed is returned by a write the store can no longer apply, the writer having stopped. var ErrStoreClosed = errors.New("receipt store is closed") -// queueWrite hands a write to the writer, waiting for room when the queue is full. -// -// A closed store is refused rather than accepted. The writer drains and exits during Close, so a -// send after that point would sit in a channel nobody reads, and a send once the queue is full -// would never return — inside a commit, which hangs the node instead of failing it. +// queueWrite hands a write to the writer, waiting for room when the queue is full and refusing once +// the store is closing. func (s *littReceiptStore) queueWrite(write receiptWrite) error { - // Held across the send, not merely to read the flag. Close takes the same lock exclusively before - // it stops the writer, so a write admitted here reaches a writer that is still running, and one - // arriving after Close has begun is refused instead of landing in a queue nobody will drain. - // - // A send blocked on a full queue holds the lock and delays Close, which is the intended order: the - // writer is still draining, so the send completes and Close proceeds behind it. + // Held across the send, not merely to read the flag: Close takes it exclusively before stopping + // the writer, so a write admitted here always reaches a writer that is still running. s.admission.RLock() defer s.admission.RUnlock() if s.closing { @@ -502,14 +487,9 @@ func (s *littReceiptStore) FilterLogs(ctx sdk.Context, fromBlock, toBlock uint64 return s.filterLogsByTags(reqCtx, fromBlock, toBlock, crit, budget) } -// startWriter applies queued receipt writes, in the order they were enqueued, until the store -// closes. -// -// It drains what is queued before returning, so a clean shutdown applies the writes it holds. An -// unclean exit does not: everything queued is lost, up to the AsyncWriteBuffer blocks the queue -// holds, and that includes each block's log index and version marker rather than only its bodies. -// The store therefore comes back that far behind, which recovery resolves by rolling every other -// store down to it, so the buffer's size is a recovery cost and not only a memory one. +// startWriter applies queued receipt writes in the order they were enqueued, until the store closes. +// It drains what it holds before returning, so a clean shutdown applies them all and an unclean exit +// loses the queue. func (s *littReceiptStore) startWriter() { s.backgroundWg.Add(1) go func() { @@ -533,12 +513,8 @@ func (s *littReceiptStore) startWriter() { } // applyWrite performs one queued write, keeping the first failure for its callers to collect. -// -// Nothing is applied after a failure. A later block would carry its own version marker, publishing a -// head above one whose receipts were never written: reads of the missing block would report no logs -// and no such transaction, and recovery converges on the published head, so the gap never refills. -// The queue is drained rather than left to fill, so a writer blocked on it is released to see the -// error instead of waiting on a consumer that will never take its block. +// Nothing is applied after a failure: a later block carries its own version marker and would publish +// a head above one whose receipts were never written. func (s *littReceiptStore) applyWrite(write receiptWrite) { if s.writeFailure() != nil { return @@ -549,8 +525,8 @@ func (s *littReceiptStore) applyWrite(write receiptWrite) { } } -// writeFailure returns the first failure a queued write hit. It latches rather than clearing, so -// every later SetReceipts and Close reports it and no single caller can consume it from the rest. +// writeFailure returns the first failure a queued write hit. It latches, so every later caller sees +// it rather than the first to ask consuming it. func (s *littReceiptStore) writeFailure() error { if err := s.writeErr.Load(); err != nil { return *err @@ -582,7 +558,7 @@ func (s *littReceiptStore) startFlusher() { func (s *littReceiptStore) Close() error { var err error s.closeOnce.Do(func() { - // Before the writer is stopped, and exclusive, so it takes effect only once the writes already + // Exclusive and before the writer stops, so it takes effect only once the writes already // admitted have been handed over. s.admission.Lock() s.closing = true diff --git a/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go b/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go index eff942b910..e53242cd06 100644 --- a/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go +++ b/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go @@ -14,9 +14,8 @@ import ( var errIndexCommit = errors.New("injected index commit failure") -// failingIndex is the store's log index with its batch commits made to fail on demand. Committing -// the index is the one step of a receipt write with no other way to fail in a test, and holding the -// commit is what lets a test queue a block behind the one that is failing. +// failingIndex is the store's log index with its batch commits made to fail on demand. Holding a +// commit open is what lets a test queue a block behind the one that is failing. type failingIndex struct { dbtypes.KeyValueDB failing atomic.Bool @@ -43,13 +42,9 @@ func (b *failingBatch) Commit(opts dbtypes.WriteOptions) error { return b.Batch.Commit(opts) } -// TestWriteFailureHoldsTheHeadAgainstAQueuedBlock covers what a failed background write owes the -// blocks already queued behind it. Applying one would commit its own version marker and publish a -// head above the block that never landed: reads of the missing block report no logs and no such -// transaction, and recovery converges on the published head, so the gap never refills. -// -// The block behind the failure is queued while the failing commit is held, which is the ordering -// that makes this reachable — SetReceipts refuses new blocks once the failure is visible. +// TestWriteFailureHoldsTheHeadAgainstAQueuedBlock covers what a failed write owes the blocks queued +// behind it: applying one would publish a head above the block that never landed. The follower is +// queued while the failing commit is held, since SetReceipts refuses blocks once the failure shows. func TestWriteFailureHoldsTheHeadAgainstAQueuedBlock(t *testing.T) { s, closeStore := setupLittCtxStore(t) defer closeStore() @@ -86,8 +81,8 @@ func TestWriteFailureHoldsTheHeadAgainstAQueuedBlock(t *testing.T) { "the head must not move past a block whose receipts were never written") } -// TestWriteFailureLatches covers the failure reaching every later caller rather than the first one -// to ask, which is what lets both a commit and Close act on it. +// TestWriteFailureLatches covers the failure reaching every later caller rather than only the first +// to ask. func TestWriteFailureLatches(t *testing.T) { s, closeStore := setupLittCtxStore(t) defer closeStore() @@ -114,9 +109,8 @@ func TestWriteFailureLatches(t *testing.T) { require.ErrorIs(t, s.Close(), errIndexCommit, "Close must report it too") } -// TestWriteAfterCloseIsRefused covers a commit that races shutdown. The writer has drained and gone -// by then, so a write it accepted would sit in a channel nobody reads, and one arriving on a full -// queue would never return — inside a commit, which hangs the node rather than failing it. +// TestWriteAfterCloseIsRefused covers a commit arriving after shutdown, which the writer is no +// longer there to apply. func TestWriteAfterCloseIsRefused(t *testing.T) { s, _ := setupLittCtxStore(t) require.NoError(t, s.Close()) @@ -127,7 +121,7 @@ func TestWriteAfterCloseIsRefused(t *testing.T) { } // TestWriteAfterCloseIsRefusedWithAFullQueue is the same refusal with no room left to send into, -// which is the case that would otherwise block forever rather than return. +// which would otherwise block forever. func TestWriteAfterCloseIsRefusedWithAFullQueue(t *testing.T) { s, _ := setupLittCtxStore(t) require.NoError(t, s.Close()) @@ -150,10 +144,8 @@ func TestWriteAfterCloseIsRefusedWithAFullQueue(t *testing.T) { } } -// TestWriteRacingCloseIsEitherAppliedOrRefused covers the ordering the two tests above cannot reach, -// where a write is admitted while Close is running rather than after it has returned. Admission and -// shutdown have to be mutually exclusive: a write that returns success must have reached a writer -// that was still running, so the height it reports is one the store actually holds. +// TestWriteRacingCloseIsEitherAppliedOrRefused covers a write admitted while Close is running, +// which the two tests above cannot reach. A write reporting success must have been applied. func TestWriteRacingCloseIsEitherAppliedOrRefused(t *testing.T) { for attempt := range 50 { s, _ := setupLittCtxStore(t) diff --git a/sei-db/ledger_db/receipt/littidx_test.go b/sei-db/ledger_db/receipt/littidx_test.go index 1a4fa3132a..a1891ca7c6 100644 --- a/sei-db/ledger_db/receipt/littidx_test.go +++ b/sei-db/ledger_db/receipt/littidx_test.go @@ -124,9 +124,8 @@ func writeLitBlock(t *testing.T, store receipt.ReceiptStore, ctx sdk.Context, bl if len(records) == 0 { return } - // The write may be applied after SetReceipts returns, and neither signal alone marks the end of - // it: the bodies land before the version marker, and a block written in parts advances the - // marker on its first part. Wait for both. + // Neither signal alone marks the end of a write: the bodies land before the version marker, and + // a block written in parts advances the marker on its first part. Wait for both. last := records[len(records)-1].TxHash require.Eventually(t, func() bool { if store.LatestVersion() < int64(block) { //nolint:gosec // small test heights diff --git a/sei-db/ledger_db/receipt/receipt_store.go b/sei-db/ledger_db/receipt/receipt_store.go index c9f9b3f6d7..2a713c9881 100644 --- a/sei-db/ledger_db/receipt/receipt_store.go +++ b/sei-db/ledger_db/receipt/receipt_store.go @@ -51,9 +51,8 @@ func NewTooManyLogBytesError(maxBytes int64) error { type ReceiptStore interface { controller.PrunableStore - // LatestVersion is the highest block whose receipts are queryable. A write may be applied - // after SetReceipts returns, so this is the watermark a reader follows rather than the - // height it last wrote. + // LatestVersion is the highest block whose receipts are queryable. A write may land after + // SetReceipts returns, so a reader follows this rather than the height it last wrote. LatestVersion() int64 EarliestVersion() int64 GetReceipt(ctx sdk.Context, txHash common.Hash) (*types.Receipt, error) @@ -71,16 +70,15 @@ type ReceiptStore interface { Close() error } -// VersionPinner is implemented by receipt stores whose version markers can be written directly. -// SetReceipts carries those markers, so they are not on ReceiptStore; this is for a caller that has -// put receipts in place by other means and has to state the window they cover. +// VersionPinner is implemented by receipt stores whose version markers can be written directly. It +// is for a caller that put receipts in place by other means and has to state the window they cover. type VersionPinner interface { SetLatestVersion(version int64) error SetEarliestVersion(version int64) error } -// PinVersions widens store's queryable window to [earliest, latest]. It reports a store that does -// not support being pinned rather than leaving the window silently unset. +// PinVersions widens store's queryable window to [earliest, latest], reporting a store that cannot +// be pinned rather than leaving the window unset. func PinVersions(store ReceiptStore, earliest, latest int64) error { pinner, ok := store.(VersionPinner) if !ok { diff --git a/sei-db/state_db/giga/state_db.go b/sei-db/state_db/giga/state_db.go index a714e86e2a..b48d33e091 100644 --- a/sei-db/state_db/giga/state_db.go +++ b/sei-db/state_db/giga/state_db.go @@ -69,13 +69,9 @@ const gigaMeterName = "seidb_giga" // // The returned StateDB owns all three stores and closes them on Close. A failed call closes whatever it // had already opened. -// stateStoreConfigFor is the state store config a StateDB opens SS with, settled once here rather -// than at each open so that every path reaches the same databases. The rollback path opens them too, -// through DiscardStateAbove, and a config differing there would leave it writing a changelog beside -// stores the commit path keeps none for. -// -// The changelog is off because this StateDB already logs every block: its state WAL is written -// before SS and is what catchUpTo replays into it, so a second log inside SS is never read. +// stateStoreConfigFor is the config a StateDB opens SS with. It is settled here rather than at each +// open because the rollback path opens the same databases through DiscardStateAbove. The changelog +// is off: this StateDB's own state WAL is what catchUpTo replays into SS. func stateStoreConfigFor(cfg config.StateStoreConfig) config.StateStoreConfig { cfg.DisableInternalWAL = true return cfg diff --git a/sei-db/state_db/giga/state_db_replay_test.go b/sei-db/state_db/giga/state_db_replay_test.go index 792e162cda..cfd41c9e8c 100644 --- a/sei-db/state_db/giga/state_db_replay_test.go +++ b/sei-db/state_db/giga/state_db_replay_test.go @@ -13,9 +13,8 @@ import ( "github.com/stretchr/testify/require" ) -// SS keeps no changelog of its own under giga: the state WAL written before every commit is what -// catchUpTo replays into it, so a second log would be written on the commit path and never read. -// Recovery rests on that, which is why the absence is pinned rather than left to the config. +// SS keeps no changelog of its own under giga, the state WAL being what catchUpTo replays into it. +// The absence is pinned here rather than left to the config, since recovery rests on it. func TestGigaOpensSSWithoutAChangelog(t *testing.T) { newStateDB := func(t *testing.T) *StateDB { t.Helper() @@ -36,9 +35,9 @@ func TestGigaOpensSSWithoutAChangelog(t *testing.T) { requireNoSSChangelog(t, s.ssCfg.EVMDBDirectory) }) - // The rollback path opens the same databases through DiscardStateAbove rather than openSS, so it - // is the one a config settled per-open would miss. It reaches them via StoredVersions, which - // returns without opening anything when the directory is absent, so the store has to exist first. + // The rollback path opens the same databases through DiscardStateAbove rather than openSS, so a + // config settled per-open would miss it. StoredVersions opens nothing when the directory is + // absent, so the store has to exist first. t.Run("opened to roll back", func(t *testing.T) { s := newStateDB(t) require.NoError(t, s.openSS()) @@ -49,8 +48,8 @@ func TestGigaOpensSSWithoutAChangelog(t *testing.T) { }) } -// TestStateStoreConfigForDisablesTheInternalWAL pins what the constructors apply, since every path -// that opens SS reads the config they settled rather than disabling the log for itself. +// TestStateStoreConfigForDisablesTheInternalWAL pins what the constructors apply, every path that +// opens SS reading the config they settled rather than disabling the log for itself. func TestStateStoreConfigForDisablesTheInternalWAL(t *testing.T) { handedIn := config.DefaultStateStoreConfig() require.False(t, handedIn.DisableInternalWAL, "a caller is not expected to have set it") From 03cd6bbd1ebd87865da413cf813e283b169f30fc Mon Sep 17 00:00:00 2001 From: YimingZang Date: Sun, 13 Sep 2026 16:40:07 -0700 Subject: [PATCH 11/11] Address comment --- sei-db/db_engine/pebbledb/mvcc/db.go | 8 ++++---- sei-db/ledger_db/receipt/littidx_test.go | 14 +++++++++----- sei-db/state_db/giga/state_db.go | 16 ++++++++-------- 3 files changed, 21 insertions(+), 17 deletions(-) diff --git a/sei-db/db_engine/pebbledb/mvcc/db.go b/sei-db/db_engine/pebbledb/mvcc/db.go index 47b613126a..7d0bc44ec5 100644 --- a/sei-db/db_engine/pebbledb/mvcc/db.go +++ b/sei-db/db_engine/pebbledb/mvcc/db.go @@ -59,7 +59,7 @@ const ( // compactor cannot keep up with the tombstone churn that pruning generates, // so deleted data accumulates and slows every subsequent prune scan. Allowing // Pebble to burst up to a few compactions clears that backlog. - maxConcurrentCompactions = 16 + maxConcurrentCompactions = 4 ) var ( @@ -152,12 +152,12 @@ func newPebbleOptions(config config.StateStoreConfig, cache *pebble.Cache) *pebb FormatMajorVersion: pebble.FormatVirtualSSTables, L0CompactionThreshold: 2, L0StopWritesThreshold: 1000, - LBaseMaxBytes: 256 << 20, // 256 MiB - MemTableSize: 256 << 20, + LBaseMaxBytes: 64 << 20, // 64 MiB + MemTableSize: 64 << 20, MemTableStopWritesThreshold: 4, // Let Pebble run several compactions in parallel so it can keep up with // the tombstone churn produced by pruning. See maxConcurrentCompactions. - CompactionConcurrencyRange: func() (int, int) { return 2, maxConcurrentCompactions }, + CompactionConcurrencyRange: func() (int, int) { return 1, maxConcurrentCompactions }, } // Configure L0 with explicit settings diff --git a/sei-db/ledger_db/receipt/littidx_test.go b/sei-db/ledger_db/receipt/littidx_test.go index a1891ca7c6..3d123159fb 100644 --- a/sei-db/ledger_db/receipt/littidx_test.go +++ b/sei-db/ledger_db/receipt/littidx_test.go @@ -2,10 +2,12 @@ package receipt_test import ( "fmt" + "slices" "testing" "time" "github.com/ethereum/go-ethereum/common" + ethtypes "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth/filters" storetypes "github.com/sei-protocol/sei-chain/sei-cosmos/store/types" "github.com/sei-protocol/sei-chain/sei-cosmos/testutil" @@ -124,15 +126,17 @@ func writeLitBlock(t *testing.T, store receipt.ReceiptStore, ctx sdk.Context, bl if len(records) == 0 { return } - // Neither signal alone marks the end of a write: the bodies land before the version marker, and - // a block written in parts advances the marker on its first part. Wait for both. + // A write puts its bodies in litt before it commits its log index, so a readable receipt does not + // mean a queryable one. LatestVersion does not close that gap either: a block written in parts + // does not advance it past the first part. Waiting for the last record's log covers both stages. last := records[len(records)-1].TxHash require.Eventually(t, func() bool { - if store.LatestVersion() < int64(block) { //nolint:gosec // small test heights + //nolint:gosec // small test heights + logs, err := store.FilterLogs(ctx, block, block, filters.FilterCriteria{}, nil) + if err != nil { return false } - _, err := store.GetReceiptFromStore(ctx, last) - return err == nil + return slices.ContainsFunc(logs, func(l *ethtypes.Log) bool { return l.TxHash == last }) }, 5*time.Second, time.Millisecond) } diff --git a/sei-db/state_db/giga/state_db.go b/sei-db/state_db/giga/state_db.go index b48d33e091..036eb4e7d7 100644 --- a/sei-db/state_db/giga/state_db.go +++ b/sei-db/state_db/giga/state_db.go @@ -69,14 +69,6 @@ const gigaMeterName = "seidb_giga" // // The returned StateDB owns all three stores and closes them on Close. A failed call closes whatever it // had already opened. -// stateStoreConfigFor is the config a StateDB opens SS with. It is settled here rather than at each -// open because the rollback path opens the same databases through DiscardStateAbove. The changelog -// is off: this StateDB's own state WAL is what catchUpTo replays into SS. -func stateStoreConfigFor(cfg config.StateStoreConfig) config.StateStoreConfig { - cfg.DisableInternalWAL = true - return cfg -} - func NewStateDB( ctx context.Context, flatkvCfg *flatkvconfig.Config, @@ -116,6 +108,14 @@ func NewStateDB( return s, nil } +// stateStoreConfigFor is the config a StateDB opens SS with. It is settled here rather than at each +// open because the rollback path opens the same databases through DiscardStateAbove. The changelog +// is off: this StateDB's own state WAL is what catchUpTo replays into SS. +func stateStoreConfigFor(cfg config.StateStoreConfig) config.StateStoreConfig { + cfg.DisableInternalWAL = true + return cfg +} + // NewStateDBWithRollback rolls SC, SS and the state WAL back to target and then opens them, so the // returned StateDB commits target+1. It cuts the WAL's tail to target and puts whichever of SC and SS // sits above target on its newest snapshot at or below it, all while the stores are closed, then opens