From 86ba362d3049f2ad513b5f4bd845c820501661db Mon Sep 17 00:00:00 2001 From: Tony Chen Date: Thu, 3 Sep 2026 11:20:52 +0800 Subject: [PATCH 1/2] perf(evmonly): improve Autobahn throughput --- Makefile | 11 +- docker/docker-compose.yml | 28 +++ docker/localnode/scripts/step2_genesis.sh | 12 +- .../scripts/step4_config_override.sh | 34 +++ docker/localnode/scripts/step5_start_sei.sh | 22 ++ giga/evmonly/executor.go | 16 +- giga/evmonly/executor_test.go | 52 +++++ giga/evmonly/giga_store.go | 22 +- giga/evmonly/memory_store.go | 79 +++++-- giga/evmonly/memory_store_test.go | 13 ++ giga/evmonly/parser.go | 26 ++- giga/evmonly/parser_benchmark_test.go | 65 ++++++ giga/evmonly/rpc/server.go | 13 +- giga/evmonly/rpc/server_test.go | 16 +- giga/evmonly/types.go | 4 + integration_test/autobahn/README.md | 23 ++ sei-tendermint/abci/types/application.go | 12 + sei-tendermint/autobahn/types/block.go | 4 +- .../internal/autobahn/autobahn.proto | 4 +- .../internal/autobahn/pb/autobahn.pb.go | 13 +- .../autobahn/pb/autobahn.wireguard.go | 13 +- .../internal/autobahn/producer/mempool.go | 176 ++++++++++----- .../autobahn/producer/mempool_test.go | 57 +++++ .../internal/p2p/evmonly_inmemory_app.go | 206 +++++++++++++----- .../internal/p2p/evmonly_inmemory_app_test.go | 43 ++++ .../internal/p2p/evmonly_prepared_tx_cache.go | 62 ++++++ sei-tendermint/internal/p2p/giga/api.go | 4 +- .../internal/p2p/giga/pb/api.wireguard.go | 7 +- sei-tendermint/internal/p2p/giga_router.go | 1 + .../internal/p2p/giga_router_common.go | 132 ++++++++--- .../internal/p2p/giga_router_fullnode.go | 4 + .../internal/p2p/giga_router_validator.go | 4 + sei-tendermint/internal/proxy/proxy.go | 21 ++ sei-tendermint/internal/rpc/core/mempool.go | 9 + 34 files changed, 1009 insertions(+), 199 deletions(-) create mode 100644 giga/evmonly/parser_benchmark_test.go create mode 100644 sei-tendermint/internal/p2p/evmonly_prepared_tx_cache.go diff --git a/Makefile b/Makefile index fcce17fb4d..4952c3d71a 100644 --- a/Makefile +++ b/Makefile @@ -245,8 +245,8 @@ build-linux: fi .PHONY: build-linux -# Auto-detect platform: use arm64 on ARM Macs, amd64 elsewhere -DOCKER_PLATFORM ?= $(shell if [ "$$(uname -m)" = "arm64" ]; then echo "linux/arm64"; else echo "linux/amd64"; fi) +# Auto-detect the native Docker platform on ARM and x86 hosts. +DOCKER_PLATFORM ?= $(shell if [ "$$(uname -m)" = "arm64" ] || [ "$$(uname -m)" = "aarch64" ]; then echo "linux/arm64"; else echo "linux/amd64"; fi) export DOCKER_PLATFORM # Build docker image for detected platform @@ -404,6 +404,13 @@ CLUSTER_ENV_VARS = DOCKER_PLATFORM=$(DOCKER_PLATFORM) USERID=$(shell id -u) GROU RECEIPT_BACKEND=$(RECEIPT_BACKEND) \ AUTOBAHN=$(AUTOBAHN) \ AUTOBAHN_EVMONLY_IN_MEMORY=$(AUTOBAHN_EVMONLY_IN_MEMORY) \ + AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK=$(AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK) \ + AUTOBAHN_EVMONLY_BLOCK_INTERVAL=$(AUTOBAHN_EVMONLY_BLOCK_INTERVAL) \ + AUTOBAHN_EVMONLY_MAX_GAS=$(AUTOBAHN_EVMONLY_MAX_GAS) \ + AUTOBAHN_EVMONLY_MEMPOOL_SIZE=$(AUTOBAHN_EVMONLY_MEMPOOL_SIZE) \ + AUTOBAHN_EVMONLY_GOGC=$(AUTOBAHN_EVMONLY_GOGC) \ + AUTOBAHN_EVMONLY_GOMAXPROCS=$(AUTOBAHN_EVMONLY_GOMAXPROCS) \ + AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY=$(AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY) \ GIGA_STORAGE=$(GIGA_STORAGE) \ GIGA_MIGRATE_FROM_MEMIAVL=$(GIGA_MIGRATE_FROM_MEMIAVL) \ GIGA_FLATKV_ONLY=$(GIGA_FLATKV_ONLY) diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 247bcb79c5..fe734f5bb7 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -21,6 +21,13 @@ services: - RECEIPT_BACKEND - AUTOBAHN - AUTOBAHN_EVMONLY_IN_MEMORY + - AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK + - AUTOBAHN_EVMONLY_BLOCK_INTERVAL + - AUTOBAHN_EVMONLY_MAX_GAS + - AUTOBAHN_EVMONLY_MEMPOOL_SIZE + - AUTOBAHN_EVMONLY_GOGC + - AUTOBAHN_EVMONLY_GOMAXPROCS + - AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY - GIGA_STORAGE - GIGA_MIGRATE_FROM_MEMIAVL - GIGA_FLATKV_ONLY @@ -57,6 +64,13 @@ services: - RECEIPT_BACKEND - AUTOBAHN - AUTOBAHN_EVMONLY_IN_MEMORY + - AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK + - AUTOBAHN_EVMONLY_BLOCK_INTERVAL + - AUTOBAHN_EVMONLY_MAX_GAS + - AUTOBAHN_EVMONLY_MEMPOOL_SIZE + - AUTOBAHN_EVMONLY_GOGC + - AUTOBAHN_EVMONLY_GOMAXPROCS + - AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY - GIGA_STORAGE - GIGA_MIGRATE_FROM_MEMIAVL - GIGA_FLATKV_ONLY @@ -89,6 +103,13 @@ services: - RECEIPT_BACKEND - AUTOBAHN - AUTOBAHN_EVMONLY_IN_MEMORY + - AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK + - AUTOBAHN_EVMONLY_BLOCK_INTERVAL + - AUTOBAHN_EVMONLY_MAX_GAS + - AUTOBAHN_EVMONLY_MEMPOOL_SIZE + - AUTOBAHN_EVMONLY_GOGC + - AUTOBAHN_EVMONLY_GOMAXPROCS + - AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY - GIGA_STORAGE - GIGA_MIGRATE_FROM_MEMIAVL - GIGA_FLATKV_ONLY @@ -125,6 +146,13 @@ services: - RECEIPT_BACKEND - AUTOBAHN - AUTOBAHN_EVMONLY_IN_MEMORY + - AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK + - AUTOBAHN_EVMONLY_BLOCK_INTERVAL + - AUTOBAHN_EVMONLY_MAX_GAS + - AUTOBAHN_EVMONLY_MEMPOOL_SIZE + - AUTOBAHN_EVMONLY_GOGC + - AUTOBAHN_EVMONLY_GOMAXPROCS + - AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY - GIGA_STORAGE - GIGA_MIGRATE_FROM_MEMIAVL - GIGA_FLATKV_ONLY diff --git a/docker/localnode/scripts/step2_genesis.sh b/docker/localnode/scripts/step2_genesis.sh index 9373ac6745..d23fc1971c 100755 --- a/docker/localnode/scripts/step2_genesis.sh +++ b/docker/localnode/scripts/step2_genesis.sh @@ -2,6 +2,14 @@ # Input parameters NODE_ID=${ID:-0} +AUTOBAHN_EVMONLY_MAX_GAS=${AUTOBAHN_EVMONLY_MAX_GAS:-35000000} + +case "$AUTOBAHN_EVMONLY_MAX_GAS" in + ''|*[!0-9]*|0) + echo "AUTOBAHN_EVMONLY_MAX_GAS must be a positive integer" >&2 + exit 1 + ;; +esac echo "Preparing genesis file" @@ -19,9 +27,9 @@ override_genesis '.app_state["oracle"]["params"]["vote_period"]="2"' override_genesis '.app_state["slashing"]["params"]["signed_blocks_window"]="10000"' override_genesis '.app_state["slashing"]["params"]["min_signed_per_window"]="0.050000000000000000"' override_genesis '.app_state["staking"]["params"]["max_validators"]="50"' -override_genesis '.consensus_params["block"]["max_gas"]="35000000"' +override_genesis ".consensus_params[\"block\"][\"max_gas\"]=\"$AUTOBAHN_EVMONLY_MAX_GAS\"" # Set MaxGasWanted to be 2x of MaxGas, similar to mainnet, in order to avoid false-positive gas related issue reports. -override_genesis '.consensus_params["block"]["max_gas_wanted"]="70000000"' +override_genesis ".consensus_params[\"block\"][\"max_gas_wanted\"]=\"$((AUTOBAHN_EVMONLY_MAX_GAS * 2))\"" override_genesis '.app_state["staking"]["params"]["unbonding_time"]="10s"' # Set a token release schedule for the genesis file diff --git a/docker/localnode/scripts/step4_config_override.sh b/docker/localnode/scripts/step4_config_override.sh index edcee91187..d3bc876029 100755 --- a/docker/localnode/scripts/step4_config_override.sh +++ b/docker/localnode/scripts/step4_config_override.sh @@ -8,6 +8,30 @@ GIGA_EXECUTOR=${GIGA_EXECUTOR:-true} GIGA_OCC=${GIGA_OCC:-true} AUTOBAHN=${AUTOBAHN:-false} AUTOBAHN_EVMONLY_IN_MEMORY=${AUTOBAHN_EVMONLY_IN_MEMORY:-false} +AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK=${AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK:-2000} +AUTOBAHN_EVMONLY_BLOCK_INTERVAL=${AUTOBAHN_EVMONLY_BLOCK_INTERVAL:-400ms} +AUTOBAHN_EVMONLY_MEMPOOL_SIZE=${AUTOBAHN_EVMONLY_MEMPOOL_SIZE:-5000} +AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY=${AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY:-true} + +validate_positive_integer() { + case "$2" in + ''|*[!0-9]*|0) + echo "$1 must be a positive integer" >&2 + exit 1 + ;; + esac +} + +validate_positive_integer AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK "$AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK" +validate_positive_integer AUTOBAHN_EVMONLY_MEMPOOL_SIZE "$AUTOBAHN_EVMONLY_MEMPOOL_SIZE" + +case "$AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY" in + true|false) ;; + *) + echo "AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY must be true or false" >&2 + exit 1 + ;; +esac GIGA_STORAGE=${GIGA_STORAGE:-false} # GIGA_FLATKV_ONLY=true boots the cluster directly in the terminal v3 # steady state: all SC writes route to FlatKV and memiavl is not allocated. @@ -171,7 +195,17 @@ if [ "$AUTOBAHN" = "true" ]; then if [ "$AUTOBAHN_EVMONLY_IN_MEMORY" = "true" ]; then seid tendermint gen-autobahn-config $NODE_DIRS --output "$AUTOBAHN_CONFIG" --persistent-state-dir= + jq \ + --argjson max_txs_per_block "$AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK" \ + --arg block_interval "$AUTOBAHN_EVMONLY_BLOCK_INTERVAL" \ + --argjson enable_evm_proxy "$AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY" \ + '.max_txs_per_block = $max_txs_per_block | + .block_interval = $block_interval | + .enable_evm_proxy = $enable_evm_proxy' \ + "$AUTOBAHN_CONFIG" > "$AUTOBAHN_CONFIG.tmp" + mv "$AUTOBAHN_CONFIG.tmp" "$AUTOBAHN_CONFIG" sed -i 's/^evm-only-in-memory = .*/evm-only-in-memory = true/' ~/.sei/config/config.toml + sed -i "s/^size = .*/size = $AUTOBAHN_EVMONLY_MEMPOOL_SIZE/" ~/.sei/config/config.toml sed -i '/^\[rpc\]/,/^\[/ s|^laddr = .*|laddr = ""|' ~/.sei/config/config.toml sed -i '/^\[api\]/,/^\[/ s/^enable = .*/enable = false/' ~/.sei/config/app.toml sed -i '/^\[grpc\]/,/^\[/ s/^enable = .*/enable = false/' ~/.sei/config/app.toml diff --git a/docker/localnode/scripts/step5_start_sei.sh b/docker/localnode/scripts/step5_start_sei.sh index 57961fd332..34f50f94d1 100755 --- a/docker/localnode/scripts/step5_start_sei.sh +++ b/docker/localnode/scripts/step5_start_sei.sh @@ -4,6 +4,28 @@ NODE_ID=${ID:-0} INVARIANT_CHECK_INTERVAL=${INVARIANT_CHECK_INTERVAL:-0} FREEZE_HEIGHT=${FREEZE_HEIGHT:-0} +if [ "${AUTOBAHN_EVMONLY_IN_MEMORY:-false}" = "true" ]; then + if [ -n "${AUTOBAHN_EVMONLY_GOGC:-}" ]; then + case "$AUTOBAHN_EVMONLY_GOGC" in + off) ;; + *[!0-9]*|0) + echo "AUTOBAHN_EVMONLY_GOGC must be off or a positive integer" >&2 + exit 1 + ;; + esac + export GOGC="$AUTOBAHN_EVMONLY_GOGC" + fi + if [ -n "${AUTOBAHN_EVMONLY_GOMAXPROCS:-}" ]; then + case "$AUTOBAHN_EVMONLY_GOMAXPROCS" in + *[!0-9]*|0) + echo "AUTOBAHN_EVMONLY_GOMAXPROCS must be a positive integer" >&2 + exit 1 + ;; + esac + export GOMAXPROCS="$AUTOBAHN_EVMONLY_GOMAXPROCS" + fi +fi + LOG_DIR="build/generated/logs" mkdir -p $LOG_DIR diff --git a/giga/evmonly/executor.go b/giga/evmonly/executor.go index 91ca00034a..1e1cb1d774 100644 --- a/giga/evmonly/executor.go +++ b/giga/evmonly/executor.go @@ -95,12 +95,26 @@ func (e *Executor) ExecuteBlock(ctx context.Context, req BlockRequest) (*BlockRe } func (e *Executor) PrepareBlock(ctx context.Context, req BlockRequest) (PreparedBlock, error) { + return e.prepareBlock(ctx, req, nil) +} + +// PrepareBlockWithLookup prepares a block while reusing transaction preparation +// results validated under the same chain rules. +func (e *Executor) PrepareBlockWithLookup( + ctx context.Context, + req BlockRequest, + lookup PreparedTxLookup, +) (PreparedBlock, error) { + return e.prepareBlock(ctx, req, lookup) +} + +func (e *Executor) prepareBlock(ctx context.Context, req BlockRequest, lookup PreparedTxLookup) (PreparedBlock, error) { chainConfig := e.chainConfig(req.Context) if err := validateBlockContext(chainConfig, req.Context); err != nil { return PreparedBlock{}, err } signer := ethtypes.MakeSigner(chainConfig, new(big.Int).SetUint64(req.Context.Number), req.Context.Time) - parsed, err := parseBlockTxs(ctx, req.Txs, signer, e.cfg.ParseWorkers) + parsed, err := parseBlockTxs(ctx, req.Txs, signer, e.cfg.ParseWorkers, lookup) if err != nil { return PreparedBlock{}, err } diff --git a/giga/evmonly/executor_test.go b/giga/evmonly/executor_test.go index 7e3c01bef2..b71d4f4331 100644 --- a/giga/evmonly/executor_test.go +++ b/giga/evmonly/executor_test.go @@ -416,6 +416,58 @@ func TestPrepareBlockParallelParsePreservesOrderAndReturnsIndexedError(t *testin require.ErrorContains(t, err, "parse tx 1") } +func TestPrepareBlockWithLookupReusesValidatedTransaction(t *testing.T) { + chainID := big.NewInt(testChainID) + key, err := crypto.GenerateKey() + require.NoError(t, err) + recipient := testAddress(0xe8) + raw := signLegacyTx(t, key, chainID, 0, &recipient, big.NewInt(1), nil) + var tx ethtypes.Transaction + require.NoError(t, tx.UnmarshalBinary(raw)) + cachedSender := testAddress(0xe9) + lookedUp := false + + prepared, err := NewExecutor(Config{ParseWorkers: 4}).PrepareBlockWithLookup( + t.Context(), + BlockRequest{Context: blockContext(chainID), Txs: [][]byte{raw}}, + func(hash common.Hash) (PreparedTx, bool) { + lookedUp = true + require.Equal(t, tx.Hash(), hash) + return PreparedTx{Tx: &tx, Sender: cachedSender}, true + }, + ) + + require.NoError(t, err) + require.True(t, lookedUp) + require.Len(t, prepared.Txs, 1) + require.Same(t, &tx, prepared.Txs[0].Tx) + require.Equal(t, cachedSender, prepared.Txs[0].Sender) +} + +func TestPrepareBlockWithLookupIgnoresMismatchedTransaction(t *testing.T) { + chainID := big.NewInt(testChainID) + key, err := crypto.GenerateKey() + require.NoError(t, err) + wantSender := crypto.PubkeyToAddress(key.PublicKey) + recipient := testAddress(0xeb) + raw := signLegacyTx(t, key, chainID, 0, &recipient, big.NewInt(1), nil) + otherRaw := signLegacyTx(t, key, chainID, 1, &recipient, big.NewInt(1), nil) + otherTx := new(ethtypes.Transaction) + require.NoError(t, otherTx.UnmarshalBinary(otherRaw)) + + prepared, err := NewExecutor(Config{}).PrepareBlockWithLookup( + t.Context(), + BlockRequest{Context: blockContext(chainID), Txs: [][]byte{raw}}, + func(common.Hash) (PreparedTx, bool) { + return PreparedTx{Tx: otherTx, Sender: testAddress(0xec)}, true + }, + ) + + require.NoError(t, err) + require.Len(t, prepared.Txs, 1) + require.Equal(t, wantSender, prepared.Txs[0].Sender) +} + func TestExecutorDynamicFeeTx(t *testing.T) { chainID := big.NewInt(testChainID) key, err := crypto.GenerateKey() diff --git a/giga/evmonly/giga_store.go b/giga/evmonly/giga_store.go index 188e01e32c..a273bb6eaf 100644 --- a/giga/evmonly/giga_store.go +++ b/giga/evmonly/giga_store.go @@ -21,6 +21,10 @@ var ( var _ StateReader = gigaSnapshotStateReader{} +type executorViewStore interface { + OpenExecutorView() gigastore.StateView +} + // NamedChangeSetEncoder converts an executor-native state result into the // on-disk changesets understood by a giga store. It is called synchronously // while the block's read snapshot is still open. It must treat the input as @@ -48,11 +52,16 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar if err := ctx.Err(); err != nil { return nil, err } - snapshot := e.store.OpenView() + snapshot := openExecutorView(e.store) if snapshot == nil { return nil, errors.New("giga store returned a nil snapshot") } - defer snapshot.Close() + snapshotOpen := true + defer func() { + if snapshotOpen { + snapshot.Close() + } + }() result, err := e.executePreparedBlock(ctx, req, gigaSnapshotStateReader{snapshot: snapshot}) if err != nil { @@ -72,6 +81,8 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar if err != nil { return nil, fmt.Errorf("encode state changes for block %d: %w", req.Context.Number, err) } + snapshot.Close() + snapshotOpen = false if err := ctx.Err(); err != nil { return nil, err } @@ -82,6 +93,13 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar return result, nil } +func openExecutorView(store gigastore.StateDB) gigastore.StateView { + if store, ok := store.(executorViewStore); ok { + return store.OpenExecutorView() + } + return store.OpenView() +} + type gigaSnapshotStateReader struct { snapshot gigastore.EVMStateView } diff --git a/giga/evmonly/memory_store.go b/giga/evmonly/memory_store.go index 15ea2e14c5..8300811eb4 100644 --- a/giga/evmonly/memory_store.go +++ b/giga/evmonly/memory_store.go @@ -355,6 +355,17 @@ func (s *MemoryStore) OpenView() gigastore.StateView { return &memoryStoreSnapshot{store: s, height: height} } +// OpenExecutorView returns a current StateView that prevents commits until the +// caller closes it. +func (s *MemoryStore) OpenExecutorView() gigastore.StateView { + s.mu.RLock() + height := int64(0) + if s.hasCurrentHeight { + height = s.currentHeight + } + return &memoryStoreSnapshot{store: s, height: height, storeReadLockHeld: true} +} + func (s *MemoryStore) OpenViewAt(blockNum int64) (gigastore.StateView, bool) { s.mu.RLock() _, ok := s.committedHeights[blockNum] @@ -366,21 +377,22 @@ func (s *MemoryStore) OpenViewAt(blockNum int64) (gigastore.StateView, bool) { } type memoryStoreSnapshot struct { - store *MemoryStore - height int64 - closed atomic.Bool + store *MemoryStore + height int64 + storeReadLockHeld bool + closed atomic.Bool } var _ gigastore.StateView = (*memoryStoreSnapshot)(nil) func (s *memoryStoreSnapshot) AccountExists(address gigastore.Address) bool { s.requireOpen() - s.store.mu.RLock() + locked := s.lockStoreForRead() _, balanceTouched := latestMemoryStoreValue(s.store.balances[address], s.height) _, nonceTouched := latestMemoryStoreValue(s.store.nonces[address], s.height) _, codeTouched := latestMemoryStoreValue(s.store.code[address], s.height) firstStorageTouch, storageTouched := s.store.storageTouch[address] - s.store.mu.RUnlock() + s.unlockStoreAfterRead(locked) if balanceTouched || nonceTouched || codeTouched || storageTouched && firstStorageTouch <= s.height { return true } @@ -394,10 +406,10 @@ func (s *memoryStoreSnapshot) AccountExists(address gigastore.Address) bool { func (s *memoryStoreSnapshot) GetStorage(address gigastore.Address, slot gigastore.Hash) gigastore.Hash { s.requireOpen() key := memoryStoreStorageKey{address: address, slot: slot} - s.store.mu.RLock() + locked := s.lockStoreForRead() value, valueOK := latestMemoryStoreValue(s.store.storage[key], s.height) clearHeight, clearOK := latestHeightAt(s.store.storageClear[address], s.height) - s.store.mu.RUnlock() + s.unlockStoreAfterRead(locked) if valueOK && (!clearOK || value.height >= clearHeight) { if value.delete { return gigastore.Hash{} @@ -412,9 +424,9 @@ func (s *memoryStoreSnapshot) GetStorage(address gigastore.Address, slot gigasto func (s *memoryStoreSnapshot) GetBalance(address gigastore.Address) gigastore.Hash { s.requireOpen() - s.store.mu.RLock() + locked := s.lockStoreForRead() value, ok := latestMemoryStoreValue(s.store.balances[address], s.height) - s.store.mu.RUnlock() + s.unlockStoreAfterRead(locked) if ok { return value.value } @@ -431,9 +443,9 @@ func (s *memoryStoreSnapshot) GetBalance(address gigastore.Address) gigastore.Ha func (s *memoryStoreSnapshot) GetNonce(address gigastore.Address) uint64 { s.requireOpen() - s.store.mu.RLock() + locked := s.lockStoreForRead() value, ok := latestMemoryStoreValue(s.store.nonces[address], s.height) - s.store.mu.RUnlock() + s.unlockStoreAfterRead(locked) if ok { return value.value } @@ -454,9 +466,9 @@ func (s *memoryStoreSnapshot) GetCodeHash(address gigastore.Address) gigastore.H func (s *memoryStoreSnapshot) GetCode(address gigastore.Address) []byte { s.requireOpen() - s.store.mu.RLock() + locked := s.lockStoreForRead() value, ok := latestMemoryStoreValue(s.store.code[address], s.height) - s.store.mu.RUnlock() + s.unlockStoreAfterRead(locked) if ok { if value.delete { return nil @@ -483,9 +495,9 @@ func (s *memoryStoreSnapshot) Get(module string, key []byte) ([]byte, bool) { return nil, false } address := common.Address(key[1:]) - s.store.mu.RLock() + locked := s.lockStoreForRead() value, ok := latestMemoryStoreValue(s.store.balances[address], s.height) - s.store.mu.RUnlock() + s.unlockStoreAfterRead(locked) if !ok { return nil, false } @@ -497,9 +509,9 @@ func (s *memoryStoreSnapshot) Get(module string, key []byte) ([]byte, bool) { return nil, false } address := common.Address(key[1:]) - s.store.mu.RLock() + locked := s.lockStoreForRead() value, ok := latestMemoryStoreValue(s.store.nonces[address], s.height) - s.store.mu.RUnlock() + s.unlockStoreAfterRead(locked) if !ok { return nil, false } @@ -511,9 +523,9 @@ func (s *memoryStoreSnapshot) Get(module string, key []byte) ([]byte, bool) { return nil, false } address := common.Address(key[1:]) - s.store.mu.RLock() + locked := s.lockStoreForRead() value, ok := latestMemoryStoreValue(s.store.code[address], s.height) - s.store.mu.RUnlock() + s.unlockStoreAfterRead(locked) if !ok || value.delete { return nil, false } @@ -523,9 +535,9 @@ func (s *memoryStoreSnapshot) Get(module string, key []byte) ([]byte, bool) { return nil, false } address := common.Address(key[1:]) - s.store.mu.RLock() + locked := s.lockStoreForRead() _, ok := latestHeightAt(s.store.storageClear[address], s.height) - s.store.mu.RUnlock() + s.unlockStoreAfterRead(locked) if !ok { return nil, false } @@ -538,10 +550,10 @@ func (s *memoryStoreSnapshot) Get(module string, key []byte) ([]byte, bool) { address: common.Address(key[1:memoryStoreAccountKeyLen]), slot: common.Hash(key[memoryStoreAccountKeyLen:]), } - s.store.mu.RLock() + locked := s.lockStoreForRead() value, valueOK := latestMemoryStoreValue(s.store.storage[storageKey], s.height) clearHeight, clearOK := latestHeightAt(s.store.storageClear[storageKey.address], s.height) - s.store.mu.RUnlock() + s.unlockStoreAfterRead(locked) if !valueOK || value.delete || clearOK && value.height < clearHeight { return nil, false } @@ -554,7 +566,26 @@ func (s *memoryStoreSnapshot) Get(module string, key []byte) ([]byte, bool) { } func (s *memoryStoreSnapshot) Close() { - s.closed.Store(true) + if !s.closed.CompareAndSwap(false, true) { + return + } + if s.storeReadLockHeld { + s.store.mu.RUnlock() + } +} + +func (s *memoryStoreSnapshot) lockStoreForRead() bool { + if s.storeReadLockHeld { + return false + } + s.store.mu.RLock() + return true +} + +func (s *memoryStoreSnapshot) unlockStoreAfterRead(locked bool) { + if locked { + s.store.mu.RUnlock() + } } func (s *memoryStoreSnapshot) requireOpen() { diff --git a/giga/evmonly/memory_store_test.go b/giga/evmonly/memory_store_test.go index f43c0c7c0e..86ac2906c8 100644 --- a/giga/evmonly/memory_store_test.go +++ b/giga/evmonly/memory_store_test.go @@ -139,6 +139,19 @@ func TestMemoryStoreSnapshotsRemainVersionedAcrossCommits(t *testing.T) { afterDelete.Close() } +func TestMemoryStoreExecutorViewClosesOnce(t *testing.T) { + store := NewMemoryStore(NewMemoryState()) + view := store.OpenExecutorView() + require.Equal(t, int64(0), view.GetBlockHeight()) + + view.Close() + view.Close() + + changesets, err := store.EncodeChangeSet(StateChangeSet{}) + require.NoError(t, err) + require.NoError(t, store.CommitStateChanges(1, changesets)) +} + func TestMemoryStoreRejectsInvalidCommits(t *testing.T) { store := NewMemoryStore(NewMemoryState()) changesets, err := store.EncodeChangeSet(StateChangeSet{}) diff --git a/giga/evmonly/parser.go b/giga/evmonly/parser.go index 2955efbc66..dd58ccb21e 100644 --- a/giga/evmonly/parser.go +++ b/giga/evmonly/parser.go @@ -6,10 +6,17 @@ import ( "github.com/ethereum/go-ethereum/common" ethtypes "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" "golang.org/x/sync/errgroup" ) -func parseBlockTxs(ctx context.Context, txs [][]byte, signer ethtypes.Signer, workers int) ([]PreparedTx, error) { +func parseBlockTxs( + ctx context.Context, + txs [][]byte, + signer ethtypes.Signer, + workers int, + lookup PreparedTxLookup, +) ([]PreparedTx, error) { parsed := make([]PreparedTx, len(txs)) if len(txs) == 0 { return parsed, nil @@ -19,7 +26,7 @@ func parseBlockTxs(ctx context.Context, txs [][]byte, signer ethtypes.Signer, wo if err := ctx.Err(); err != nil { return nil, err } - prepared, err := parsePreparedTx(raw, signer) + prepared, err := lookupOrParsePreparedTx(raw, signer, lookup) if err != nil { return nil, fmt.Errorf("parse tx %d: %w", i, err) } @@ -45,7 +52,7 @@ func parseBlockTxs(ctx context.Context, txs [][]byte, signer ethtypes.Signer, wo for range workers { g.Go(func() error { for i := range jobs { - prepared, err := parsePreparedTx(txs[i], signer) + prepared, err := lookupOrParsePreparedTx(txs[i], signer, lookup) if err != nil { return fmt.Errorf("parse tx %d: %w", i, err) } @@ -60,6 +67,19 @@ func parseBlockTxs(ctx context.Context, txs [][]byte, signer ethtypes.Signer, wo return parsed, nil } +func lookupOrParsePreparedTx(raw []byte, signer ethtypes.Signer, lookup PreparedTxLookup) (PreparedTx, error) { + if lookup != nil { + hash := crypto.Keccak256Hash(raw) + if prepared, ok := lookup(hash); ok && prepared.Tx != nil && prepared.Tx.Hash() == hash { + if err := validateSupportedTx(prepared.Tx); err != nil { + return PreparedTx{}, err + } + return prepared, nil + } + } + return parsePreparedTx(raw, signer) +} + func parsePreparedTx(raw []byte, signer ethtypes.Signer) (PreparedTx, error) { tx, sender, err := parseTx(raw, signer) if err != nil { diff --git a/giga/evmonly/parser_benchmark_test.go b/giga/evmonly/parser_benchmark_test.go new file mode 100644 index 0000000000..f655cac6cc --- /dev/null +++ b/giga/evmonly/parser_benchmark_test.go @@ -0,0 +1,65 @@ +package evmonly + +import ( + "math/big" + "testing" + + "github.com/ethereum/go-ethereum/common" + ethtypes "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/stretchr/testify/require" +) + +func BenchmarkPrepareTransferBlock(b *testing.B) { + const txCount = 5_000 + chainID := big.NewInt(testChainID) + key, err := crypto.GenerateKey() + require.NoError(b, err) + sender := crypto.PubkeyToAddress(key.PublicKey) + recipient := testAddress(0xea) + txs := make([][]byte, txCount) + preparedByHash := make(map[common.Hash]PreparedTx, txCount) + for i := range txCount { + txs[i] = signLegacyTx(b, key, chainID, uint64(i), &recipient, big.NewInt(1), nil) + tx := new(ethtypes.Transaction) + require.NoError(b, tx.UnmarshalBinary(txs[i])) + preparedByHash[tx.Hash()] = PreparedTx{Tx: tx, Sender: sender} + } + executor := NewExecutor(Config{}) + request := BlockRequest{Context: blockContext(chainID), Txs: txs} + + b.Run("recover_sender", func(b *testing.B) { + benchmarkPrepareBlock(b, txCount, func() (PreparedBlock, error) { + return executor.PrepareBlock(b.Context(), request) + }) + }) + b.Run("reuse_check_tx", func(b *testing.B) { + benchmarkPrepareBlock(b, txCount, func() (PreparedBlock, error) { + return executor.PrepareBlockWithLookup( + b.Context(), + request, + func(hash common.Hash) (PreparedTx, bool) { + prepared, ok := preparedByHash[hash] + return prepared, ok + }, + ) + }) + }) +} + +func benchmarkPrepareBlock(b *testing.B, txCount int, prepare func() (PreparedBlock, error)) { + b.Helper() + b.ReportAllocs() + b.ResetTimer() + for range b.N { + prepared, err := prepare() + if err != nil { + b.Fatal(err) + } + if len(prepared.Txs) != txCount { + b.Fatalf("prepared %d transactions, want %d", len(prepared.Txs), txCount) + } + } + b.StopTimer() + b.ReportMetric(float64(b.N*txCount)/b.Elapsed().Seconds(), "tx/s") +} diff --git a/giga/evmonly/rpc/server.go b/giga/evmonly/rpc/server.go index 09a19a7a30..f1e0e318b2 100644 --- a/giga/evmonly/rpc/server.go +++ b/giga/evmonly/rpc/server.go @@ -32,6 +32,7 @@ var logger = seilog.NewLogger("giga", "evmonly", "rpc") // Autobahn shard owner. type Backend interface { BroadcastTx(context.Context, *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTx, error) + EvmProxyEnabled() bool EvmProxy(common.Address) utils.Option[*ethrpc.Client] } @@ -48,12 +49,14 @@ func (api *sendAPI) SendRawTransaction(ctx context.Context, input hexutil.Bytes) } hash := tx.Hash() - if sender, err := ethtypes.Sender(ethtypes.LatestSignerForChainID(tx.ChainId()), tx); err == nil { - if client, ok := api.backend.EvmProxy(sender).Get(); ok { - if err := client.CallContext(ctx, &hash, "eth_sendRawTransaction", input); err != nil { - return hash, err + if api.backend.EvmProxyEnabled() { + if sender, err := ethtypes.Sender(ethtypes.LatestSignerForChainID(tx.ChainId()), tx); err == nil { + if client, ok := api.backend.EvmProxy(sender).Get(); ok { + if err := client.CallContext(ctx, &hash, "eth_sendRawTransaction", input); err != nil { + return hash, err + } + return hash, nil } - return hash, nil } } diff --git a/giga/evmonly/rpc/server_test.go b/giga/evmonly/rpc/server_test.go index 043dfcabed..8f132bf443 100644 --- a/giga/evmonly/rpc/server_test.go +++ b/giga/evmonly/rpc/server_test.go @@ -18,15 +18,22 @@ import ( ) type testBackend struct { - broadcast func(context.Context, *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTx, error) - proxy utils.Option[*ethrpc.Client] + broadcast func(context.Context, *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTx, error) + proxyEnabled bool + proxyChecks int + proxy utils.Option[*ethrpc.Client] } func (b *testBackend) BroadcastTx(ctx context.Context, req *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTx, error) { return b.broadcast(ctx, req) } +func (b *testBackend) EvmProxyEnabled() bool { + return b.proxyEnabled +} + func (b *testBackend) EvmProxy(common.Address) utils.Option[*ethrpc.Client] { + b.proxyChecks++ return b.proxy } @@ -53,6 +60,7 @@ func TestSendRawTransaction(t *testing.T) { require.NoError(t, client.CallContext(t.Context(), &got, "eth_sendRawTransaction", hexutil.Bytes(raw))) require.Equal(t, tx.Hash(), got) require.Equal(t, raw, broadcastRaw) + require.Zero(t, backend.proxyChecks) var chainID hexutil.Big err = client.CallContext(t.Context(), &chainID, "eth_chainId") @@ -107,12 +115,14 @@ func TestProxiesTransactionToShardOwner(t *testing.T) { t.Fatal("proxied transaction reached local broadcaster") return nil, nil }, - proxy: utils.Some(remoteClient), + proxyEnabled: true, + proxy: utils.Some(remoteClient), } got, err := (&sendAPI{backend: backend}).SendRawTransaction(t.Context(), raw) require.NoError(t, err) require.Equal(t, tx.Hash(), got) require.Equal(t, hexutil.Bytes(raw), proxiedRaw) + require.Equal(t, 1, backend.proxyChecks) } type testRemoteSendAPI struct { diff --git a/giga/evmonly/types.go b/giga/evmonly/types.go index b787a20257..1fc0ffba61 100644 --- a/giga/evmonly/types.go +++ b/giga/evmonly/types.go @@ -42,6 +42,10 @@ type BlockRequest struct { Txs [][]byte } +// PreparedTxLookup returns a previously validated transaction preparation by +// the Keccak-256 hash of its raw Ethereum encoding. +type PreparedTxLookup func(common.Hash) (PreparedTx, bool) + // PreparedBlock contains decoded transactions with recovered senders. It is a // trusted executor-produced value: ExecutePreparedBlock assumes callers pass the // result of PrepareBlock unchanged and does not recover senders again. diff --git a/integration_test/autobahn/README.md b/integration_test/autobahn/README.md index 7da72101ae..9d99769d6b 100644 --- a/integration_test/autobahn/README.md +++ b/integration_test/autobahn/README.md @@ -79,6 +79,29 @@ together when using another architecture. `--repo-url` and `--ref` select the source deployed remotely; they default to the current checkout's origin and commit. +The in-memory topology can be tuned through environment variables passed to +the cluster start target: + +```sh +AUTOBAHN=true \ +AUTOBAHN_EVMONLY_IN_MEMORY=true \ +AUTOBAHN_EVMONLY_MAX_TXS_PER_BLOCK=5000 \ +AUTOBAHN_EVMONLY_BLOCK_INTERVAL=100ms \ +AUTOBAHN_EVMONLY_MAX_GAS=120000000 \ +AUTOBAHN_EVMONLY_MEMPOOL_SIZE=100000 \ +AUTOBAHN_EVMONLY_GOGC=200 \ +AUTOBAHN_EVMONLY_GOMAXPROCS=16 \ +AUTOBAHN_EVMONLY_ENABLE_EVM_PROXY=false \ +DOCKER_DETACH=true \ +make docker-cluster-start-skipbuild +``` + +The defaults remain 2,000 transactions per block, a 400 ms block interval, +35,000,000 gas, and 5,000 mempool entries. `GOGC` and `GOMAXPROCS` retain the +Go runtime defaults unless set explicitly. EVM RPC proxying defaults to true; +disable it only for load tests that can submit independent sender streams to +every validator. + If provisioning fails after AWS resources are created, the state is retained with status `failed`. Run `list` to inspect it and `teardown` to remove the instance, security group, and any managed key pair. diff --git a/sei-tendermint/abci/types/application.go b/sei-tendermint/abci/types/application.go index f16dd68c02..f06f2e723c 100644 --- a/sei-tendermint/abci/types/application.go +++ b/sei-tendermint/abci/types/application.go @@ -45,6 +45,18 @@ type Application interface { ApplySnapshotChunk(context.Context, *RequestApplySnapshotChunk) (*ResponseApplySnapshotChunk, error) // Apply a shapshot chunk } +// PreparedBlock is stateless block preparation completed ahead of ordered +// application finalization. +type PreparedBlock interface { + Finalize(context.Context) (*ResponseFinalizeBlock, error) +} + +// BlockPreparingApplication prepares finalized blocks before their ordered +// state transition. +type BlockPreparingApplication interface { + PrepareBlock(context.Context, *RequestFinalizeBlock) (PreparedBlock, error) +} + //------------------------------------------------------- // BaseApplication is a base form of Application diff --git a/sei-tendermint/autobahn/types/block.go b/sei-tendermint/autobahn/types/block.go index d42f172471..186db1f93e 100644 --- a/sei-tendermint/autobahn/types/block.go +++ b/sei-tendermint/autobahn/types/block.go @@ -79,10 +79,10 @@ func (h *BlockHeader) Verify(c *Committee) error { const standardTxBytes uint64 = 1024 // Maximum number of transactions in a block. -const MaxTxsPerBlock uint64 = 2000 +const MaxTxsPerBlock uint64 = 5000 // Maximum total size of all the transactions. -// It can be split arbitrarily across transactions (1 large, 2000 small ones, etc.) +// It can be split arbitrarily across transactions (1 large, 5000 small ones, etc.) // up to MaxTxsPerBlock limit. const MaxTxsBytesPerBlock = MaxTxsPerBlock * standardTxBytes diff --git a/sei-tendermint/internal/autobahn/autobahn.proto b/sei-tendermint/internal/autobahn/autobahn.proto index 23ab14cfdd..9a842a3c8f 100644 --- a/sei-tendermint/internal/autobahn/autobahn.proto +++ b/sei-tendermint/internal/autobahn/autobahn.proto @@ -98,8 +98,8 @@ message Payload { optional uint64 total_gas_wanted = 7; // required optional uint64 total_gas_estimated = 8; // required repeated bytes txs = 6 [ - (wireguard.max_count) = 2000, - (wireguard.max_total_size) = 2048000 + (wireguard.max_count) = 5000, + (wireguard.max_total_size) = 5120000 ]; } diff --git a/sei-tendermint/internal/autobahn/pb/autobahn.pb.go b/sei-tendermint/internal/autobahn/pb/autobahn.pb.go index c9b26b3f5b..823343e1c6 100644 --- a/sei-tendermint/internal/autobahn/pb/autobahn.pb.go +++ b/sei-tendermint/internal/autobahn/pb/autobahn.pb.go @@ -7,13 +7,14 @@ package pb import ( + reflect "reflect" + sync "sync" + unsafe "unsafe" + _ "github.com/sei-protocol/sei-chain/sei-tendermint/internal/hashable/pb" _ "github.com/sei-protocol/sei-chain/sei-tendermint/proto/wireguard" protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" - reflect "reflect" - sync "sync" - unsafe "unsafe" ) const ( @@ -2506,13 +2507,13 @@ const file_autobahn_autobahn_proto_rawDesc = "" + "\b_lane_idB\x0f\n" + "\r_block_numberB\x0e\n" + "\f_parent_hashB\x0f\n" + - "\r_payload_hashJ\x04\b\x01\x10\x02R\x04lane\"\xd5\x02\n" + + "\r_payload_hashJ\x04\b\x01\x10\x02R\x04lane\"\xd6\x02\n" + "\aPayload\x127\n" + "\n" + "created_at\x18\x01 \x01(\v2\x13.autobahn.TimestampH\x00R\tcreatedAt\x88\x01\x01\x12-\n" + "\x10total_gas_wanted\x18\a \x01(\x04H\x01R\x0etotalGasWanted\x88\x01\x01\x123\n" + - "\x13total_gas_estimated\x18\b \x01(\x04H\x02R\x11totalGasEstimated\x88\x01\x01\x12!\n" + - "\x03txs\x18\x06 \x03(\fB\x0fЈ\xe2\xab\f\xd0\x0f\xe0\x88\xe2\xab\f\x80\x80}R\x03txs:\fȈ\xe2\xab\f\x01\xe8\x88\xe2\xab\f\x01B\r\n" + + "\x13total_gas_estimated\x18\b \x01(\x04H\x02R\x11totalGasEstimated\x88\x01\x01\x12\"\n" + + "\x03txs\x18\x06 \x03(\fB\x10Ј\xe2\xab\f\x88'\xe0\x88\xe2\xab\f\x80\xc0\xb8\x02R\x03txs:\fȈ\xe2\xab\f\x01\xe8\x88\xe2\xab\f\x01B\r\n" + "\v_created_atB\x13\n" + "\x11_total_gas_wantedB\x16\n" + "\x14_total_gas_estimatedJ\x04\b\x02\x10\x03J\x04\b\x03\x10\x04J\x04\b\x04\x10\x05J\x04\b\x05\x10\x06R\ttotal_gasR\n" + diff --git a/sei-tendermint/internal/autobahn/pb/autobahn.wireguard.go b/sei-tendermint/internal/autobahn/pb/autobahn.wireguard.go index f64f973969..47ed96d6f9 100644 --- a/sei-tendermint/internal/autobahn/pb/autobahn.wireguard.go +++ b/sei-tendermint/internal/autobahn/pb/autobahn.wireguard.go @@ -2,9 +2,10 @@ package pb import ( + reflect "reflect" + runtime "github.com/sei-protocol/sei-chain/sei-tendermint/internal/protoutils/runtime" utils "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" - reflect "reflect" ) func (*Timestamp) MaxSize() int { @@ -28,11 +29,11 @@ func (*BlockHeader) MaxSize() int { } func (*Payload) MaxSize() int { - return 2056046 + return 5140046 } func (*Block) MaxSize() int { - return 2056181 + return 5140182 } func (*LaneQC) MaxSize() int { @@ -88,7 +89,7 @@ func (*AppProposal) MaxSize() int { } func (*Msg) MaxSize() int { - return 2056185 + return 5140187 } func (*SignedProposal) MaxSize() int { @@ -104,7 +105,7 @@ func (*SignedAppVote) MaxSize() int { } func (*SignedBlock) MaxSize() int { - return 2056289 + return 5140291 } func (*SignedBlockHeader) MaxSize() int { @@ -202,7 +203,7 @@ func init() { 1: {MaxCount: 1, Nested: utils.Some(reflect.TypeFor[*Timestamp]())}, 7: {MaxCount: 1}, 8: {MaxCount: 1}, - 6: {MaxCount: 2000, MaxTotalSize: 2048000}, + 6: {MaxCount: 5000, MaxTotalSize: 5120000}, }) // Register the wireguard.Schema generated for autobahn.Block. diff --git a/sei-tendermint/internal/autobahn/producer/mempool.go b/sei-tendermint/internal/autobahn/producer/mempool.go index 13d4f1e359..ac8265174a 100644 --- a/sei-tendermint/internal/autobahn/producer/mempool.go +++ b/sei-tendermint/internal/autobahn/producer/mempool.go @@ -48,6 +48,11 @@ type mempoolInner struct { evmTxs map[common.Hash]tmtypes.Tx } +type evmNonceSnapshot struct { + first types.BlockNumber + nonce uint64 +} + func newMempoolInner(capacity uint64, lane types.LaneID, n types.BlockNumber) *mempoolInner { return &mempoolInner{ capacity: capacity, @@ -244,63 +249,134 @@ func (s *State) insertTx(ctx context.Context, tx tmtypes.Tx, waitIfFull bool) (* return nil, errTooLarge } - for m, ctrl := range mp.inner.Lock() { - if m.closed { - return nil, ErrNotProducing - } - if m.IsFull() && !waitIfFull { - return nil, errMempoolFull - } - for m.IsFull() { - // mempool is constructed as a FIFO - we do not delay insertions of large txs (going over cap) - // in favor of waiting for smaller txs. This simple algorithm allows us to cap - // pending txs to size of a single block. We can refine this rule later if needed. - // NOTE: in case there are N concurrent InsertTx calls, this condition is reevaluated N times - // every time mempool is updated. Depending on proportion of N to the block size it might get too - // expensive. - if err := ctrl.Wait(ctx); err != nil { - return nil, err + if err := s.insertCheckedTx(ctx, mp, tx, resp, gasWanted, gasEstimated, waitIfFull); err != nil { + return nil, err + } + return resp.ResponseCheckTx, nil +} + +func (s *State) insertCheckedTx( + ctx context.Context, + mp *mempool, + tx tmtypes.Tx, + resp *abci.ResponseCheckTxV2, + gasWanted uint64, + gasEstimated uint64, + waitIfFull bool, +) error { + nonceSnapshot := utils.None[evmNonceSnapshot]() + for { + var nonceReadAt types.BlockNumber + readNonce := false + for m, ctrl := range mp.inner.Lock() { + if err := waitForMempoolCapacity(ctx, m, ctrl, waitIfFull); err != nil { + return err } - if m.closed { - return nil, ErrNotProducing + if resp.IsEVM { + reserved, err := reserveEVMNonce(m, resp, nonceSnapshot) + if err != nil { + return err + } + if !reserved { + nonceReadAt = m.first + readNonce = true + break + } } + s.appendCheckedTx(m, ctrl, tx, resp, gasWanted, gasEstimated) + return nil } - if resp.IsEVM { - addr := resp.EVMSenderAddress - nonce, ok := m.evmNonces[addr] - if !ok { - nonce = s.app.EvmNonce(addr) - } - if nonce != resp.EVMNonce { - return nil, fmt.Errorf("%w: got %v, want %v", errBadNonce, resp.EVMNonce, nonce) - } - m.evmNonces[addr] = nonce + 1 + if !readNonce { + panic("unreachable") } - // If any limit would be exceeded, then construct a payload. - // Note that we use subtraction in a way avoiding arithmetic overflows. - ok := s.cfg.maxTxsPerBlock()-uint64(len(m.nextBlock.txs)) >= 1 - ok = ok && types.MaxTxsBytesPerBlock-m.nextBlock.sizeBytes >= uint64(len(tx)) - ok = ok && s.cfg.MaxGasWantedPerBlock-m.nextBlock.gasWanted >= gasWanted - ok = ok && s.cfg.MaxGasEstimatedPerBlock-m.nextBlock.gasEstimated >= gasEstimated - if !ok { - m.SealBlock() + nonceSnapshot = utils.Some(evmNonceSnapshot{ + first: nonceReadAt, + nonce: s.app.EvmNonce(resp.EVMSenderAddress), + }) + } +} + +func waitForMempoolCapacity( + ctx context.Context, + m *mempoolInner, + ctrl *utils.WatchCtrl, + waitIfFull bool, +) error { + if m.closed { + return ErrNotProducing + } + if m.IsFull() && !waitIfFull { + return errMempoolFull + } + for m.IsFull() { + // mempool is constructed as a FIFO - we do not delay insertions of large txs (going over cap) + // in favor of waiting for smaller txs. This simple algorithm allows us to cap + // pending txs to size of a single block. We can refine this rule later if needed. + // NOTE: in case there are N concurrent InsertTx calls, this condition is reevaluated N times + // every time mempool is updated. Depending on proportion of N to the block size it might get too + // expensive. + if err := ctrl.Wait(ctx); err != nil { + return err } - if len(m.nextBlock.txs) == 0 { - // We notify that we start a new block. - ctrl.Updated() + if m.closed { + return ErrNotProducing } + } + return nil +} - b := m.nextBlock - b.gasEstimated += utils.Clamp[uint64](gasEstimated) - b.gasWanted += utils.Clamp[uint64](resp.GasWanted) - b.sizeBytes += uint64(len(tx)) - b.txs = append(b.txs, tx) - if resp.IsEVM { - addr := resp.EVMSenderAddress - b.evmNonces[addr] = m.evmNonces[addr] - b.evmHashes = append(b.evmHashes, resp.EVMHash) - m.evmTxs[resp.EVMHash] = tx +func reserveEVMNonce( + m *mempoolInner, + resp *abci.ResponseCheckTxV2, + snapshot utils.Option[evmNonceSnapshot], +) (bool, error) { + addr := resp.EVMSenderAddress + nonce, ok := m.evmNonces[addr] + if !ok { + cached, present := snapshot.Get() + if !present || cached.first != m.first { + return false, nil } + nonce = cached.nonce + } + if nonce != resp.EVMNonce { + return false, fmt.Errorf("%w: got %v, want %v", errBadNonce, resp.EVMNonce, nonce) + } + m.evmNonces[addr] = nonce + 1 + return true, nil +} + +func (s *State) appendCheckedTx( + m *mempoolInner, + ctrl *utils.WatchCtrl, + tx tmtypes.Tx, + resp *abci.ResponseCheckTxV2, + gasWanted uint64, + gasEstimated uint64, +) { + // If any limit would be exceeded, then construct a payload. + // Note that we use subtraction in a way avoiding arithmetic overflows. + ok := s.cfg.maxTxsPerBlock()-uint64(len(m.nextBlock.txs)) >= 1 + ok = ok && types.MaxTxsBytesPerBlock-m.nextBlock.sizeBytes >= uint64(len(tx)) + ok = ok && s.cfg.MaxGasWantedPerBlock-m.nextBlock.gasWanted >= gasWanted + ok = ok && s.cfg.MaxGasEstimatedPerBlock-m.nextBlock.gasEstimated >= gasEstimated + if !ok { + m.SealBlock() + } + if len(m.nextBlock.txs) == 0 { + // We notify that we start a new block. + ctrl.Updated() + } + + b := m.nextBlock + b.gasEstimated += gasEstimated + b.gasWanted += utils.Clamp[uint64](resp.GasWanted) + b.sizeBytes += uint64(len(tx)) + b.txs = append(b.txs, tx) + if resp.IsEVM { + addr := resp.EVMSenderAddress + b.evmNonces[addr] = m.evmNonces[addr] + b.evmHashes = append(b.evmHashes, resp.EVMHash) + m.evmTxs[resp.EVMHash] = tx } - return resp.ResponseCheckTx, nil } diff --git a/sei-tendermint/internal/autobahn/producer/mempool_test.go b/sei-tendermint/internal/autobahn/producer/mempool_test.go index 4da150ac16..5334f04dc9 100644 --- a/sei-tendermint/internal/autobahn/producer/mempool_test.go +++ b/sei-tendermint/internal/autobahn/producer/mempool_test.go @@ -94,6 +94,12 @@ type testApp struct { inner utils.Mutex[*testAppInner] } +type blockingNonceApp struct { + *testApp + entered chan common.Address + release chan struct{} +} + func newTestApp() *testApp { return &testApp{ inner: utils.NewMutex(&testAppInner{ @@ -131,6 +137,12 @@ func (a *testApp) EvmNonce(addr common.Address) uint64 { panic("unreachable") } +func (a *blockingNonceApp) EvmNonce(addr common.Address) uint64 { + a.entered <- addr + <-a.release + return a.testApp.EvmNonce(addr) +} + func (a *testApp) CheckTx(_ context.Context, req *abci.RequestCheckTxV2) *abci.ResponseCheckTxV2 { tx, err := decodeTxSpec(req.Tx) if err != nil { @@ -368,6 +380,51 @@ func TestMempool_BadNonce(t *testing.T) { require.NoError(t, err) } +func TestMempool_NonceReadsDoNotHoldAdmissionLock(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + app := newTestApp() + firstAddr, firstNonce := app.NewAccount(rng) + secondAddr, secondNonce := app.NewAccount(rng) + blockingApp := &blockingNonceApp{ + testApp: app, + entered: make(chan common.Address), + release: make(chan struct{}), + } + env := newTestEnv(rng, app.Cfg(), proxy.New(blockingApp)) + env.alignLocalMempool() + firstTx := env.genTx(rng, firstAddr, firstNonce) + secondTx := env.genTx(rng, secondAddr, secondNonce) + + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + defer close(blockingApp.release) + s.SpawnNamed("first insert", func() error { + _, err := env.state.InsertTx(ctx, firstTx.encode()) + return err + }) + got, err := utils.Recv(ctx, blockingApp.entered) + if err != nil { + return err + } + if got != firstAddr { + return fmt.Errorf("first nonce read address = %v, want %v", got, firstAddr) + } + + s.SpawnNamed("second insert", func() error { + _, err := env.state.InsertTx(ctx, secondTx.encode()) + return err + }) + got, err = utils.Recv(ctx, blockingApp.entered) + if err != nil { + return err + } + if got != secondAddr { + return fmt.Errorf("second nonce read address = %v, want %v", got, secondAddr) + } + return nil + })) +} + type blockStats struct { count uint64 sizeBytes uint64 diff --git a/sei-tendermint/internal/p2p/evmonly_inmemory_app.go b/sei-tendermint/internal/p2p/evmonly_inmemory_app.go index 4468103a42..0b13f9ebe4 100644 --- a/sei-tendermint/internal/p2p/evmonly_inmemory_app.go +++ b/sei-tendermint/internal/p2p/evmonly_inmemory_app.go @@ -8,6 +8,7 @@ import ( "math/big" "runtime" "slices" + "sync/atomic" "github.com/ethereum/go-ethereum/common" ethtypes "github.com/ethereum/go-ethereum/core/types" @@ -32,12 +33,13 @@ type evmOnlyInMemoryApplication struct { chainConfig *params.ChainConfig store *evmonly.MemoryStore validators []abci.ValidatorUpdate + preparedTxs *evmOnlyPreparedTxCache + executor utils.AtomicSend[utils.Option[*evmonly.Executor]] + gasLimit atomic.Uint64 state utils.Mutex[*evmOnlyInMemoryState] } type evmOnlyInMemoryState struct { - executor utils.Option[*evmonly.Executor] - gasLimit uint64 nextHeight int64 committedHeight int64 appHash common.Hash @@ -51,7 +53,17 @@ type evmOnlyInMemoryPending struct { blockHash common.Hash } +type evmOnlyInMemoryPreparedBlock struct { + app *evmOnlyInMemoryApplication + height int64 + number uint64 + blockHash common.Hash + block evmonly.PreparedBlock +} + var _ abci.Application = (*evmOnlyInMemoryApplication)(nil) +var _ abci.BlockPreparingApplication = (*evmOnlyInMemoryApplication)(nil) +var _ abci.PreparedBlock = (*evmOnlyInMemoryPreparedBlock)(nil) // NewEVMOnlyInMemoryApplication returns an ephemeral raw-Ethereum application for // Autobahn Docker load tests. @@ -65,6 +77,8 @@ func NewEVMOnlyInMemoryApplication(chainID uint64, validators []abci.ValidatorUp chainConfig: &chainConfig, store: store, validators: slices.Clone(validators), + preparedTxs: newEVMOnlyPreparedTxCache(), + executor: utils.NewAtomicSend(utils.None[*evmonly.Executor]()), state: utils.NewMutex(&evmOnlyInMemoryState{}), } } @@ -78,17 +92,18 @@ func (a *evmOnlyInMemoryApplication) InitChain(req *abci.RequestInitChain) (*abc return nil, err } for state := range a.state.Lock() { - if state.executor.IsPresent() { + if a.executor.Load().IsPresent() { return nil, fmt.Errorf("EVM-only application already initialized") } - state.executor = utils.Some(evmonly.NewExecutor(evmonly.Config{ + executor := evmonly.NewExecutor(evmonly.Config{ ChainConfig: a.chainConfig, MinGasPrice: big.NewInt(evmOnlyInMemoryMinGasPrice), OCCWorkers: runtime.GOMAXPROCS(0), ParseWorkers: runtime.GOMAXPROCS(0), BlockResultPoolSize: 1, - }, evmonly.WithStore(a.store, a.store.EncodeChangeSet))) - state.gasLimit = gasLimit + }, evmonly.WithStore(a.store, a.store.EncodeChangeSet)) + a.gasLimit.Store(gasLimit) + a.executor.Store(utils.Some(executor)) state.nextHeight = req.InitialHeight state.committedHeight = req.InitialHeight - 1 return &abci.ResponseInitChain{}, nil @@ -140,6 +155,8 @@ func (a *evmOnlyInMemoryApplication) CheckTx(_ context.Context, req *abci.Reques if !ok { return &abci.ResponseCheckTxV2{ResponseCheckTx: &abci.ResponseCheckTx{Code: 1, Log: "transaction gas limit exceeds int64"}} } + hash := tx.Hash() + a.preparedTxs.Put(hash, evmonly.PreparedTx{Tx: tx, Sender: sender}) return &abci.ResponseCheckTxV2{ ResponseCheckTx: &abci.ResponseCheckTx{ Code: abci.CodeTypeOK, @@ -148,7 +165,7 @@ func (a *evmOnlyInMemoryApplication) CheckTx(_ context.Context, req *abci.Reques }, IsEVM: true, EVMNonce: tx.Nonce(), - EVMHash: tx.Hash(), + EVMHash: hash, EVMSenderAddress: sender, SeiSenderAddress: append([]byte(nil), sender[:]...), } @@ -198,6 +215,27 @@ func (a *evmOnlyInMemoryApplication) EvmBalance(address common.Address, _ []byte } func (a *evmOnlyInMemoryApplication) FinalizeBlock(ctx context.Context, req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) { + var parentHash common.Hash + for state := range a.state.Lock() { + parentHash = state.parentHash + } + prepared, err := a.prepareBlock(ctx, req, parentHash) + if err != nil { + return nil, err + } + return prepared.Finalize(ctx) +} + +// PrepareBlock decodes transactions and recovers their senders before ordered finalization. +func (a *evmOnlyInMemoryApplication) PrepareBlock(ctx context.Context, req *abci.RequestFinalizeBlock) (abci.PreparedBlock, error) { + return a.prepareBlock(ctx, req, common.BytesToHash(req.Header.LastBlockId.Hash)) +} + +func (a *evmOnlyInMemoryApplication) prepareBlock( + ctx context.Context, + req *abci.RequestFinalizeBlock, + parentHash common.Hash, +) (*evmOnlyInMemoryPreparedBlock, error) { height := req.Header.Height if height <= 0 { return nil, fmt.Errorf("EVM-only block height must be positive: %d", height) @@ -211,40 +249,70 @@ func (a *evmOnlyInMemoryApplication) FinalizeBlock(ctx context.Context, req *abc return nil, fmt.Errorf("EVM-only block timestamp is negative: %s", req.Header.Time) } blockHash := common.BytesToHash(req.Hash) + executor, ok := a.executor.Load().Get() + if !ok { + return nil, fmt.Errorf("EVM-only block prepared before InitChain") + } + prepared, err := executor.PrepareBlockWithLookup(ctx, evmonly.BlockRequest{ + Context: evmonly.BlockContext{ + Number: number, + Time: timestamp, + GasLimit: a.gasLimit.Load(), + ChainID: new(big.Int).Set(a.chainID), + BaseFee: new(big.Int), + BlobBaseFee: new(big.Int), + ParentHash: parentHash, + BlockHash: blockHash, + PrevRandao: crypto.Keccak256Hash(binary.BigEndian.AppendUint64(nil, timestamp)), + }, + Txs: req.Txs, + }, a.preparedTxs.Lookup) + if err != nil { + return nil, err + } + return &evmOnlyInMemoryPreparedBlock{ + app: a, + height: height, + number: number, + blockHash: blockHash, + block: prepared, + }, nil +} + +// Finalize applies the prepared transactions to the application state. +func (b *evmOnlyInMemoryPreparedBlock) Finalize(ctx context.Context) (*abci.ResponseFinalizeBlock, error) { + return b.app.finalizePreparedBlock(ctx, b) +} + +func (a *evmOnlyInMemoryApplication) finalizePreparedBlock( + ctx context.Context, + prepared *evmOnlyInMemoryPreparedBlock, +) (*abci.ResponseFinalizeBlock, error) { for state := range a.state.Lock() { - executor, ok := state.executor.Get() + executor, ok := a.executor.Load().Get() if !ok { return nil, fmt.Errorf("EVM-only block finalized before InitChain") } if state.pending.IsPresent() { - return nil, fmt.Errorf("EVM-only block %d finalized before committing the previous block", height) + return nil, fmt.Errorf("EVM-only block %d finalized before committing the previous block", prepared.height) } - if height != state.nextHeight { - return nil, fmt.Errorf("EVM-only block height %d does not match next height %d", height, state.nextHeight) + if prepared.height != state.nextHeight { + return nil, fmt.Errorf("EVM-only block height %d does not match next height %d", prepared.height, state.nextHeight) } - result, err := executor.ExecuteBlock(ctx, evmonly.BlockRequest{ - Context: evmonly.BlockContext{ - Number: number, - Time: timestamp, - GasLimit: state.gasLimit, - ChainID: new(big.Int).Set(a.chainID), - BaseFee: new(big.Int), - BlobBaseFee: new(big.Int), - ParentHash: state.parentHash, - BlockHash: blockHash, - PrevRandao: crypto.Keccak256Hash(binary.BigEndian.AppendUint64(nil, timestamp)), - }, - Txs: req.Txs, - }) - if err != nil { - return nil, err + if prepared.block.Context.ParentHash != state.parentHash { + return nil, fmt.Errorf("EVM-only block %d parent hash does not match committed parent", prepared.height) } - defer result.Release() - appHash, err := hashEVMOnlyInMemoryResult(state.appHash, number, blockHash, result) + result, err := executor.ExecutePreparedBlock(ctx, prepared.block) if err != nil { return nil, err } - state.pending = utils.Some(evmOnlyInMemoryPending{height: height, appHash: appHash, blockHash: blockHash}) + defer result.Release() + appHash := hashEVMOnlyInMemoryResult(state.appHash, prepared.number, prepared.blockHash, result) + state.pending = utils.Some(evmOnlyInMemoryPending{ + height: prepared.height, + appHash: appHash, + blockHash: prepared.blockHash, + }) return &abci.ResponseFinalizeBlock{ AppHash: append([]byte(nil), appHash[:]...), TxResults: evmOnlyABCIResults(result), @@ -271,40 +339,59 @@ func (a *evmOnlyInMemoryApplication) Commit(context.Context) (*abci.ResponseComm func evmOnlyABCIResults(result *evmonly.BlockResult) []*abci.ExecTxResult { txResults := make([]*abci.ExecTxResult, len(result.Txs)) + values := make([]abci.ExecTxResult, len(result.Txs)) for i, tx := range result.Txs { gasUsed := utils.Clamp[int64](tx.GasUsed) - txResults[i] = &abci.ExecTxResult{ + values[i] = abci.ExecTxResult{ Code: abci.CodeTypeOK, GasWanted: gasUsed, GasUsed: gasUsed, } + txResults[i] = &values[i] } return txResults } -func hashEVMOnlyInMemoryResult(previous common.Hash, height uint64, blockHash common.Hash, result *evmonly.BlockResult) (common.Hash, error) { +func hashEVMOnlyInMemoryResult(previous common.Hash, height uint64, blockHash common.Hash, result *evmonly.BlockResult) common.Hash { h := sha256.New() _, _ = h.Write(previous[:]) - _, _ = h.Write(binary.BigEndian.AppendUint64(nil, height)) + writeEVMOnlyHashUint64(h, height) _, _ = h.Write(blockHash[:]) - _, _ = h.Write(binary.BigEndian.AppendUint64(nil, result.GasUsed)) - changesets, err := evmonly.EncodeMemoryStoreChangeSet(result.ChangeSet) - if err != nil { - return common.Hash{}, err - } - for _, changeset := range changesets { - writeEVMOnlyHashBytes(h, []byte(changeset.Name)) - for _, pair := range changeset.Changeset.Pairs { - writeEVMOnlyHashBytes(h, pair.Key) - if pair.Delete { - _, _ = h.Write([]byte{1}) - } else { - _, _ = h.Write([]byte{0}) - } - writeEVMOnlyHashBytes(h, pair.Value) + writeEVMOnlyHashUint64(h, result.GasUsed) + + changes := result.ChangeSet + writeEVMOnlyHashSection(h, 1, len(changes.Balances)) + for _, change := range changes.Balances { + _, _ = h.Write(change.Address[:]) + var balance [common.HashLength]byte + if change.Balance != nil { + change.Balance.FillBytes(balance[:]) } + _, _ = h.Write(balance[:]) + } + writeEVMOnlyHashSection(h, 2, len(changes.Nonces)) + for _, change := range changes.Nonces { + _, _ = h.Write(change.Address[:]) + writeEVMOnlyHashUint64(h, change.Nonce) } - return common.BytesToHash(h.Sum(nil)), nil + writeEVMOnlyHashSection(h, 3, len(changes.Code)) + for _, change := range changes.Code { + _, _ = h.Write(change.Address[:]) + writeEVMOnlyHashBool(h, change.Delete) + writeEVMOnlyHashBytes(h, change.Code) + } + writeEVMOnlyHashSection(h, 4, len(changes.StorageClears)) + for _, address := range changes.StorageClears { + _, _ = h.Write(address[:]) + } + writeEVMOnlyHashSection(h, 5, len(changes.Storage)) + for _, change := range changes.Storage { + _, _ = h.Write(change.Address[:]) + _, _ = h.Write(change.Key[:]) + writeEVMOnlyHashBool(h, change.Delete) + _, _ = h.Write(change.Value[:]) + } + return common.BytesToHash(h.Sum(nil)) } type byteWriter interface { @@ -312,10 +399,29 @@ type byteWriter interface { } func writeEVMOnlyHashBytes(w byteWriter, value []byte) { - _, _ = w.Write(binary.BigEndian.AppendUint64(nil, uint64(len(value)))) + writeEVMOnlyHashUint64(w, uint64(len(value))) _, _ = w.Write(value) } +func writeEVMOnlyHashSection(w byteWriter, kind byte, count int) { + _, _ = w.Write([]byte{kind}) + writeEVMOnlyHashUint64(w, utils.Clamp[uint64](count)) +} + +func writeEVMOnlyHashBool(w byteWriter, value bool) { + if value { + _, _ = w.Write([]byte{1}) + return + } + _, _ = w.Write([]byte{0}) +} + +func writeEVMOnlyHashUint64(w byteWriter, value uint64) { + var encoded [8]byte + binary.BigEndian.PutUint64(encoded[:], value) + _, _ = w.Write(encoded[:]) +} + type evmOnlyFundedState struct{} func (evmOnlyFundedState) AccountExists(common.Address) bool { return true } diff --git a/sei-tendermint/internal/p2p/evmonly_inmemory_app_test.go b/sei-tendermint/internal/p2p/evmonly_inmemory_app_test.go index 81fbcfe32e..4161d90977 100644 --- a/sei-tendermint/internal/p2p/evmonly_inmemory_app_test.go +++ b/sei-tendermint/internal/p2p/evmonly_inmemory_app_test.go @@ -106,6 +106,49 @@ func TestEVMOnlyInMemoryApplicationProducesDeterministicRoot(t *testing.T) { require.Equal(t, firstResponse.AppHash, secondResponse.AppHash) } +func TestEVMOnlyInMemoryApplicationPreparesNextBlockBeforeFinalization(t *testing.T) { + app := newInitializedEVMOnlyTestApp(t) + preparer, ok := app.(abci.BlockPreparingApplication) + require.True(t, ok) + firstRaw, firstSender := signedEVMOnlyTestTx(t, evmOnlyTestChainID, 0) + secondRaw, secondSender := signedEVMOnlyTestTx(t, evmOnlyTestChainID, 0) + firstHash := crypto.Keccak256([]byte("prepared-block-1")) + secondHash := crypto.Keccak256([]byte("prepared-block-2")) + + first, err := preparer.PrepareBlock(t.Context(), &abci.RequestFinalizeBlock{ + Txs: [][]byte{firstRaw}, + Hash: firstHash, + Header: &tmproto.Header{ + Height: 1, + Time: time.Unix(1_700_000_001, 0), + }, + }) + require.NoError(t, err) + second, err := preparer.PrepareBlock(t.Context(), &abci.RequestFinalizeBlock{ + Txs: [][]byte{secondRaw}, + Hash: secondHash, + Header: &tmproto.Header{ + Height: 2, + Time: time.Unix(1_700_000_002, 0), + LastBlockId: tmproto.BlockID{Hash: firstHash}, + }, + }) + require.NoError(t, err) + + _, err = first.Finalize(t.Context()) + require.NoError(t, err) + _, err = app.Commit(t.Context()) + require.NoError(t, err) + _, err = second.Finalize(t.Context()) + require.NoError(t, err) + _, err = app.Commit(t.Context()) + require.NoError(t, err) + + require.Equal(t, int64(2), app.LastBlockHeight()) + require.Equal(t, uint64(1), app.EvmNonce(firstSender)) + require.Equal(t, uint64(1), app.EvmNonce(secondSender)) +} + func TestEVMOnlyInMemoryApplicationRequiresInitChain(t *testing.T) { app := NewEVMOnlyInMemoryApplication(evmOnlyTestChainID, nil) diff --git a/sei-tendermint/internal/p2p/evmonly_prepared_tx_cache.go b/sei-tendermint/internal/p2p/evmonly_prepared_tx_cache.go new file mode 100644 index 0000000000..56c4a5ddc1 --- /dev/null +++ b/sei-tendermint/internal/p2p/evmonly_prepared_tx_cache.go @@ -0,0 +1,62 @@ +package p2p + +import ( + "github.com/ethereum/go-ethereum/common" + + "github.com/sei-protocol/sei-chain/giga/evmonly" + autobahntypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" +) + +const evmOnlyPreparedTxCacheShards = 64 +const evmOnlyPreparedTxCacheCapacity = 3 * autobahntypes.MaxLaneRangeInProposal * autobahntypes.MaxTxsPerBlock + +type evmOnlyPreparedTxCache struct { + shards [evmOnlyPreparedTxCacheShards]utils.RWMutex[*evmOnlyPreparedTxCacheShard] +} + +type evmOnlyPreparedTxCacheShard struct { + entries map[common.Hash]evmonly.PreparedTx + order []common.Hash + next int +} + +func newEVMOnlyPreparedTxCache() *evmOnlyPreparedTxCache { + cache := &evmOnlyPreparedTxCache{} + shardCapacity := (int(evmOnlyPreparedTxCacheCapacity) + evmOnlyPreparedTxCacheShards - 1) / + evmOnlyPreparedTxCacheShards + for i := range cache.shards { + cache.shards[i] = utils.NewRWMutex(&evmOnlyPreparedTxCacheShard{ + entries: make(map[common.Hash]evmonly.PreparedTx, shardCapacity), + order: make([]common.Hash, 0, shardCapacity), + }) + } + return cache +} + +func (c *evmOnlyPreparedTxCache) Put(hash common.Hash, prepared evmonly.PreparedTx) { + shard := &c.shards[int(hash[0])%len(c.shards)] + for entries := range shard.Lock() { + if _, ok := entries.entries[hash]; ok { + entries.entries[hash] = prepared + return + } + if len(entries.order) < cap(entries.order) { + entries.order = append(entries.order, hash) + } else { + delete(entries.entries, entries.order[entries.next]) + entries.order[entries.next] = hash + entries.next = (entries.next + 1) % cap(entries.order) + } + entries.entries[hash] = prepared + } +} + +func (c *evmOnlyPreparedTxCache) Lookup(hash common.Hash) (evmonly.PreparedTx, bool) { + shard := &c.shards[int(hash[0])%len(c.shards)] + for entries := range shard.RLock() { + prepared, ok := entries.entries[hash] + return prepared, ok + } + panic("unreachable") +} diff --git a/sei-tendermint/internal/p2p/giga/api.go b/sei-tendermint/internal/p2p/giga/api.go index 4e8f54eec3..7ba1452d1e 100644 --- a/sei-tendermint/internal/p2p/giga/api.go +++ b/sei-tendermint/internal/p2p/giga/api.go @@ -19,7 +19,7 @@ var Ping = rpc.Register[API](0, "ping", var StreamLaneProposals = rpc.Register[API](1, "stream_lane_proposals", rpc.Limit{Rate: 1, Concurrent: 1}, rpc.Msg[*pb.StreamLaneProposalsReq]{MsgSize: kB, Window: 1}, - rpc.Msg[*pb.LaneProposal]{MsgSize: 2 * MB, Window: 5}, + rpc.Msg[*pb.LaneProposal]{MsgSize: 6 * MB, Window: 5}, ) var StreamLaneVotes = rpc.Register[API](2, "stream_lane_votes", rpc.Limit{Rate: 1, Concurrent: 1}, @@ -56,5 +56,5 @@ var StreamAppQCs = rpc.Register[API](5, "stream_app_qcs", var GetBlock = rpc.Register[API](8, "get_block", rpc.Limit{Rate: 10, Concurrent: 10}, rpc.Msg[*pb.GetBlockReq]{MsgSize: 10 * kB, Window: 1}, - rpc.Msg[*pb.GetBlockResp]{MsgSize: 2 * MB, Window: 1}, + rpc.Msg[*pb.GetBlockResp]{MsgSize: 6 * MB, Window: 1}, ) diff --git a/sei-tendermint/internal/p2p/giga/pb/api.wireguard.go b/sei-tendermint/internal/p2p/giga/pb/api.wireguard.go index 36c5e0a985..9291d72eb9 100644 --- a/sei-tendermint/internal/p2p/giga/pb/api.wireguard.go +++ b/sei-tendermint/internal/p2p/giga/pb/api.wireguard.go @@ -2,10 +2,11 @@ package pb import ( + reflect "reflect" + pb "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/pb" runtime "github.com/sei-protocol/sei-chain/sei-tendermint/internal/protoutils/runtime" utils "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" - reflect "reflect" ) func (*ConsensusResp) MaxSize() int { @@ -25,7 +26,7 @@ func (*LaneVote) MaxSize() int { } func (*LaneProposal) MaxSize() int { - return 2056293 + return 5140296 } func (*AppVote) MaxSize() int { @@ -53,7 +54,7 @@ func (*GetBlockReq) MaxSize() int { } func (*GetBlockResp) MaxSize() int { - return 2056185 + return 5140187 } func (*StreamFullCommitQCsReq) MaxSize() int { diff --git a/sei-tendermint/internal/p2p/giga_router.go b/sei-tendermint/internal/p2p/giga_router.go index 5a156dbc42..22e691af81 100644 --- a/sei-tendermint/internal/p2p/giga_router.go +++ b/sei-tendermint/internal/p2p/giga_router.go @@ -73,6 +73,7 @@ type GigaRouter interface { MaxGasEstimatedPerBlock() uint64 BlockByNumber(ctx context.Context, n atypes.GlobalBlockNumber) (*coretypes.ResultBlock, error) BlockByHash(ctx context.Context, hash atypes.BlockHeaderHash) (*coretypes.ResultBlock, error) + EvmProxyEnabled() bool EvmProxy(sender common.Address) utils.Option[*rpc.Client] Mempool() utils.Option[*producer.State] } diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go index bc50d1b714..fb524c2f68 100644 --- a/sei-tendermint/internal/p2p/giga_router_common.go +++ b/sei-tendermint/internal/p2p/giga_router_common.go @@ -52,6 +52,12 @@ type gigaRouterCommon struct { inboundFullnodeCap int64 } +type preparedGlobalBlock struct { + block *atypes.GlobalBlock + request *abci.RequestFinalizeBlock + prepared utils.Option[abci.PreparedBlock] +} + // BuildDataState validates the common config, constructs the committee, and // returns an initialised data.State backed by blockStore. // @@ -195,9 +201,12 @@ func (r *gigaRouterCommon) translateGlobalBlock(gb *atypes.GlobalBlock) *coretyp } } -func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlock, hashVault hashvault.HashVault) (*abci.ResponseCommit, error) { +func (r *gigaRouterCommon) prepareBlock( + ctx context.Context, + b *atypes.GlobalBlock, + parentHash atypes.BlockHeaderHash, +) (preparedGlobalBlock, error) { app := r.app - hash := b.Header.Hash() var proposerAddress types.Address if vals := app.GetValidators(); len(vals) > 0 { // Deterministically select a proposer from the app's validator committee. @@ -205,13 +214,12 @@ func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlo proposer := slices.MinFunc(vals, func(a, b abci.ValidatorUpdate) int { return a.PubKey.Compare(b.PubKey) }) key, err := crypto.PubKeyFromProto(proposer.PubKey) if err != nil { - return nil, fmt.Errorf("crypto.PubKeyFromProto(): %w", err) + return preparedGlobalBlock{}, fmt.Errorf("crypto.PubKeyFromProto(): %w", err) } proposerAddress = key.Address() } - - // TODO: add metrics to understand execution latency. - resp, err := app.FinalizeBlock(ctx, &abci.RequestFinalizeBlock{ + hash := b.Header.Hash() + request := &abci.RequestFinalizeBlock{ Txs: b.Payload.Txs(), // Empty DecidedLastCommit does not indicate missing votes. DecidedLastCommit: abci.CommitInfo{}, @@ -220,14 +228,58 @@ func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlo // and is fed as block hash to EVM contracts. Hash: hash[:], Header: (&types.Header{ - ChainID: r.cfg.GenDoc.ChainID, - Height: int64(b.GlobalNumber), // nolint:gosec // different representations of the same value - Time: b.Timestamp, + ChainID: r.cfg.GenDoc.ChainID, + LastBlockID: types.BlockID{Hash: tmbytes.HexBytes(parentHash[:])}, + Height: int64(b.GlobalNumber), // nolint:gosec // different representations of the same value + Time: b.Timestamp, // WARNING: the reward distribution has corner cases where it forgets the proposer, // because reward is distributed with a delay. This is not our problem here though. ProposerAddress: proposerAddress, }).ToProto(), - }) + } + prepared, err := app.PrepareBlock(ctx, request) + if err != nil { + return preparedGlobalBlock{}, fmt.Errorf("app.PrepareBlock(): %w", err) + } + return preparedGlobalBlock{block: b, request: request, prepared: prepared}, nil +} + +func (r *gigaRouterCommon) prepareBlocks( + ctx context.Context, + next atypes.GlobalBlockNumber, + parentHash atypes.BlockHeaderHash, + out chan<- preparedGlobalBlock, +) error { + for n := next; ; n += 1 { + block, err := r.data.GlobalBlock(ctx, n) + if err != nil { + return fmt.Errorf("r.data.GlobalBlock(%v): %w", n, err) + } + prepared, err := r.prepareBlock(ctx, block, parentHash) + if err != nil { + return fmt.Errorf("r.prepareBlock(%v): %w", n, err) + } + if err := utils.Send(ctx, out, prepared); err != nil { + return err + } + parentHash = block.Header.Hash() + } +} + +func (r *gigaRouterCommon) executeBlock( + ctx context.Context, + prepared preparedGlobalBlock, + hashVault hashvault.HashVault, +) (*abci.ResponseCommit, error) { + app := r.app + b := prepared.block + var resp *abci.ResponseFinalizeBlock + var err error + if block, ok := prepared.prepared.Get(); ok { + resp, err = app.FinalizePreparedBlock(ctx, block) + } else { + resp, err = app.FinalizeBlock(ctx, prepared.request) + } if err != nil { return nil, fmt.Errorf("app.FinalizeBlock(): %w", err) } @@ -387,6 +439,7 @@ func (r *gigaRouterCommon) runExecute(ctx context.Context) error { return fmt.Errorf("invalid info.LastBlockHeight = %v", info.LastBlockHeight) } next := last + 1 + var parentHash atypes.BlockHeaderHash if last == 0 { // Fresh start: CometBFT handshaker is skipped in giga mode (see // node.go: shouldHandshake = !stateSync && !gigaEnabled), so we @@ -442,36 +495,43 @@ func (r *gigaRouterCommon) runExecute(ctx context.Context) error { if err := r.data.PushAppHash(ctx, last, info.LastBlockAppHash, weights); err != nil { return fmt.Errorf("r.data.PushAppHash(): %w", err) } + parentHash = b.Header.Hash() } - for n := next; ; n += 1 { - b, err := r.data.GlobalBlock(ctx, n) - if err != nil { - return fmt.Errorf("r.data.GlobalBlock(%v): %w", n, err) - } - commitResp, err := r.executeBlock(ctx, b, hashVault) - if err != nil { - return fmt.Errorf("r.executeBlock(%v): %w", n, err) - } - pruneBefore, ok := utils.SafeCast[atypes.GlobalBlockNumber](commitResp.RetainHeight) - if !ok { - return fmt.Errorf("invalid commitResp.RetainHeight = %v", commitResp.RetainHeight) - } - if err := r.data.PruneBefore(pruneBefore); err != nil { - return fmt.Errorf("r.data.PruneBefore(%v): %w", pruneBefore, err) - } - // Align the vault's retention with the data layer's prune boundary. - if err := hashVault.Prune(ctx, uint64(pruneBefore)); err != nil { - // A canceled context just means we're shutting down between a successful executeBlock - // and this prune; that's benign, not a prune failure, so don't alarm operators. - if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { - logger.Info("hashvault prune aborted by context cancellation during shutdown", - "prune_before", pruneBefore, "err", err) - } else { - logger.Error("failed to prune hashvault", "prune_before", pruneBefore, "err", err) + return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + preparedBlocks := make(chan preparedGlobalBlock, 1) + s.SpawnNamed("prepareBlocks", func() error { + return r.prepareBlocks(ctx, next, parentHash, preparedBlocks) + }) + for n := next; ; n += 1 { + prepared, err := utils.Recv(ctx, preparedBlocks) + if err != nil { + return err + } + commitResp, err := r.executeBlock(ctx, prepared, hashVault) + if err != nil { + return fmt.Errorf("r.executeBlock(%v): %w", n, err) + } + pruneBefore, ok := utils.SafeCast[atypes.GlobalBlockNumber](commitResp.RetainHeight) + if !ok { + return fmt.Errorf("invalid commitResp.RetainHeight = %v", commitResp.RetainHeight) + } + if err := r.data.PruneBefore(pruneBefore); err != nil { + return fmt.Errorf("r.data.PruneBefore(%v): %w", pruneBefore, err) + } + // Align the vault's retention with the data layer's prune boundary. + if err := hashVault.Prune(ctx, uint64(pruneBefore)); err != nil { + // A canceled context just means we're shutting down between a successful executeBlock + // and this prune; that's benign, not a prune failure, so don't alarm operators. + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + logger.Info("hashvault prune aborted by context cancellation during shutdown", + "prune_before", pruneBefore, "err", err) + } else { + logger.Error("failed to prune hashvault", "prune_before", pruneBefore, "err", err) + } } } - } + }) } // dialAndRunConn dials a peer, handshakes as a SeiGiga connection, diff --git a/sei-tendermint/internal/p2p/giga_router_fullnode.go b/sei-tendermint/internal/p2p/giga_router_fullnode.go index cdc7afd9e5..79d8183308 100644 --- a/sei-tendermint/internal/p2p/giga_router_fullnode.go +++ b/sei-tendermint/internal/p2p/giga_router_fullnode.go @@ -63,6 +63,10 @@ func (r *gigaFullnodeRouter) Run(ctx context.Context) error { }) } +func (r *gigaFullnodeRouter) EvmProxyEnabled() bool { + return true +} + // EvmProxy on the fullnode always returns the shard owner's EVM RPC client. // EnableEvmProxy is a no-op here because fullnodes do not have a local mempool. func (r *gigaFullnodeRouter) EvmProxy(sender common.Address) utils.Option[*ethrpc.Client] { diff --git a/sei-tendermint/internal/p2p/giga_router_validator.go b/sei-tendermint/internal/p2p/giga_router_validator.go index 3476fbcf1b..b945372a02 100644 --- a/sei-tendermint/internal/p2p/giga_router_validator.go +++ b/sei-tendermint/internal/p2p/giga_router_validator.go @@ -98,6 +98,10 @@ func (r *gigaValidatorRouter) Run(ctx context.Context) error { }) } +func (r *gigaValidatorRouter) EvmProxyEnabled() bool { + return r.cfg.EnableEvmProxy +} + // EvmProxy on the validator returns None when the sender's shard owner is // us (handle locally via mempool). For remote // shards, we proxy only while the target validator is currently connected; diff --git a/sei-tendermint/internal/proxy/proxy.go b/sei-tendermint/internal/proxy/proxy.go index 8f435fc26f..2d71bb12f8 100644 --- a/sei-tendermint/internal/proxy/proxy.go +++ b/sei-tendermint/internal/proxy/proxy.go @@ -11,6 +11,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" tmproto "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/types" ) @@ -39,6 +40,26 @@ func (app *Proxy) FinalizeBlock(ctx context.Context, req *types.RequestFinalizeB return app.app.FinalizeBlock(ctx, req) } +// PrepareBlock prepares req when the application supports pipelined block preparation. +func (app *Proxy) PrepareBlock(ctx context.Context, req *types.RequestFinalizeBlock) (utils.Option[types.PreparedBlock], error) { + preparer, ok := app.app.(types.BlockPreparingApplication) + if !ok { + return utils.None[types.PreparedBlock](), nil + } + defer addTimeSample(Global.MethodTimingAt("prepare_block", "sync"))() + prepared, err := preparer.PrepareBlock(ctx, req) + if err != nil { + return utils.None[types.PreparedBlock](), err + } + return utils.Some(prepared), nil +} + +// FinalizePreparedBlock applies a previously prepared block. +func (app *Proxy) FinalizePreparedBlock(ctx context.Context, prepared types.PreparedBlock) (*types.ResponseFinalizeBlock, error) { + defer addTimeSample(Global.MethodTimingAt("finalize_block", "sync"))() + return prepared.Finalize(ctx) +} + func (app *Proxy) GetTxPriorityHint(ctx context.Context, req *types.RequestGetTxPriorityHintV2) (*types.ResponseGetTxPriorityHint, error) { defer addTimeSample(Global.MethodTimingAt("get_tx_priority", "sync"))() return app.app.GetTxPriorityHint(ctx, req) diff --git a/sei-tendermint/internal/rpc/core/mempool.go b/sei-tendermint/internal/rpc/core/mempool.go index eed02aaea0..c59387b99a 100644 --- a/sei-tendermint/internal/rpc/core/mempool.go +++ b/sei-tendermint/internal/rpc/core/mempool.go @@ -16,6 +16,15 @@ import ( "github.com/sei-protocol/sei-chain/sei-tendermint/types" ) +// EvmProxyEnabled reports whether EVM transactions require sender-based RPC +// routing. +func (env *Environment) EvmProxyEnabled() bool { + if r, ok := env.gigaRouter().Get(); ok { + return r.EvmProxyEnabled() + } + return false +} + // EvmProxy returns the EVM RPC client of the autobahn validator that owns the // sender shard, or None if the sender maps to the local validator (handle // locally) or autobahn isn't configured. From cca7589ab89f42b4edcc7ca19eab5eba037df50f Mon Sep 17 00:00:00 2001 From: Tony Chen Date: Thu, 3 Sep 2026 13:18:00 +0800 Subject: [PATCH 2/2] perf(evmonly): fast-path conflict-free OCC merge --- giga/evmonly/executor_test.go | 2 + giga/evmonly/occ.go | 131 +++++++++++++++++++++++++++++++++- 2 files changed, 132 insertions(+), 1 deletion(-) diff --git a/giga/evmonly/executor_test.go b/giga/evmonly/executor_test.go index b71d4f4331..0ac09e05e4 100644 --- a/giga/evmonly/executor_test.go +++ b/giga/evmonly/executor_test.go @@ -644,6 +644,7 @@ func TestExecutorOCCNonConflictingTransfersMatchSequential(t *testing.T) { require.True(t, occResult.OCCStats.Attempted) require.False(t, occResult.OCCStats.Fallback) require.Zero(t, occResult.OCCStats.ConflictCount) + require.Equal(t, seqResult.ChangeSet, occResult.ChangeSet) for i := range txCount { require.Equal(t, seqResult.Txs[i].Hash, occResult.Txs[i].Hash) require.Equal(t, seqResult.Txs[i].Status, occResult.Txs[i].Status) @@ -734,6 +735,7 @@ func TestExecutorOCCFeePayingTransfersDoNotConflictOnCoinbase(t *testing.T) { require.True(t, occResult.OCCStats.Attempted) require.False(t, occResult.OCCStats.Fallback) require.Equal(t, seqResult.GasUsed, occResult.GasUsed) + require.Equal(t, seqResult.ChangeSet, occResult.ChangeSet) seqState.ApplyChangeSet(seqResult.ChangeSet) occState.ApplyChangeSet(occResult.ChangeSet) diff --git a/giga/evmonly/occ.go b/giga/evmonly/occ.go index 4bc6006f7c..5bc4824c22 100644 --- a/giga/evmonly/occ.go +++ b/giga/evmonly/occ.go @@ -69,6 +69,18 @@ func (e *Executor) executeBlockOCC(ctx context.Context, req PreparedBlock, sourc } return nil, err } + validation, conflictFree, err := validateInitialOCCResults(ctx, runner, results) + if err != nil { + return nil, err + } + if conflictFree { + result, err := e.mergeConflictFreeOCCResults(ctx, results, source) + if err != nil { + return nil, err + } + result.OCCStats = validation.stats(false) + return result, nil + } results, finalState, validation, err := e.validateBlockSTM(ctx, runner, executionPool, source, results) if errors.Is(err, errOCCMaxIncarnation) || errors.Is(err, errOCCWorkerPoolClosed) { @@ -92,6 +104,48 @@ func (e *Executor) executeBlockOCC(ctx context.Context, req PreparedBlock, sourc return result, nil } +// validateInitialOCCResults reports whether all initial speculative results are +// valid in transaction order. +func validateInitialOCCResults( + ctx context.Context, + runner occSpeculativeRunner, + results []occTxExecution, +) (occValidationResult, bool, error) { + writes := newStateAccessIndex() + validation := occValidationResult{} + var cumulativeGasUsed uint64 + for txIndex, result := range results { + if err := ctx.Err(); err != nil { + return validation, false, err + } + validation.validationCount++ + needsRerun, err := needsSTMRerun( + &validation, + writes, + result, + cumulativeGasUsed, + runner.blockGasLimit, + result.sourcePrefix, + txIndex, + txIndex, + ) + if err != nil { + return validation, false, err + } + if needsRerun { + return occValidationResult{}, false, nil + } + if result.gasUsed > math.MaxUint64-cumulativeGasUsed { + validation.fallbackReason = occFallbackReasonGasOverflow + return validation, false, errors.New(occFallbackReasonGasOverflow) + } + cumulativeGasUsed += result.gasUsed + writes.addAllAt(txIndex, result.writeSet) + writes.addCommutativeBalanceDeltasAt(txIndex, result.commutativeBalanceDeltas) + } + return validation, true, nil +} + func (e *Executor) executeBlockOCCSequentialFallback(ctx context.Context, req PreparedBlock, source StateReader, validation occValidationResult, reason string) (*BlockResult, error) { if reason != "" { validation.fallbackReason = reason @@ -563,6 +617,28 @@ func (k stateAccessKind) String() string { } func (e *Executor) mergeOCCResults(ctx context.Context, results []occTxExecution, finalState *blockSTMState) (*BlockResult, error) { + blockResult, err := e.mergeOCCTransactionResults(ctx, results) + if err != nil { + return nil, err + } + finalState.ChangeSetInto(&blockResult.ChangeSet) + return blockResult, nil +} + +func (e *Executor) mergeConflictFreeOCCResults( + ctx context.Context, + results []occTxExecution, + source StateReader, +) (*BlockResult, error) { + blockResult, err := e.mergeOCCTransactionResults(ctx, results) + if err != nil { + return nil, err + } + mergeConflictFreeStateChanges(&blockResult.ChangeSet, results, source) + return blockResult, nil +} + +func (e *Executor) mergeOCCTransactionResults(ctx context.Context, results []occTxExecution) (*BlockResult, error) { blockResult, err := e.acquireBlockResult(ctx, len(results)) if err != nil { return nil, err @@ -580,10 +656,63 @@ func (e *Executor) mergeOCCResults(ctx context.Context, results []occTxExecution blockResult.Txs[i] = result.txResult blockResult.Receipts[i] = result.receipt } - finalState.ChangeSetInto(&blockResult.ChangeSet) return blockResult, nil } +func mergeConflictFreeStateChanges(changes *StateChangeSet, results []occTxExecution, source StateReader) { + changes.resetForReuse() + balanceIndexes := make(map[common.Address]int) + for _, result := range results { + for _, change := range result.changeSet.Balances { + delta := result.commutativeBalanceDeltas[change.Address] + _, normalWrite := result.writeSet[stateAccessKey{kind: stateAccessBalance, address: change.Address}] + if delta != nil && !normalWrite { + if index, ok := balanceIndexes[change.Address]; ok { + changes.Balances[index].Balance.Add(changes.Balances[index].Balance, delta) + continue + } + balance := cloneBig(source.GetBalance(change.Address)) + balance.Add(balance, delta) + balanceIndexes[change.Address] = len(changes.Balances) + changes.Balances = append(changes.Balances, BalanceChange{Address: change.Address, Balance: balance}) + continue + } + balance := cloneBig(change.Balance) + if index, ok := balanceIndexes[change.Address]; ok { + changes.Balances[index].Balance = balance + continue + } + balanceIndexes[change.Address] = len(changes.Balances) + changes.Balances = append(changes.Balances, BalanceChange{Address: change.Address, Balance: balance}) + } + changes.Nonces = append(changes.Nonces, result.changeSet.Nonces...) + for _, change := range result.changeSet.Code { + change.Code = cloneBytes(change.Code) + changes.Code = append(changes.Code, change) + } + changes.StorageClears = append(changes.StorageClears, result.changeSet.StorageClears...) + changes.Storage = append(changes.Storage, result.changeSet.Storage...) + } + sort.Slice(changes.Balances, func(i, j int) bool { + return bytes.Compare(changes.Balances[i].Address[:], changes.Balances[j].Address[:]) < 0 + }) + sort.Slice(changes.Nonces, func(i, j int) bool { + return bytes.Compare(changes.Nonces[i].Address[:], changes.Nonces[j].Address[:]) < 0 + }) + sort.Slice(changes.Code, func(i, j int) bool { + return bytes.Compare(changes.Code[i].Address[:], changes.Code[j].Address[:]) < 0 + }) + sort.Slice(changes.StorageClears, func(i, j int) bool { + return bytes.Compare(changes.StorageClears[i][:], changes.StorageClears[j][:]) < 0 + }) + sort.Slice(changes.Storage, func(i, j int) bool { + if cmp := bytes.Compare(changes.Storage[i].Address[:], changes.Storage[j].Address[:]); cmp != 0 { + return cmp < 0 + } + return bytes.Compare(changes.Storage[i].Key[:], changes.Storage[j].Key[:]) < 0 + }) +} + type blockSTMState struct { source StateReader balances map[common.Address]*big.Int