feat: Optimistic Execution

baseapp/abci.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"fmt"
+	"log"
 	"os"
 	"sort"
 	"strings"
@@ -532,6 +533,18 @@ func (app *BaseApp) ProcessProposal(req *abci.RequestProcessProposal) (resp *abc
 		return &abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_REJECT}, nil
 	}
 
+	if app.oeEnabled {
+		if app.oeInfo == nil {
+			app.oeInfo = SetupOptimisticExecution(req, app.internalFinalizeBlock)
+			app.oeInfo.Execute()
+		} else if app.oeInfo.AbortIfNeeded(req.Hash) {
+			// TODO: this will block until the OE is aborted, which can take a bit. Maybe do it async?
+			// if aborted, restart with the new proposal
+			app.oeInfo = SetupOptimisticExecution(req, app.internalFinalizeBlock)
+			app.oeInfo.Execute()
+		}
+	}
+
 	return resp, nil
 }
 
@@ -635,17 +648,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.validateFinalizeBlockHeight(req); err != nil {
@@ -711,7 +714,19 @@ func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.Respons
 			WithHeaderHash(req.Hash)
 	}
 
-	beginBlock := app.beginBlock(req)
+	beginBlock, err := app.beginBlock(req)
+	if err != nil {
+		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.oeEnabled && app.oeInfo != nil {
+		if app.oeInfo.ShouldAbort() {
+			return nil, nil
+		}
+	}
+
 	events = append(events, beginBlock.Events...)
 
 	// Iterate over all raw transactions in the proposal and attempt to execute
@@ -721,6 +736,13 @@ func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.Respons
 	// vote extensions, so skip those.
 	txResults := make([]*abci.ExecTxResult, 0, len(req.Txs))
 	for _, rawTx := range req.Txs {
+		// check before every tx if we should abort
+		if app.oeEnabled && app.oeInfo != nil {
+			if app.oeInfo.ShouldAbort() {
+				return nil, nil
+			}
+		}
+
 		if _, err := app.txDecoder(rawTx); err == nil {
 			txResults = append(txResults, app.deliverTx(rawTx))
 		}
@@ -747,6 +769,33 @@ func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.Respons
 	}, 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) {
+	defer func() {
+		app.oeInfo = nil
+	}()
+
+	if app.oeInfo != nil && app.oeEnabled {
+		// check if the hash we got is the same as the one we are executing
+		if !app.oeInfo.AbortIfNeeded(req.Hash) {
+			// no need to abort OE, wait for the result
+			return app.oeInfo.WaitResult()
+		}
+	}
+
+	log.Println("NOT running OE ❌")
+	return app.internalFinalizeBlock(req)
+}
+
 // Commit implements the ABCI interface. It will commit all state that exists in
 // the deliver state's multi-store and includes the resulting commit ID in the
 // returned abci.ResponseCommit. Commit will set the check state based on the

baseapp/abci_test.go

