feat: Optimistic Execution

baseapp/abci.go

@@ -494,6 +494,12 @@ func (app *BaseApp) ProcessProposal(req *abci.RequestProcessProposal) (resp *abc
 	// processed the first block, as we want to avoid overwriting the finalizeState
 	// after state changes during InitChain.
 	if req.Height > app.initialHeight {
+		// abort any running OE
+		if app.optimisticExec.Running() {
+			app.optimisticExec.Abort()
+			_, _ = app.optimisticExec.WaitResult() // ignore the result
+		}
+
 		app.setState(execModeFinalize, header)
 	}
 
@@ -534,6 +540,12 @@ func (app *BaseApp) ProcessProposal(req *abci.RequestProcessProposal) (resp *abc
 		return &abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_REJECT}, nil
 	}
 
+	// Only execute optimistic execution if OE is enabled and the block height is greater than the initial height.
+	// During the first block we'll be carrying state from InitChain, so it would be impossible for us to easily revert.
+	if app.optimisticExec.Enabled() && req.Height > app.initialHeight {
+		app.optimisticExec.Execute(req)
+	}
+
 	return resp, nil
 }
 
@@ -647,17 +659,7 @@ func (app *BaseApp) VerifyVoteExtension(req *abci.RequestVerifyVoteExtension) (r
 	return resp, err
 }
 
-// FinalizeBlock will execute the block proposal provided by RequestFinalizeBlock.
-// Specifically, it will execute an application's BeginBlock (if defined), followed
-// by the transactions in the proposal, finally followed by the application's
-// EndBlock (if defined).
-//
-// For each raw transaction, i.e. a byte slice, BaseApp will only execute it if
-// it adheres to the sdk.Tx interface. Otherwise, the raw transaction will be
-// skipped. This is to support compatibility with proposers injecting vote
-// extensions into the proposal, which should not themselves be executed in cases
-// where they adhere to the sdk.Tx interface.
-func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) {
+func (app *BaseApp) internalFinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) {
 	var events []abci.Event
 
 	if err := app.checkHalt(req.Height, req.Time); err != nil {
@@ -732,6 +734,12 @@ func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.Respons
 		return nil, err
 	}
 
+	// First check for an abort signal after beginBlock, as it's the first place
+	// we spend any significant amount of time.
+	if app.optimisticExec.Running() && app.optimisticExec.ShouldAbort() {
+		return nil, nil
+	}
+
 	events = append(events, beginBlock.Events...)
 
 	// Iterate over all raw transactions in the proposal and attempt to execute
@@ -758,6 +766,11 @@ func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.Respons
 			)
 		}
 
+		// check after every tx if we should abort
+		if app.optimisticExec.Running() && app.optimisticExec.ShouldAbort() {
+			return nil, nil
+		}
+
 		txResults = append(txResults, response)
 	}
 
@@ -770,6 +783,11 @@ func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.Respons
 		return nil, err
 	}
 
+	// check after endBlock if we should abort, to avoid propagating the result
+	if app.optimisticExec.Running() && app.optimisticExec.ShouldAbort() {
+		return nil, nil
+	}
+
 	events = append(events, endBlock.Events...)
 	cp := app.GetConsensusParams(app.finalizeBlockState.ctx)
 
@@ -778,10 +796,45 @@ func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.Respons
 		TxResults:             txResults,
 		ValidatorUpdates:      endBlock.ValidatorUpdates,
 		ConsensusParamUpdates: &cp,
-		AppHash:               app.workingHash(),
 	}, nil
 }
 
