Make Receipt and SS write fully async

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,12 @@ const MockHeight103 = 103
 const MockHeight101 = 101
 const MockHeight100 = 100
 
+// 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 {
+	return receipt.PinVersions(store, 1, latest)
+}
+
 // 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 +667,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 +1268,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)
 	}
 
 }

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).

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,11 +188,10 @@ func setupTestServer(
 	}
 	pinStateStoreLatestVersion(a, ctxProvider)
 	if store := a.EvmKeeper.ReceiptStore(); store != nil {
-		latest := int64(math.MaxInt64)
-		if err := store.SetLatestVersion(latest); 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)
 		}
-		_ = store.SetEarliestVersion(1)
 	}
 	return TestServer{EVMServer: s, port: port, mockClient: mockClient, app: a, ctxProvider: ctxProvider}
 }

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.

sei-db/bench/gigasim/block_generator.go

@@ -3,13 +3,11 @@ 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"
-	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
@@ -22,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 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.
+	receiptBytes int64
 
 	// The transaction bytes the block store persists. These stand in for encoded transactions, which
 	// the block store holds as opaque bytes.
@@ -77,7 +80,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 +116,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,8 +204,8 @@ func (g *blockGenerator) buildBlock() (*simulatedBlock, error) {
 	}
 	var receipts *receiptBuffer
 	if g.config.EnableReceiptStore {
-		receipts = newReceiptBuffer(count, g.bloomHasher)
-		block.receipts = receipts.receipts
+		receipts = newReceiptBuffer(count, g.receiptCache)
+		block.receiptRecords = receipts.records
 	}
 
 	for i := range count {
@@ -214,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()

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
 		}
 	}

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 {