@@ -1171,11 +1171,11 @@ func TestABCI_Proposal_HappyPath(t *testing.T) {
 
 	res, err := suite.baseApp.FinalizeBlock(&abci.RequestFinalizeBlock{
 		Height: suite.baseApp.LastBlockHeight() + 1,
-		Txs:    [][]byte{txBytes},
+		Txs:    reqProposalTxBytes[:],
 	})
 	require.NoError(t, err)
 
-	require.Equal(t, 1, pool.CountTx())
+	require.Equal(t, 0, pool.CountTx())
 
 	require.NotEmpty(t, res.TxResults[0].Events)
 	require.True(t, res.TxResults[0].IsOK(), fmt.Sprintf("%v", res))

baseapp/baseapp.go

@@ -179,6 +179,9 @@ type BaseApp struct {
 	chainID string
 
 	cdc codec.Codec
+
+	oeInfo    *OptimisticExecutionInfo
+	oeEnabled bool
 }
 
 // NewBaseApp returns a reference to an initialized BaseApp. It accepts a
@@ -197,6 +200,7 @@ func NewBaseApp(
 		msgServiceRouter: NewMsgServiceRouter(),
 		txDecoder:        txDecoder,
 		fauxMerkleMode:   false,
+		oeEnabled:        true,
 	}
 
 	for _, option := range options {
@@ -679,7 +683,7 @@ func (app *BaseApp) cacheTxContext(ctx sdk.Context, txBytes []byte) (sdk.Context
 	return ctx.WithMultiStore(msCache), msCache
 }
 
-func (app *BaseApp) beginBlock(req *abci.RequestFinalizeBlock) sdk.BeginBlock {
+func (app *BaseApp) beginBlock(req *abci.RequestFinalizeBlock) (sdk.BeginBlock, error) {
 	var (
 		resp sdk.BeginBlock
 		err  error
@@ -688,7 +692,7 @@ func (app *BaseApp) beginBlock(req *abci.RequestFinalizeBlock) sdk.BeginBlock {
 	if app.beginBlocker != nil {
 		resp, err = app.beginBlocker(app.finalizeBlockState.ctx)
 		if err != nil {
-			panic(err)
+			return resp, err
 		}
 
 		// append BeginBlock attributes to all events in the EndBlock response
@@ -702,7 +706,7 @@ func (app *BaseApp) beginBlock(req *abci.RequestFinalizeBlock) sdk.BeginBlock {
 		resp.Events = sdk.MarkEventsToIndex(resp.Events, app.indexEvents)
 	}
 
-	return resp
+	return resp, nil
 }
 
 func (app *BaseApp) deliverTx(tx []byte) *abci.ExecTxResult {
@@ -750,7 +754,7 @@ func (app *BaseApp) endBlock(ctx context.Context) (sdk.EndBlock, error) {
 	if app.endBlocker != nil {
 		eb, err := app.endBlocker(app.finalizeBlockState.ctx)
 		if err != nil {
-			panic(err)
+			return endblock, err
 		}
 
 		// append EndBlock attributes to all events in the EndBlock response

baseapp/optimistic_execution.go

@@ -0,0 +1,77 @@
+package baseapp
+
+import (
+	"bytes"
+	"log"
+
+	abci "github.com/cometbft/cometbft/abci/types"
+)
+
+type OptimisticExecutionInfo struct {
+	completeSignal chan struct{}
+	abortSignal    chan struct{}
+	fn             func(*abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error)
+	Aborted        bool
+	Request        *abci.RequestFinalizeBlock
+	Response       *abci.ResponseFinalizeBlock
+	Error          error
+}
+
+func SetupOptimisticExecution(
+	req *abci.RequestProcessProposal,
+	fn func(*abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error),
+) *OptimisticExecutionInfo {
+	return &OptimisticExecutionInfo{
+		completeSignal: make(chan struct{}),
+		abortSignal:    make(chan struct{}),
+		fn:             fn,
+		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,
+		},
+	}
+}
+
+func (oe *OptimisticExecutionInfo) Execute() {
+	go func() {
+		log.Println("Running OE ✅")
+		oe.Response, oe.Error = oe.fn(oe.Request)
+		oe.completeSignal <- struct{}{}
+	}()
+}
+
+// AbortIfNeeded
+// If the request hash is not the same as the one in the OE, then abort the OE
+// and wait for the abort to happen. Returns true if the OE was aborted.
+func (oe *OptimisticExecutionInfo) AbortIfNeeded(reqHash []byte) bool {
+	if !bytes.Equal(oe.Request.Hash, reqHash) {
+		oe.abortSignal <- struct{}{}
+		// wait for the abort to happen
+		<-oe.completeSignal
+		oe.Aborted = true
+		return true
+	}
+	return false
+}
+
+// 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 *OptimisticExecutionInfo) ShouldAbort() bool {
+	select {
+	case <-oe.abortSignal:
+		return true
+	default:
+		return false
+	}
+}
+
+func (oe *OptimisticExecutionInfo) WaitResult() (*abci.ResponseFinalizeBlock, error) {
+	<-oe.completeSignal
+	return oe.Response, oe.Error
+}