+// FinalizeBlock will execute the block proposal provided by RequestFinalizeBlock.
+// Specifically, it will execute an application's BeginBlock (if defined), followed
+// by the transactions in the proposal, finally followed by the application's
+// EndBlock (if defined).
+//
+// For each raw transaction, i.e. a byte slice, BaseApp will only execute it if
+// it adheres to the sdk.Tx interface. Otherwise, the raw transaction will be
+// skipped. This is to support compatibility with proposers injecting vote
+// extensions into the proposal, which should not themselves be executed in cases
+// where they adhere to the sdk.Tx interface.
+func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) {
+	if app.optimisticExec.Initialized() {
+		// check if the hash we got is the same as the one we are executing
+		aborted := app.optimisticExec.AbortIfNeeded(req.Hash)
+		// Wait for the OE to finish, regardless of whether it was aborted or not
+		res, err := app.optimisticExec.WaitResult()
+
+		// only return if we are not aborting
+		if !aborted {
+			res.AppHash = app.workingHash()
+			return res, err
+		}
+
+		// if it was aborted, we need to reset the state
+		app.finalizeBlockState = nil
+		app.optimisticExec.Reset()
+	}
+
+	// if no OE is running, just run the block (this is either a block replay or a OE that got aborted)
+	res, err := app.internalFinalizeBlock(req)
+	if res != nil {
+		res.AppHash = app.workingHash()
+	}
+	return res, err
+}
+
 // checkHalt checkes if height or time exceeds halt-height or halt-time respectively.
 func (app *BaseApp) checkHalt(height int64, time time.Time) error {
 	var halt bool

baseapp/baseapp.go

@@ -23,6 +23,7 @@ import (
 	"cosmossdk.io/store/snapshots"
 	storetypes "cosmossdk.io/store/types"
 
+	"github.com/cosmos/cosmos-sdk/baseapp/oe"
 	"github.com/cosmos/cosmos-sdk/codec"
 	codectypes "github.com/cosmos/cosmos-sdk/codec/types"
 	servertypes "github.com/cosmos/cosmos-sdk/server/types"
@@ -187,6 +188,8 @@ type BaseApp struct {
 	chainID string
 
 	cdc codec.Codec
+
+	optimisticExec *oe.OptimisticExecution
 }
 
 // NewBaseApp returns a reference to an initialized BaseApp. It accepts a

baseapp/oe/optimistic_execution.go

@@ -0,0 +1,174 @@
+package oe
+
+import (
+	"bytes"
+	"math/rand"
+	"sync"
+	"time"
+
+	abci "github.com/cometbft/cometbft/abci/types"
+
+	"cosmossdk.io/log"
+)
+
+type OptimisticExecution struct {
+	mtx         sync.RWMutex
+	stopCh      chan struct{}
+	shouldAbort bool
+	running     bool
+	initialized bool
+
+	// we could use generics here in the future to allow other types of req/resp
+	fn            func(*abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error)
+	request       *abci.RequestFinalizeBlock
+	response      *abci.ResponseFinalizeBlock
+	err           error
+	executionTime time.Duration
+	logger        log.Logger
+
+	// debugging options
+	abortRate int // number from 0 to 100
+}
+
+func NewOptimisticExecution(logger log.Logger, fn func(*abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error), opts ...func(*OptimisticExecution)) *OptimisticExecution {
+	oe := &OptimisticExecution{logger: logger, fn: fn}
+	for _, opt := range opts {
+		opt(oe)
+	}
+	return oe
+}
+
+func WithAbortRate(rate int) func(*OptimisticExecution) {
+	return func(oe *OptimisticExecution) {
+		oe.abortRate = rate
+	}
+}
+
+// Reset resets the OE context. Must be called whenever we want to invalidate
+// the current OE. For example when on FinalizeBlock we want to process the
+// block async, we run Reset() to make sure ShouldAbort() returns always false.
+func (oe *OptimisticExecution) Reset() {
+	oe.mtx.Lock()
+	defer oe.mtx.Unlock()
+	oe.request = nil
+	oe.response = nil
+	oe.err = nil
+	oe.executionTime = 0
+	oe.shouldAbort = false
+	oe.running = false
+	oe.initialized = false
+}
+
+func (oe *OptimisticExecution) Enabled() bool {
+	return oe != nil
+}
+
+// Initialized returns true if the OE was initialized, meaning that it contains
+// a request and it was run or it is running.
+func (oe *OptimisticExecution) Initialized() bool {
+	if oe == nil {
+		return false
+	}
+	oe.mtx.RLock()
+	defer oe.mtx.RUnlock()
+
+	return oe.initialized
+}
+
+// Execute initializes the OE and starts it in a goroutine.
+func (oe *OptimisticExecution) Execute(
+	req *abci.RequestProcessProposal,
+) {
+	oe.mtx.Lock()
+	defer oe.mtx.Unlock()
+
+	oe.stopCh = make(chan struct{})
+	oe.request = &abci.RequestFinalizeBlock{
+		Txs:                req.Txs,
+		DecidedLastCommit:  req.ProposedLastCommit,
+		Misbehavior:        req.Misbehavior,
+		Hash:               req.Hash,
+		Height:             req.Height,
+		Time:               req.Time,
+		NextValidatorsHash: req.NextValidatorsHash,
+		ProposerAddress:    req.ProposerAddress,
+	}
+
+	oe.logger.Debug("OE started")
+	start := time.Now()
+	oe.running = true
+	oe.initialized = true
+
+	go func() {
+		resp, err := oe.fn(oe.request)
+		oe.mtx.Lock()
+		oe.executionTime = time.Since(start)
+		oe.logger.Debug("OE finished", "duration", oe.executionTime.String())
+		oe.response, oe.err = resp, err
+		oe.running = false
+		close(oe.stopCh)
+		oe.mtx.Unlock()
+	}()
+}
+
+// AbortIfNeeded aborts the OE if the request hash is not the same as the one in
+// the running OE. Returns true if the OE was aborted.
+func (oe *OptimisticExecution) AbortIfNeeded(reqHash []byte) bool {
+	if oe == nil {
+		return false
+	}
+
+	oe.mtx.Lock()
+	defer oe.mtx.Unlock()
+
+	if !bytes.Equal(oe.request.Hash, reqHash) {
+		oe.logger.Debug("OE aborted due to hash mismatch", "oe_hash", oe.request.Hash, "req_hash", reqHash)
+		oe.shouldAbort = true
+	}
+
+	// test abort rate
+	if oe.abortRate > 0 && !oe.shouldAbort {
+		oe.shouldAbort = rand.Intn(100) < oe.abortRate
+		if oe.shouldAbort {
+			oe.logger.Debug("OE aborted due to test abort rate")
+		}
+	}
+
+	return oe.shouldAbort
+}
+
+// Abort aborts the OE unconditionally.
+func (oe *OptimisticExecution) Abort() {
+	oe.mtx.Lock()
+	defer oe.mtx.Unlock()
+	oe.shouldAbort = true
+}
+
+// ShouldAbort must only be used in the fn passed to SetupOptimisticExecution to
+// check if the OE was aborted and return as soon as possible.
+func (oe *OptimisticExecution) ShouldAbort() bool {
+	if oe == nil {
+		return false
+	}
+
+	oe.mtx.RLock()
+	defer oe.mtx.RUnlock()
+	return oe.shouldAbort
+}
+
+// Running returns true if the OE is still running.
+func (oe *OptimisticExecution) Running() bool {
+	if oe == nil {
+		return false
+	}
+
+	oe.mtx.RLock()
+	defer oe.mtx.RUnlock()
+	return oe.running
+}
+
+// WaitResult waits for the OE to finish and returns the result.
+func (oe *OptimisticExecution) WaitResult() (*abci.ResponseFinalizeBlock, error) {
+	<-oe.stopCh
+	return oe.response, oe.err
+}

baseapp/options.go

@@ -13,6 +13,7 @@ import (
 	snapshottypes "cosmossdk.io/store/snapshots/types"
 	storetypes "cosmossdk.io/store/types"
 
+	"github.com/cosmos/cosmos-sdk/baseapp/oe"
 	"github.com/cosmos/cosmos-sdk/codec"
 	"github.com/cosmos/cosmos-sdk/codec/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
@@ -104,6 +105,13 @@ func SetChainID(chainID string) func(*BaseApp) {
 	return func(app *BaseApp) { app.chainID = chainID }
 }
 
+// SetOptimisticExecution enables optimistic execution.
+func SetOptimisticExecution(opts ...func(*oe.OptimisticExecution)) func(*BaseApp) {
+	return func(app *BaseApp) {
+		app.optimisticExec = oe.NewOptimisticExecution(app.logger, app.internalFinalizeBlock, opts...)
+	}
+}
+
 func (app *BaseApp) SetName(name string) {
 	if app.sealed {
 		panic("SetName() on sealed BaseApp")

server/util.go

@@ -517,6 +517,7 @@ func DefaultBaseappOptions(appOpts types.AppOptions) []func(*baseapp.BaseApp) {
 		defaultMempool,
 		baseapp.SetChainID(chainID),
 		baseapp.SetQueryGasLimit(cast.ToUint64(appOpts.Get(FlagQueryGasLimit))),
+		baseapp.SetOptimisticExecution(),
 	}
 }