fix bugs

Author: stevenlanders

app/app.go

@@ -124,6 +124,7 @@ import (
 	"github.com/sei-protocol/sei-chain/x/evm"
 	evmante "github.com/sei-protocol/sei-chain/x/evm/ante"
 	"github.com/sei-protocol/sei-chain/x/evm/blocktest"
+	evmconfig "github.com/sei-protocol/sei-chain/x/evm/config"
 	evmkeeper "github.com/sei-protocol/sei-chain/x/evm/keeper"
 	"github.com/sei-protocol/sei-chain/x/evm/querier"
 	"github.com/sei-protocol/sei-chain/x/evm/replay"
@@ -263,8 +264,11 @@ var (
 	// EnableOCC allows tests to override default OCC enablement behavior
 	EnableOCC = true
 	// EnablePipelineProcessing enables the new pipeline-based block processing path
-	EnablePipelineProcessing = true
+	EnablePipelineProcessing = false
 	EmptyAppOptions          []AppOption
+
+	// EnableBenchmarkMode enables benchmark mode using sei-load generator
+	EnableBenchmarkMode = false
 )
 
 var (
@@ -397,6 +401,10 @@ type App struct {
 
 	// Pipeline processing (if enabled)
 	blockPipeline *BlockPipeline
+
+	// Benchmark mode - generator channel for load testing
+	benchmarkGeneratorCh <-chan *abci.ResponsePrepareProposal
+	enableBenchmarkMode  bool
 }
 
 type AppOption func(*App)
@@ -466,6 +474,7 @@ func New(
 		stateStore:            stateStore,
 		httpServerStartSignal: make(chan struct{}, 1),
 		wsServerStartSignal:   make(chan struct{}, 1),
+		enableBenchmarkMode:   EnableBenchmarkMode,
 	}
 
 	for _, option := range appOptions {
@@ -921,7 +930,11 @@ func New(
 	app.SetAnteDepGenerator(anteDepGenerator)
 	app.SetMidBlocker(app.MidBlocker)
 	app.SetEndBlocker(app.EndBlocker)
-	app.SetPrepareProposalHandler(app.PrepareProposalHandler)
+	if app.enableBenchmarkMode {
+		app.SetPrepareProposalHandler(app.PrepareProposalGeneratorHandler)
+	} else {
+		app.SetPrepareProposalHandler(app.PrepareProposalHandler)
+	}
 	app.SetProcessProposalHandler(app.ProcessProposalHandler)
 	app.SetFinalizeBlocker(app.FinalizeBlocker)
 	app.SetInplaceTestnetInitializer(app.inplacetestnetInitializer)
@@ -996,7 +1009,7 @@ func New(
 			PreprocessBlock,
 			ExecutePreprocessedEVMTransaction,
 			ExecuteCosmosTransaction,
-			10,   // preprocessorWorkers
+			25,   // preprocessorWorkers
 			1000, // bufferSize
 			1,    // startSequence
 		)
@@ -1008,6 +1021,20 @@ func New(
 		logger.Info("Using legacy block processing", "mode", "legacy")
 	}
 
+	// Initialize benchmark generator if enabled
+	if app.enableBenchmarkMode {
+		logger.Info("Initializing benchmark mode generator", "mode", "benchmark")
+		genCtx := context.Background()
+		// Get EVM chain ID from config mapping (not Cosmos chain ID)
+		evmChainID := evmconfig.GetEVMChainID(app.ChainID).Int64()
+		// Use a reasonable default for MaxTxBytes (20MB - conservative estimate)
+		// The generator will filter transactions to fit within this limit
+		// Tendermint will still validate the final proposal doesn't exceed req.MaxTxBytes
+		defaultMaxTxBytes := int64(20 * 1024 * 1024) // 20MB
+		app.benchmarkGeneratorCh = NewGeneratorCh(genCtx, app.encodingConfig.TxConfig, evmChainID, defaultMaxTxBytes, logger)
+		logger.Info("Benchmark generator initialized and started", "maxTxBytes", defaultMaxTxBytes)
+	}
+
 	return app
 }
 
@@ -1147,6 +1174,25 @@ func (app *App) PrepareProposalHandler(_ sdk.Context, req *abci.RequestPreparePr
 	}, nil
 }
 
+func (app *App) PrepareProposalGeneratorHandler(_ sdk.Context, req *abci.RequestPrepareProposal) (*abci.ResponsePrepareProposal, error) {
+	// Pull from generator channel - the generator has already filtered transactions by size
+	select {
+	case proposal, ok := <-app.benchmarkGeneratorCh:
+		if proposal == nil || !ok {
+			// Channel closed or no proposal available, return empty (req.Txs will remain in mempool)
+			return &abci.ResponsePrepareProposal{
+				TxRecords: []*abci.TxRecord{},
+			}, nil
+		}
+		return proposal, nil
+	default:
+		// No proposal ready yet, return empty (req.Txs will remain in mempool)
+		return &abci.ResponsePrepareProposal{
+			TxRecords: []*abci.TxRecord{},
+		}, nil
+	}
+}
+
 func (app *App) GetOptimisticProcessingInfo() OptimisticProcessingInfo {
 	app.optimisticProcessingInfoMutex.RLock()
 	defer app.optimisticProcessingInfoMutex.RUnlock()
@@ -1266,6 +1312,12 @@ func (app *App) FinalizeBlocker(ctx sdk.Context, req *abci.RequestFinalizeBlock)
 			return nil, fmt.Errorf("timeout waiting for pipeline result")
 		}
 
+		// Check if preprocessing failed
+		if processed.ExecutedBlock.PreprocessedBlock.PreprocessError != nil {
+			ctx.Logger().Error("Pipeline preprocessing failed", "error", processed.ExecutedBlock.PreprocessedBlock.PreprocessError, "height", req.Height)
+			return nil, fmt.Errorf("pipeline preprocessing failed: %w", processed.ExecutedBlock.PreprocessedBlock.PreprocessError)
+		}
+
 		// Verify we got the correct block
 		if processed.ExecutedBlock.PreprocessedBlock.Height != req.Height {
 			ctx.Logger().Error("Pipeline returned wrong block", "expected", req.Height, "got", processed.ExecutedBlock.PreprocessedBlock.Height)

app/execution.go

@@ -128,6 +128,26 @@ func ExecutePreprocessedEVMTransaction(ctx sdk.Context, preprocessed *pipelinety
 		vmError = result.Err.Error()
 	}
 
+	// Write receipt and deferred info during execution (needed for EndBlock)
+	// Set TxIndex in context - this is critical for deferred info storage
+	execCtx := ctx.WithTxIndex(preprocessed.TxIndex)
+	
+	// Write receipt (needed for deferred info bloom)
+	receipt, err := helper.GetKeeper().WriteReceipt(execCtx, stateDB, preprocessed.EVMMessage, uint32(etx.Type()), etx.Hash(), result.UsedGas, vmError)
+	if err != nil {
+		return &pipelinetypes.TransactionResult{
+			Code:      sdkerrors.ErrInvalidRequest.ABCICode(),
+			Codespace: sdkerrors.RootCodespace,
+			Log:       fmt.Sprintf("failed to write receipt: %s", err.Error()),
+			Surplus:   sdk.ZeroInt(),
+		}, fmt.Errorf("failed to write receipt: %w", err)
+	}
+
+	// Write deferred info (needed for EndBlock)
+	bloom := ethtypes.Bloom{}
+	bloom.SetBytes(receipt.LogsBloom)
+	helper.GetKeeper().AppendToEvmTxDeferredInfo(execCtx, bloom, etx.Hash(), surplus)
+
 	return &pipelinetypes.TransactionResult{
 		GasUsed:    int64(result.UsedGas), //nolint:gosec // GasUsed is bounded by block gas limit
 		ReturnData: result.ReturnData,

app/generator.go

@@ -0,0 +1,112 @@
+package app
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cosmos/cosmos-sdk/client"
+	evmtypes "github.com/sei-protocol/sei-chain/x/evm/types"
+	"github.com/sei-protocol/sei-chain/x/evm/types/ethtx"
+	"github.com/sei-protocol/sei-load/config"
+	"github.com/sei-protocol/sei-load/generator"
+	"github.com/sei-protocol/sei-load/generator/scenarios"
+	abci "github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/libs/log"
+)
+
+func NewGeneratorCh(ctx context.Context, txConfig client.TxConfig, chainID int64, maxTxBytes int64, logger log.Logger) <-chan *abci.ResponsePrepareProposal {
+	gen, err := generator.NewConfigBasedGenerator(&config.LoadConfig{
+		ChainID:    chainID,
+		SeiChainID: fmt.Sprintf("%d", chainID), // Use chainID as string for SeiChainID
+		Accounts:   &config.AccountConfig{Accounts: 5000},
+		Scenarios: []config.Scenario{{
+			Name:   scenarios.EVMTransfer,
+			Weight: 1,
+		}},
+	})
+	if err != nil {
+		panic("failed to initialize generator: " + err.Error())
+	}
+	ch := make(chan *abci.ResponsePrepareProposal, 1000)
+	go func() {
+		defer close(ch)
+		var height int64
+		for {
+			// bail on ctx err
+			if ctx.Err() != nil {
+				return
+			}
+			// generate txs like: txs := gen.GenerateN(1000)
+			loadTxs := gen.GenerateN(1000)
+			if len(loadTxs) == 0 {
+				continue
+			}
+
+			// Convert LoadTx to Cosmos SDK transaction bytes and filter by size
+			var totalBytes int64
+			txRecords := make([]*abci.TxRecord, 0, len(loadTxs))
+			for _, loadTx := range loadTxs {
+				if loadTx.EthTx == nil {
+					continue
+				}
+
+				// Convert Ethereum transaction to Cosmos SDK format
+				txData, err := ethtx.NewTxDataFromTx(loadTx.EthTx)
+				if err != nil {
+					logger.Error("failed to convert eth tx to tx data", "error", err)
+					continue
+				}
+
+				msg, err := evmtypes.NewMsgEVMTransaction(txData)
+				if err != nil {
+					logger.Error("failed to create msg evm transaction", "error", err)
+					continue
+				}
+
+				gasUsedEstimate := loadTx.EthTx.Gas() // Use gas limit from transaction
+
+				txBuilder := txConfig.NewTxBuilder()
+				if err = txBuilder.SetMsgs(msg); err != nil {
+					logger.Error("failed to set msgs", "error", err)
+					continue
+				}
+				txBuilder.SetGasEstimate(gasUsedEstimate)
+
+				txbz, encodeErr := txConfig.TxEncoder()(txBuilder.GetTx())
+				if encodeErr != nil {
+					logger.Error("failed to encode tx", "error", encodeErr)
+					continue
+				}
+
+				// Filter by MaxTxBytes - stop adding transactions if we exceed the limit
+				txSize := int64(len(txbz))
+				if totalBytes+txSize > maxTxBytes {
+					break
+				}
+				totalBytes += txSize
+
+				txRecords = append(txRecords, &abci.TxRecord{
+					Action: abci.TxRecord_GENERATED,
+					Tx:     txbz,
+				})
+			}
+
+			if len(txRecords) == 0 {
+				continue
+			}
+
+			proposal := &abci.ResponsePrepareProposal{
+				TxRecords: txRecords,
+			}
+
+			height++
+			fmt.Println("Generating Block ", height)
+			select {
+			case ch <- proposal:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	return ch
+}