[OCC] add async scheduler to seiv2 (unused)

.github/workflows/test.yml

@@ -24,7 +24,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.19
+          go-version: 1.21
       - name: Create a file with all core Cosmos SDK pkgs
         run: go list ./... > pkgs.txt
       - name: Split pkgs into 10 files
@@ -83,7 +83,7 @@ jobs:
       - uses: actions/setup-python@v3
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.19
+          go-version: 1.21
       - uses: technote-space/[email protected]
         with:
           PATTERNS: |
@@ -118,7 +118,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.19
+          go-version: 1.21
 
       # Download all coverage reports from the 'tests' job
       - name: Download coverage reports

baseapp/abci.go

@@ -12,8 +12,6 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/cosmos/cosmos-sdk/tasks"
-
 	"github.com/armon/go-metrics"
 	"github.com/gogo/protobuf/proto"
 	abci "github.com/tendermint/tendermint/abci/types"
@@ -23,6 +21,8 @@ import (
 
 	"github.com/cosmos/cosmos-sdk/codec"
 	snapshottypes "github.com/cosmos/cosmos-sdk/snapshots/types"
+	"github.com/cosmos/cosmos-sdk/tasks"
+	"github.com/cosmos/cosmos-sdk/tasksv2"
 	"github.com/cosmos/cosmos-sdk/telemetry"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
@@ -203,7 +203,7 @@ func (app *BaseApp) EndBlock(ctx sdk.Context, req abci.RequestEndBlock) (res abc
 // internal CheckTx state if the AnteHandler passes. Otherwise, the ResponseCheckTx
 // will contain releveant error information. Regardless of tx execution outcome,
 // the ResponseCheckTx will contain relevant gas execution context.
-func (app *BaseApp) CheckTx(ctx context.Context, req *abci.RequestCheckTx) (*abci.ResponseCheckTx, error) {
+func (app *BaseApp) CheckTx(ctx context.Context, req *abci.RequestCheckTx) (*abci.ResponseCheckTxV2, error) {
 	defer telemetry.MeasureSince(time.Now(), "abci", "check_tx")
 
 	var mode runTxMode
@@ -220,28 +220,50 @@ func (app *BaseApp) CheckTx(ctx context.Context, req *abci.RequestCheckTx) (*abc
 	}
 
 	sdkCtx := app.getContextForTx(mode, req.Tx)
-	gInfo, result, _, priority, err := app.runTx(sdkCtx, mode, req.Tx)
+	tx, err := app.txDecoder(req.Tx)
+	if err != nil {
+		res := sdkerrors.ResponseCheckTx(err, 0, 0, app.trace)
+		return &abci.ResponseCheckTxV2{ResponseCheckTx: &res}, err
+	}
+	gInfo, result, _, priority, pendingTxChecker, expireTxHandler, txCtx, err := app.runTx(sdkCtx, mode, tx, sha256.Sum256(req.Tx))
 	if err != nil {
 		res := sdkerrors.ResponseCheckTx(err, gInfo.GasWanted, gInfo.GasUsed, app.trace)
-		return &res, err
+		return &abci.ResponseCheckTxV2{ResponseCheckTx: &res}, err
 	}
 
-	return &abci.ResponseCheckTx{
-		GasWanted: int64(gInfo.GasWanted), // TODO: Should type accept unsigned ints?
-		Data:      result.Data,
-		Priority:  priority,
-	}, nil
+	res := &abci.ResponseCheckTxV2{
+		ResponseCheckTx: &abci.ResponseCheckTx{
+			GasWanted: int64(gInfo.GasWanted), // TODO: Should type accept unsigned ints?
+			Data:      result.Data,
+			Priority:  priority,
+		},
+		ExpireTxHandler:  expireTxHandler,
+		EVMNonce:         txCtx.EVMNonce(),
+		EVMSenderAddress: txCtx.EVMSenderAddress(),
+		IsEVM:            txCtx.IsEVM(),
+	}
+	if pendingTxChecker != nil {
+		res.IsPendingTransaction = true
+		res.Checker = pendingTxChecker
+	}
+
+	return res, nil
 }
 
 // DeliverTxBatch executes multiple txs
 func (app *BaseApp) DeliverTxBatch(ctx sdk.Context, req sdk.DeliverTxBatchRequest) (res sdk.DeliverTxBatchResponse) {
-	scheduler := tasks.NewScheduler(app.concurrencyWorkers, app.TracingInfo, app.DeliverTx)
-	// This will basically no-op the actual prefill if the metadata for the txs is empty
+	var scheduler tasks.Scheduler
+	if app.occAsync {
+		scheduler = tasksv2.NewScheduler(app.concurrencyWorkers, app.TracingInfo, app.DeliverTx)
+	} else {
+		scheduler = tasks.NewScheduler(app.concurrencyWorkers, app.TracingInfo, app.DeliverTx)
+	}
 
 	// process all txs, this will also initializes the MVS if prefill estimates was disabled
 	txRes, err := scheduler.ProcessAll(ctx, req.TxEntries)
 	if err != nil {
-		// TODO: handle error
+		app.logger.Error("DeliverTxBatch failed", "err", err)
+		panic(err)
 	}
 
 	responses := make([]*sdk.DeliverTxResult, 0, len(req.TxEntries))
@@ -256,9 +278,7 @@ func (app *BaseApp) DeliverTxBatch(ctx sdk.Context, req sdk.DeliverTxBatchReques
 // Otherwise, the ResponseDeliverTx will contain relevant error information.
 // Regardless of tx execution outcome, the ResponseDeliverTx will contain relevant
 // gas execution context.
-// TODO: (occ) this is the function called from sei-chain to perform execution of a transaction.
-// We'd likely replace this with an execution tasks that is scheduled by the OCC scheduler
-func (app *BaseApp) DeliverTx(ctx sdk.Context, req abci.RequestDeliverTx) (res abci.ResponseDeliverTx) {
+func (app *BaseApp) DeliverTx(ctx sdk.Context, req abci.RequestDeliverTx, tx sdk.Tx, checksum [32]byte) (res abci.ResponseDeliverTx) {
 	defer telemetry.MeasureSince(time.Now(), "abci", "deliver_tx")
 	defer func() {
 		for _, streamingListener := range app.abciListeners {
@@ -278,7 +298,7 @@ func (app *BaseApp) DeliverTx(ctx sdk.Context, req abci.RequestDeliverTx) (res a
 		telemetry.SetGauge(float32(gInfo.GasWanted), "tx", "gas", "wanted")
 	}()
 
-	gInfo, result, anteEvents, _, err := app.runTx(ctx.WithTxBytes(req.Tx).WithVoteInfos(app.voteInfos), runTxModeDeliver, req.Tx)
+	gInfo, result, anteEvents, _, _, _, resCtx, err := app.runTx(ctx.WithTxBytes(req.Tx).WithVoteInfos(app.voteInfos), runTxModeDeliver, tx, checksum)
 	if err != nil {
 		resultStr = "failed"
 		// if we have a result, use those events instead of just the anteEvents
@@ -288,13 +308,20 @@ func (app *BaseApp) DeliverTx(ctx sdk.Context, req abci.RequestDeliverTx) (res a
 		return sdkerrors.ResponseDeliverTxWithEvents(err, gInfo.GasWanted, gInfo.GasUsed, sdk.MarkEventsToIndex(anteEvents, app.indexEvents), app.trace)
 	}
 
-	return abci.ResponseDeliverTx{
+	res = abci.ResponseDeliverTx{
 		GasWanted: int64(gInfo.GasWanted), // TODO: Should type accept unsigned ints?
 		GasUsed:   int64(gInfo.GasUsed),   // TODO: Should type accept unsigned ints?
 		Log:       result.Log,
 		Data:      result.Data,
 		Events:    sdk.MarkEventsToIndex(result.Events, app.indexEvents),
 	}
+	if resCtx.IsEVM() {
+		res.EvmTxInfo = &abci.EvmTxInfo{
+			SenderAddress: resCtx.EVMSenderAddress(),
+			Nonce:         resCtx.EVMNonce(),
+		}
+	}
+	return
 }
 
 func (app *BaseApp) WriteStateToCommitAndGetWorkingHash() []byte {
@@ -1048,6 +1075,20 @@ func (app *BaseApp) ProcessProposal(ctx context.Context, req *abci.RequestProces
 		}
 	}()
 
+	defer func() {
+		if err := recover(); err != nil {
+			app.logger.Error(
+				"panic recovered in ProcessProposal",
+				"height", req.Height,
+				"time", req.Time,
+				"hash", fmt.Sprintf("%X", req.Hash),
+				"panic", err,
+			)
+
+			resp = &abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_REJECT}
+		}
+	}()
+
 	if app.processProposalHandler != nil {
 		resp, err = app.processProposalHandler(app.processProposalState.ctx, req)
 		if err != nil {

baseapp/baseapp.go

@@ -2,7 +2,6 @@ package baseapp
 
 import (
 	"context"
-	"crypto/sha256"
 	"fmt"
 	"reflect"
 	"strings"
@@ -61,6 +60,7 @@ const (
 	FlagChainID            = "chain-id"
 	FlagConcurrencyWorkers = "concurrency-workers"
 	FlagOccEnabled         = "occ-enabled"
+	FlagOccAsync           = "occ-async"
 )
 
 var (
@@ -158,8 +158,9 @@ type BaseApp struct { //nolint: maligned
 
 	ChainID string
 
-	votesInfoLock sync.RWMutex
-	commitLock    *sync.Mutex
+	votesInfoLock    sync.RWMutex
+	commitLock       *sync.Mutex
+	checkTxStateLock *sync.RWMutex
 
 	compactionInterval uint64
 
@@ -170,6 +171,7 @@ type BaseApp struct { //nolint: maligned
 
 	concurrencyWorkers int
 	occEnabled         bool
+	occAsync           bool
 }
 
 type appStore struct {
@@ -275,10 +277,12 @@ func NewBaseApp(
 		TracingInfo: &tracing.Info{
 			Tracer: &tr,
 		},
-		commitLock: &sync.Mutex{},
+		commitLock:       &sync.Mutex{},
+		checkTxStateLock: &sync.RWMutex{},
 	}
 
 	app.TracingInfo.SetContext(context.Background())
+	app.occAsync = cast.ToBool(appOpts.Get(FlagOccAsync))
 
 	for _, option := range options {
 		option(app)
@@ -305,6 +309,11 @@ func NewBaseApp(
 		app.concurrencyWorkers = config.DefaultConcurrencyWorkers
 	}
 
+	//TODO: remove occAsync feature flag after it's the default version
+	if app.occAsync {
+		app.logger.Info("asynchronous occ scheduler is enabled")
+	}
+
 	return app
 }
 
@@ -530,6 +539,8 @@ func (app *BaseApp) IsSealed() bool { return app.sealed }
 func (app *BaseApp) setCheckState(header tmproto.Header) {
 	ms := app.cms.CacheMultiStore()
 	ctx := sdk.NewContext(ms, header, true, app.logger).WithMinGasPrices(app.minGasPrices)
+	app.checkTxStateLock.Lock()
+	defer app.checkTxStateLock.Unlock()
 	if app.checkState == nil {
 		app.checkState = &state{
 			ms:  ms,
@@ -802,16 +813,15 @@ func (app *BaseApp) getContextForTx(mode runTxMode, txBytes []byte) sdk.Context
 
 // cacheTxContext returns a new context based off of the provided context with
 // a branched multi-store.
-// TODO: (occ) This is an example of where we wrap the multistore with a cache multistore, and then return a modified context using that multistore
-func (app *BaseApp) cacheTxContext(ctx sdk.Context, txBytes []byte) (sdk.Context, sdk.CacheMultiStore) {
+func (app *BaseApp) cacheTxContext(ctx sdk.Context, checksum [32]byte) (sdk.Context, sdk.CacheMultiStore) {
 	ms := ctx.MultiStore()
 	// TODO: https://github.com/cosmos/cosmos-sdk/issues/2824
 	msCache := ms.CacheMultiStore()
 	if msCache.TracingEnabled() {
 		msCache = msCache.SetTracingContext(
 			sdk.TraceContext(
 				map[string]interface{}{
-					"txHash": fmt.Sprintf("%X", sha256.Sum256(txBytes)),
+					"txHash": fmt.Sprintf("%X", checksum),
 				},
 			),
 		).(sdk.CacheMultiStore)
@@ -827,8 +837,16 @@ func (app *BaseApp) cacheTxContext(ctx sdk.Context, txBytes []byte) (sdk.Context
 // Note, gas execution info is always returned. A reference to a Result is
 // returned if the tx does not run out of gas and if all the messages are valid
 // and execute successfully. An error is returned otherwise.
-func (app *BaseApp) runTx(ctx sdk.Context, mode runTxMode, txBytes []byte) (gInfo sdk.GasInfo, result *sdk.Result, anteEvents []abci.Event, priority int64, err error) {
-
+func (app *BaseApp) runTx(ctx sdk.Context, mode runTxMode, tx sdk.Tx, checksum [32]byte) (
+	gInfo sdk.GasInfo,
+	result *sdk.Result,
+	anteEvents []abci.Event,
+	priority int64,
+	pendingTxChecker abci.PendingTxChecker,
+	expireHandler abci.ExpireTxHandler,
+	txCtx sdk.Context,
+	err error,
+) {
 	defer telemetry.MeasureThroughputSinceWithLabels(
 		telemetry.TxCount,
 		[]metrics.Label{
@@ -852,7 +870,7 @@ func (app *BaseApp) runTx(ctx sdk.Context, mode runTxMode, txBytes []byte) (gInf
 		spanCtx, span := app.TracingInfo.StartWithContext("RunTx", ctx.TraceSpanContext())
 		defer span.End()
 		ctx = ctx.WithTraceSpanContext(spanCtx)
-		span.SetAttributes(attribute.String("txHash", fmt.Sprintf("%X", sha256.Sum256(txBytes))))
+		span.SetAttributes(attribute.String("txHash", fmt.Sprintf("%X", checksum)))
 	}
 
 	// NOTE: GasWanted should be returned by the AnteHandler. GasUsed is
@@ -875,15 +893,14 @@ func (app *BaseApp) runTx(ctx sdk.Context, mode runTxMode, txBytes []byte) (gInf
 		gInfo = sdk.GasInfo{GasWanted: gasWanted, GasUsed: ctx.GasMeter().GasConsumed()}
 	}()
 
-	tx, err := app.txDecoder(txBytes)
-	if err != nil {
-		return sdk.GasInfo{}, nil, nil, 0, err
+	if tx == nil {
+		return sdk.GasInfo{}, nil, nil, 0, nil, nil, ctx, sdkerrors.Wrap(sdkerrors.ErrTxDecode, "tx decode error")
 	}
 
 	msgs := tx.GetMsgs()
 
 	if err := validateBasicTxMsgs(msgs); err != nil {
-		return sdk.GasInfo{}, nil, nil, 0, err
+		return sdk.GasInfo{}, nil, nil, 0, nil, nil, ctx, err
 	}
 
 	if app.anteHandler != nil {
@@ -904,7 +921,7 @@ func (app *BaseApp) runTx(ctx sdk.Context, mode runTxMode, txBytes []byte) (gInf
 		// NOTE: Alternatively, we could require that AnteHandler ensures that
 		// writes do not happen if aborted/failed.  This may have some
 		// performance benefits, but it'll be more difficult to get right.
-		anteCtx, msCache = app.cacheTxContext(ctx, txBytes)
+		anteCtx, msCache = app.cacheTxContext(ctx, checksum)
 		anteCtx = anteCtx.WithEventManager(sdk.NewEventManager())
 		newCtx, err := app.anteHandler(anteCtx, tx, mode == runTxModeSimulate)
 
@@ -928,7 +945,7 @@ func (app *BaseApp) runTx(ctx sdk.Context, mode runTxMode, txBytes []byte) (gInf
 		// GasMeter expected to be set in AnteHandler
 		gasWanted = ctx.GasMeter().Limit()
 		if err != nil {
-			return gInfo, nil, nil, 0, err
+			return gInfo, nil, nil, 0, nil, nil, ctx, err
 		}
 
 		// Dont need to validate in checkTx mode
@@ -944,11 +961,13 @@ func (app *BaseApp) runTx(ctx sdk.Context, mode runTxMode, txBytes []byte) (gInf
 					op.EmitValidationFailMetrics()
 				}
 				errMessage := fmt.Sprintf("Invalid Concurrent Execution antehandler missing %d access operations", len(missingAccessOps))
-				return gInfo, nil, nil, 0, sdkerrors.Wrap(sdkerrors.ErrInvalidConcurrencyExecution, errMessage)
+				return gInfo, nil, nil, 0, nil, nil, ctx, sdkerrors.Wrap(sdkerrors.ErrInvalidConcurrencyExecution, errMessage)
 			}
 		}
 
 		priority = ctx.Priority()
+		pendingTxChecker = ctx.PendingTxChecker()
+		expireHandler = ctx.ExpireTxHandler()
 		msCache.Write()
 		anteEvents = events.ToABCIEvents()
 		if app.TracingEnabled {
@@ -959,7 +978,7 @@ func (app *BaseApp) runTx(ctx sdk.Context, mode runTxMode, txBytes []byte) (gInf
 	// Create a new Context based off of the existing Context with a MultiStore branch
 	// in case message processing fails. At this point, the MultiStore
 	// is a branch of a branch.
-	runMsgCtx, msCache := app.cacheTxContext(ctx, txBytes)
+	runMsgCtx, msCache := app.cacheTxContext(ctx, checksum)
 
 	// Attempt to execute all messages and only update state if all messages pass
 	// and we're in DeliverTx. Note, runMsgs will never return a reference to a
@@ -974,7 +993,10 @@ func (app *BaseApp) runTx(ctx sdk.Context, mode runTxMode, txBytes []byte) (gInf
 		// append the events in the order of occurrence
 		result.Events = append(anteEvents, result.Events...)
 	}
-	return gInfo, result, anteEvents, priority, err
+	if ctx.CheckTxCallback() != nil {
+		ctx.CheckTxCallback()(ctx, err)
+	}
+	return gInfo, result, anteEvents, priority, pendingTxChecker, expireHandler, ctx, err
 }
 
 // runMsgs iterates through a list of messages and executes them with the provided
@@ -1021,7 +1043,7 @@ func (app *BaseApp) runMsgs(ctx sdk.Context, msgs []sdk.Msg, mode runTxMode) (*s
 			err          error
 		)
 
-		msgCtx, msgMsCache := app.cacheTxContext(ctx, []byte{})
+		msgCtx, msgMsCache := app.cacheTxContext(ctx, [32]byte{})
 		msgCtx = msgCtx.WithMessageIndex(i)
 
 		startTime := time.Now()
@@ -1164,5 +1186,7 @@ func (app *BaseApp) ReloadDB() error {
 }
 
 func (app *BaseApp) GetCheckCtx() sdk.Context {
+	app.checkTxStateLock.RLock()
+	defer app.checkTxStateLock.RUnlock()
 	return app.checkState.ctx
 }