feat(bb): Defer in proposal handlers, more interfaces for lanes, clean up (#165)

blockbuster/abci/abci.go

@@ -6,6 +6,7 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/skip-mev/pob/blockbuster"
 	"github.com/skip-mev/pob/blockbuster/lanes/terminator"
+	"github.com/skip-mev/pob/blockbuster/utils"
 )
 
 type (
@@ -33,10 +34,22 @@ func NewProposalHandler(logger log.Logger, mempool blockbuster.Mempool) *Proposa
 // the default lane will not have a boundary on the number of bytes that can be included in the proposal and
 // will include all valid transactions in the proposal (up to MaxTxBytes).
 func (h *ProposalHandler) PrepareProposalHandler() sdk.PrepareProposalHandler {
-	return func(ctx sdk.Context, req abci.RequestPrepareProposal) abci.ResponsePrepareProposal {
+	return func(ctx sdk.Context, req abci.RequestPrepareProposal) (resp abci.ResponsePrepareProposal) {
+		// In the case where there is a panic, we recover here and return an empty proposal.
+		defer func() {
+			if err := recover(); err != nil {
+				h.logger.Error("failed to prepare proposal", "err", err)
+				resp = abci.ResponsePrepareProposal{Txs: make([][]byte, 0)}
+			}
+		}()
+
 		proposal := h.prepareLanesHandler(ctx, blockbuster.NewProposal(req.MaxTxBytes))
 
-		return abci.ResponsePrepareProposal{Txs: proposal.Txs}
+		resp = abci.ResponsePrepareProposal{
+			Txs: proposal.Txs,
+		}
+
+		return
 	}
 }
 
@@ -45,7 +58,16 @@ func (h *ProposalHandler) PrepareProposalHandler() sdk.PrepareProposalHandler {
 // If a lane's portion of the proposal is invalid, we reject the proposal. After a lane's portion
 // of the proposal is verified, we pass the remaining transactions to the next lane in the chain.
 func (h *ProposalHandler) ProcessProposalHandler() sdk.ProcessProposalHandler {
-	return func(ctx sdk.Context, req abci.RequestProcessProposal) abci.ResponseProcessProposal {
+	return func(ctx sdk.Context, req abci.RequestProcessProposal) (resp abci.ResponseProcessProposal) {
+		// In the case where any of the lanes panic, we recover here and return a reject status.
+		defer func() {
+			if err := recover(); err != nil {
+				h.logger.Error("failed to process proposal", "err", err)
+				resp = abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_REJECT}
+			}
+		}()
+
+		// Verify the proposal using the verification logic from each lane.
 		if _, err := h.processLanesHandler(ctx, req.Txs); err != nil {
 			h.logger.Error("failed to validate the proposal", "err", err)
 			return abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_REJECT}
@@ -58,6 +80,9 @@ func (h *ProposalHandler) ProcessProposalHandler() sdk.ProcessProposalHandler {
 // ChainPrepareLanes chains together the proposal preparation logic from each lane
 // into a single function. The first lane in the chain is the first lane to be prepared and
 // the last lane in the chain is the last lane to be prepared.
+//
+// In the case where any of the lanes fail to prepare the partial proposal, the lane that failed
+// will be skipped and the next lane in the chain will be called to prepare the proposal.
 func ChainPrepareLanes(chain ...blockbuster.Lane) blockbuster.PrepareLanesHandler {
 	if len(chain) == 0 {
 		return nil
@@ -68,8 +93,62 @@ func ChainPrepareLanes(chain ...blockbuster.Lane) blockbuster.PrepareLanesHandle
 		chain = append(chain, terminator.Terminator{})
 	}
 
-	return func(ctx sdk.Context, proposal *blockbuster.Proposal) *blockbuster.Proposal {
-		return chain[0].PrepareLane(ctx, proposal, ChainPrepareLanes(chain[1:]...))
+	return func(ctx sdk.Context, partialProposal *blockbuster.Proposal) (finalProposal *blockbuster.Proposal) {
+		lane := chain[0]
+		lane.Logger().Info("preparing lane", "lane", lane.Name())
+
+		// Cache the context in the case where any of the lanes fail to prepare the proposal.
+		cacheCtx, write := ctx.CacheContext()
+
+		defer func() {
+			if err := recover(); err != nil {
+				lane.Logger().Error("failed to prepare lane", "lane", lane.Name(), "err", err)
+
+				lanesRemaining := len(chain)
+				switch {
+				case lanesRemaining <= 2:
+					// If there are only two lanes remaining, then the first lane in the chain
+					// is the lane that failed to prepare the partial proposal and the second lane in the
+					// chain is the terminator lane. We return the proposal as is.
+					finalProposal = partialProposal
+				default:
+					// If there are more than two lanes remaining, then the first lane in the chain
+					// is the lane that failed to prepare the proposal but the second lane in the
+					// chain is not the terminator lane so there could potentially be more transactions
+					// added to the proposal
+					maxTxBytesForLane := utils.GetMaxTxBytesForLane(
+						partialProposal,
+						chain[1].GetMaxBlockSpace(),
+					)
+
+					finalProposal = chain[1].PrepareLane(
+						ctx,
+						partialProposal,
+						maxTxBytesForLane,
+						ChainPrepareLanes(chain[2:]...),
+					)
+				}
+			} else {
+				// Write the cache to the context since we know that the lane successfully prepared
+				// the partial proposal.
+				write()
+
+				lane.Logger().Info("prepared lane", "lane", lane.Name())
+			}
+		}()
+
+		// Get the maximum number of bytes that can be included in the proposal for this lane.
+		maxTxBytesForLane := utils.GetMaxTxBytesForLane(
+			partialProposal,
+			lane.GetMaxBlockSpace(),
+		)
+
+		return lane.PrepareLane(
+			cacheCtx,
+			partialProposal,
+			maxTxBytesForLane,
+			ChainPrepareLanes(chain[1:]...),
+		)
 	}
 }
 
@@ -87,6 +166,16 @@ func ChainProcessLanes(chain ...blockbuster.Lane) blockbuster.ProcessLanesHandle
 	}
 
 	return func(ctx sdk.Context, proposalTxs [][]byte) (sdk.Context, error) {
+		// Short circuit if there are no transactions to process.
+		if len(proposalTxs) == 0 {
+			return ctx, nil
+		}
+
+		chain[0].Logger().Info("processing lane", "lane", chain[0].Name())
+		if err := chain[0].ProcessLaneBasic(proposalTxs); err != nil {
+			return ctx, err
+		}
+
 		return chain[0].ProcessLane(ctx, proposalTxs, ChainProcessLanes(chain[1:]...))
 	}
 }

blockbuster/abci/check_tx.go

@@ -0,0 +1,216 @@
+package abci
+
+import (
+	"context"
+	"fmt"
+
+	cometabci "github.com/cometbft/cometbft/abci/types"
+	log "github.com/cometbft/cometbft/libs/log"
+	tmproto "github.com/cometbft/cometbft/proto/tendermint/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
+	"github.com/skip-mev/pob/x/builder/types"
+)
+
+type (
+	// CheckTxHandler is a wrapper around baseapp's CheckTx method that allows us to
+	// verify bid transactions against the latest committed state. All other transactions
+	// are executed normally using base app's CheckTx. This defines all of the
+	// dependencies that are required to verify a bid transaction.
+	CheckTxHandler struct {
+		// baseApp is utilized to retrieve the latest committed state and to call
+		// baseapp's CheckTx method.
+		baseApp BaseApp
+
+		// txDecoder is utilized to decode transactions to determine if they are
+		// bid transactions.
+		txDecoder sdk.TxDecoder
+
+		// TOBLane is utilized to retrieve the bid info of a transaction and to
+		// insert a bid transaction into the application-side mempool.
+		tobLane TOBLane
+
+		// anteHandler is utilized to verify the bid transaction against the latest
+		// committed state.
+		anteHandler sdk.AnteHandler
+
+		// chainID is the chain ID of the blockchain.
+		chainID string
+	}
+
+	// CheckTx is baseapp's CheckTx method that checks the validity of a
+	// transaction.
+	CheckTx func(cometabci.RequestCheckTx) cometabci.ResponseCheckTx
+
+	// TOBLane is the interface that defines all of the dependencies that
+	// are required to interact with the top of block lane.
+	TOBLane interface {
+		// GetAuctionBidInfo is utilized to retrieve the bid info of a transaction.
+		GetAuctionBidInfo(tx sdk.Tx) (*types.BidInfo, error)
+
+		// Insert is utilized to insert a transaction into the application-side mempool.
+		Insert(ctx context.Context, tx sdk.Tx) error
+
+		// WrapBundleTransaction is utilized to wrap a transaction included in a bid transaction
+		// into an sdk.Tx.
+		WrapBundleTransaction(tx []byte) (sdk.Tx, error)
+	}
+
+	// BaseApp is an interface that allows us to call baseapp's CheckTx method
+	// as well as retrieve the latest committed state.
+	BaseApp interface {
+		// CommitMultiStore is utilized to retrieve the latest committed state.
+		CommitMultiStore() sdk.CommitMultiStore
+
+		// CheckTx is baseapp's CheckTx method that checks the validity of a
+		// transaction.
+		CheckTx(cometabci.RequestCheckTx) cometabci.ResponseCheckTx
+
+		// Logger is utilized to log errors.
+		Logger() log.Logger
+
+		// LastBlockHeight is utilized to retrieve the latest block height.
+		LastBlockHeight() int64
+
+		// GetConsensusParams is utilized to retrieve the consensus params.
+		GetConsensusParams(ctx sdk.Context) *tmproto.ConsensusParams
+	}
+)
+
+// NewCheckTxHandler is a constructor for CheckTxHandler.
+func NewCheckTxHandler(
+	baseApp BaseApp,
+	txDecoder sdk.TxDecoder,
+	tobLane TOBLane,
+	anteHandler sdk.AnteHandler,
+	chainID string,
+) *CheckTxHandler {
+	return &CheckTxHandler{
+		baseApp:     baseApp,
+		txDecoder:   txDecoder,
+		tobLane:     tobLane,
+		anteHandler: anteHandler,
+		chainID:     chainID,
+	}
+}
+
+// CheckTxHandler is a wrapper around baseapp's CheckTx method that allows us to
+// verify bid transactions against the latest committed state. All other transactions
+// are executed normally. We must verify each bid tx and all of its bundled transactions
+// before we can insert it into the mempool against the latest commit state because
+// otherwise the auction can be griefed. No state changes are applied to the state
+// during this process.
+func (handler *CheckTxHandler) CheckTx() CheckTx {
+	return func(req cometabci.RequestCheckTx) (resp cometabci.ResponseCheckTx) {
+		defer func() {
+			if err := recover(); err != nil {
+				resp = sdkerrors.ResponseCheckTxWithEvents(fmt.Errorf("panic in check tx handler: %s", err), 0, 0, nil, false)
+			}
+		}()
+
+		tx, err := handler.txDecoder(req.Tx)
+		if err != nil {
+			return sdkerrors.ResponseCheckTxWithEvents(fmt.Errorf("failed to decode tx: %w", err), 0, 0, nil, false)
+		}
+
+		// Attempt to get the bid info of the transaction.
+		bidInfo, err := handler.tobLane.GetAuctionBidInfo(tx)
+		if err != nil {
+			return sdkerrors.ResponseCheckTxWithEvents(fmt.Errorf("failed to get auction bid info: %w", err), 0, 0, nil, false)
+		}
+
+		// If this is not a bid transaction, we just execute it normally.
+		if bidInfo == nil {
+			return handler.baseApp.CheckTx(req)
+		}
+
+		// We attempt to get the latest committed state in order to verify transactions
+		// as if they were to be executed at the top of the block. After verification, this
+		// context will be discarded and will not apply any state changes.
+		ctx := handler.GetContextForBidTx(req)
+
+		// Verify the bid transaction.
+		gasInfo, err := handler.ValidateBidTx(ctx, tx, bidInfo)
+		if err != nil {
+			return sdkerrors.ResponseCheckTxWithEvents(fmt.Errorf("invalid bid tx: %w", err), gasInfo.GasWanted, gasInfo.GasUsed, nil, false)
+		}
+
+		// If the bid transaction is valid, we know we can insert it into the mempool for consideration in the next block.
+		if err := handler.tobLane.Insert(ctx, tx); err != nil {
+			return sdkerrors.ResponseCheckTxWithEvents(fmt.Errorf("invalid bid tx; failed to insert bid transaction into mempool: %w", err), gasInfo.GasWanted, gasInfo.GasUsed, nil, false)
+		}
+
+		return cometabci.ResponseCheckTx{
+			Code:      cometabci.CodeTypeOK,
+			GasWanted: int64(gasInfo.GasWanted),
+			GasUsed:   int64(gasInfo.GasUsed),
+		}
+	}
+}
+
+// ValidateBidTx is utilized to verify the bid transaction against the latest committed state.
+func (handler *CheckTxHandler) ValidateBidTx(ctx sdk.Context, bidTx sdk.Tx, bidInfo *types.BidInfo) (sdk.GasInfo, error) {
+	// Verify the bid transaction.
+	ctx, err := handler.anteHandler(ctx, bidTx, false)
+	if err != nil {
+		return sdk.GasInfo{}, fmt.Errorf("invalid bid tx; failed to execute ante handler: %w", err)
+	}
+
+	// Store the gas info and priority of the bid transaction before applying changes with other transactions.
+	gasInfo := sdk.GasInfo{
+		GasWanted: ctx.GasMeter().Limit(),
+		GasUsed:   ctx.GasMeter().GasConsumed(),
+	}
+
+	// Verify all of the bundled transactions.
+	for _, tx := range bidInfo.Transactions {
+		bundledTx, err := handler.tobLane.WrapBundleTransaction(tx)
+		if err != nil {
+			return gasInfo, fmt.Errorf("invalid bid tx; failed to decode bundled tx: %w", err)
+		}
+
+		// bid txs cannot be included in bundled txs
+		bidInfo, _ := handler.tobLane.GetAuctionBidInfo(bundledTx)
+		if bidInfo != nil {
+			return gasInfo, fmt.Errorf("invalid bid tx; bundled tx cannot be a bid tx")
+		}
+
+		if ctx, err = handler.anteHandler(ctx, bundledTx, false); err != nil {
+			return gasInfo, fmt.Errorf("invalid bid tx; failed to execute bundled transaction: %w", err)
+		}
+	}
+
+	return gasInfo, nil
+}
+
+// GetContextForBidTx is returns the latest committed state and sets the context given
+// the checkTx request.
+func (handler *CheckTxHandler) GetContextForBidTx(req cometabci.RequestCheckTx) sdk.Context {
+	// Retrieve the commit multi-store which is used to retrieve the latest committed state.
+	ms := handler.baseApp.CommitMultiStore().CacheMultiStore()
+
+	// Create a new context based off of the latest committed state.
+	header := tmproto.Header{
+		Height:  handler.baseApp.LastBlockHeight(),
+		ChainID: handler.chainID, // TODO: Replace with actual chain ID. This is currently not exposed by the app.
+	}
+	ctx, _ := sdk.NewContext(ms, header, true, handler.baseApp.Logger()).CacheContext()
+
+	// Set the context to the correct checking mode.
+	switch req.Type {
+	case cometabci.CheckTxType_New:
+		ctx = ctx.WithIsCheckTx(true)
+	case cometabci.CheckTxType_Recheck:
+		ctx = ctx.WithIsReCheckTx(true)
+	default:
+		panic("unknown check tx type")
+	}
+
+	// Set the remaining important context values.
+	ctx = ctx.
+		WithTxBytes(req.Tx).
+		WithEventManager(sdk.NewEventManager()).
+		WithConsensusParams(handler.baseApp.GetConsensusParams(ctx))
+
+	return ctx
+}

blockbuster/lane.go

@@ -3,6 +3,7 @@ package blockbuster
 import (
 	"crypto/sha256"
 	"encoding/hex"
+	"fmt"
 
 	"github.com/cometbft/cometbft/libs/log"
 	sdk "github.com/cosmos/cosmos-sdk/types"
@@ -15,8 +16,8 @@ type (
 		// Txs is the list of transactions in the proposal.
 		Txs [][]byte
 
-		// SelectedTxs is a cache of the selected transactions in the proposal.
-		SelectedTxs map[string]struct{}
+		// Cache is a cache of the selected transactions in the proposal.
+		Cache map[string]struct{}
 
 		// TotalTxBytes is the total number of bytes currently included in the proposal.
 		TotalTxBytes int64
@@ -65,14 +66,29 @@ type (
 		// Contains returns true if the mempool contains the given transaction.
 		Contains(tx sdk.Tx) (bool, error)
 
-		// PrepareLane which builds a portion of the block. Inputs include the max
-		// number of bytes that can be included in the block and the selected transactions
-		// thus from from previous lane(s) as mapping from their HEX-encoded hash to
-		// the raw transaction.
-		PrepareLane(ctx sdk.Context, proposal *Proposal, next PrepareLanesHandler) *Proposal
+		// PrepareLane builds a portion of the block. It inputs the maxTxBytes that can be
+		// included in the proposal for the given lane, the partial proposal, and a function
+		// to call the next lane in the chain. The next lane in the chain will be called with
+		// the updated proposal and context.
+		PrepareLane(ctx sdk.Context, proposal *Proposal, maxTxBytes int64, next PrepareLanesHandler) *Proposal
 
-		// ProcessLane verifies this lane's portion of a proposed block.
+		// ProcessLaneBasic validates that transactions belonging to this lane are not misplaced
+		// in the block proposal.
+		ProcessLaneBasic(txs [][]byte) error
+
+		// ProcessLane verifies this lane's portion of a proposed block. It inputs the transactions
+		// that may belong to this lane and a function to call the next lane in the chain. The next
+		// lane in the chain will be called with the updated context and filtered down transactions.
 		ProcessLane(ctx sdk.Context, proposalTxs [][]byte, next ProcessLanesHandler) (sdk.Context, error)
+
+		// SetAnteHandler sets the lane's antehandler.
+		SetAnteHandler(antehander sdk.AnteHandler)
+
+		// Logger returns the lane's logger.
+		Logger() log.Logger
+
+		// GetMaxBlockSpace returns the max block space for the lane as a relative percentage.
+		GetMaxBlockSpace() sdk.Dec
 	}
 )
 
@@ -87,11 +103,33 @@ func NewBaseLaneConfig(logger log.Logger, txEncoder sdk.TxEncoder, txDecoder sdk
 	}
 }
 
+// ValidateBasic validates the lane configuration.
+func (c *BaseLaneConfig) ValidateBasic() error {
+	if c.Logger == nil {
+		return fmt.Errorf("logger cannot be nil")
+	}
+
+	if c.TxEncoder == nil {
+		return fmt.Errorf("tx encoder cannot be nil")
+	}
+
+	if c.TxDecoder == nil {
+		return fmt.Errorf("tx decoder cannot be nil")
+	}
+
+	if c.MaxBlockSpace.IsNil() || c.MaxBlockSpace.IsNegative() || c.MaxBlockSpace.GT(sdk.OneDec()) {
+		return fmt.Errorf("max block space must be set to a value between 0 and 1")
+	}
+
+	return nil
+}
+
+// NewProposal returns a new empty proposal.
 func NewProposal(maxTxBytes int64) *Proposal {
 	return &Proposal{
-		Txs:         make([][]byte, 0),
-		SelectedTxs: make(map[string]struct{}),
-		MaxTxBytes:  maxTxBytes,
+		Txs:        make([][]byte, 0),
+		Cache:      make(map[string]struct{}),
+		MaxTxBytes: maxTxBytes,
 	}
 }
 
@@ -104,7 +142,7 @@ func (p *Proposal) UpdateProposal(txs [][]byte, totalSize int64) *Proposal {
 		txHash := sha256.Sum256(tx)
 		txHashStr := hex.EncodeToString(txHash[:])
 
-		p.SelectedTxs[txHashStr] = struct{}{}
+		p.Cache[txHashStr] = struct{}{}
 	}
 
 	return p

blockbuster/lanes/auction/abci.go

@@ -2,17 +2,22 @@ package auction
 
 import (
 	"bytes"
-	"errors"
 	"fmt"
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/skip-mev/pob/blockbuster"
+	"github.com/skip-mev/pob/blockbuster/utils"
 )
 
 // PrepareLane will attempt to select the highest bid transaction that is valid
 // and whose bundled transactions are valid and include them in the proposal. It
 // will return an empty partial proposal if no valid bids are found.
-func (l *TOBLane) PrepareLane(ctx sdk.Context, proposal *blockbuster.Proposal, next blockbuster.PrepareLanesHandler) *blockbuster.Proposal {
+func (l *TOBLane) PrepareLane(
+	ctx sdk.Context,
+	proposal *blockbuster.Proposal,
+	maxTxBytes int64,
+	next blockbuster.PrepareLanesHandler,
+) *blockbuster.Proposal {
 	// Define all of the info we need to select transactions for the partial proposal.
 	var (
 		totalSize   int64
@@ -20,10 +25,6 @@ func (l *TOBLane) PrepareLane(ctx sdk.Context, proposal *blockbuster.Proposal, n
 		txsToRemove = make(map[sdk.Tx]struct{}, 0)
 	)
 
-	// Calculate the max tx bytes for the lane and track the total size of the
-	// transactions we have selected so far.
-	maxTxBytes := blockbuster.GetMaxTxBytesForLane(proposal, l.cfg.MaxBlockSpace)
-
 	// Attempt to select the highest bid transaction that is valid and whose
 	// bundled transactions are valid.
 	bidTxIterator := l.Select(ctx, nil)
@@ -32,19 +33,14 @@ selectBidTxLoop:
 		cacheCtx, write := ctx.CacheContext()
 		tmpBidTx := bidTxIterator.Tx()
 
-		// if the transaction is already in the (partial) block proposal, we skip it.
-		txHash, err := blockbuster.GetTxHashStr(l.cfg.TxEncoder, tmpBidTx)
+		bidTxBz, txHash, err := utils.GetTxHashStr(l.Cfg.TxEncoder, tmpBidTx)
 		if err != nil {
 			txsToRemove[tmpBidTx] = struct{}{}
 			continue
 		}
-		if _, ok := proposal.SelectedTxs[txHash]; ok {
-			continue selectBidTxLoop
-		}
 
-		bidTxBz, err := l.cfg.TxEncoder(tmpBidTx)
-		if err != nil {
-			txsToRemove[tmpBidTx] = struct{}{}
+		// if the transaction is already in the (partial) block proposal, we skip it.
+		if _, ok := proposal.Cache[txHash]; ok {
 			continue selectBidTxLoop
 		}
 
@@ -75,19 +71,14 @@ selectBidTxLoop:
 					continue selectBidTxLoop
 				}
 
-				sdkTxBz, err := l.cfg.TxEncoder(sdkTx)
+				sdkTxBz, hash, err := utils.GetTxHashStr(l.Cfg.TxEncoder, sdkTx)
 				if err != nil {
 					txsToRemove[tmpBidTx] = struct{}{}
 					continue selectBidTxLoop
 				}
 
 				// if the transaction is already in the (partial) block proposal, we skip it.
-				hash, err := blockbuster.GetTxHashStr(l.cfg.TxEncoder, sdkTx)
-				if err != nil {
-					txsToRemove[tmpBidTx] = struct{}{}
-					continue selectBidTxLoop
-				}
-				if _, ok := proposal.SelectedTxs[hash]; ok {
+				if _, ok := proposal.Cache[hash]; ok {
 					continue selectBidTxLoop
 				}
 
@@ -112,7 +103,7 @@ selectBidTxLoop:
 		}
 
 		txsToRemove[tmpBidTx] = struct{}{}
-		l.cfg.Logger.Info(
+		l.Cfg.Logger.Info(
 			"failed to select auction bid tx; tx size is too large",
 			"tx_size", bidTxSize,
 			"max_size", proposal.MaxTxBytes,
@@ -120,8 +111,8 @@ selectBidTxLoop:
 	}
 
 	// Remove all transactions that were invalid during the creation of the partial proposal.
-	if err := blockbuster.RemoveTxsFromLane(txsToRemove, l.Mempool); err != nil {
-		l.cfg.Logger.Error("failed to remove txs from mempool", "lane", l.Name(), "err", err)
+	if err := utils.RemoveTxsFromLane(txsToRemove, l.Mempool); err != nil {
+		l.Cfg.Logger.Error("failed to remove txs from mempool", "lane", l.Name(), "err", err)
 		return proposal
 	}
 
@@ -132,68 +123,99 @@ selectBidTxLoop:
 }
 
 // ProcessLane will ensure that block proposals that include transactions from
-// the top-of-block auction lane are valid. It will return an error if the
-// block proposal is invalid. The block proposal is invalid if it does not
-// respect the ordering of transactions in the bid transaction or if the bid/bundled
-// transactions are invalid.
+// the top-of-block auction lane are valid.
 func (l *TOBLane) ProcessLane(ctx sdk.Context, proposalTxs [][]byte, next blockbuster.ProcessLanesHandler) (sdk.Context, error) {
-	// Track the index of the first transaction that does not belong to this lane.
-	endIndex := 0
+	tx, err := l.Cfg.TxDecoder(proposalTxs[0])
+	if err != nil {
+		return ctx, fmt.Errorf("failed to decode tx in lane %s: %w", l.Name(), err)
+	}
 
-	for index, txBz := range proposalTxs {
-		tx, err := l.cfg.TxDecoder(txBz)
+	if !l.Match(tx) {
+		return next(ctx, proposalTxs)
+	}
+
+	bidInfo, err := l.GetAuctionBidInfo(tx)
+	if err != nil {
+		return ctx, fmt.Errorf("failed to get bid info for lane %s: %w", l.Name(), err)
+	}
+
+	if err := l.VerifyTx(ctx, tx); err != nil {
+		return ctx, fmt.Errorf("invalid bid tx: %w", err)
+	}
+
+	return next(ctx, proposalTxs[len(bidInfo.Transactions)+1:])
+}
+
+// ProcessLaneBasic ensures that if a bid transaction is present in a proposal,
+//   - it is the first transaction in the partial proposal
+//   - all of the bundled transactions are included after the bid transaction in the order
+//     they were included in the bid transaction.
+//   - there are no other bid transactions in the proposal
+func (l *TOBLane) ProcessLaneBasic(txs [][]byte) error {
+	tx, err := l.Cfg.TxDecoder(txs[0])
+	if err != nil {
+		return fmt.Errorf("failed to decode tx in lane %s: %w", l.Name(), err)
+	}
+
+	// If there is a bid transaction, it must be the first transaction in the block proposal.
+	if !l.Match(tx) {
+		for _, txBz := range txs[1:] {
+			tx, err := l.Cfg.TxDecoder(txBz)
+			if err != nil {
+				return fmt.Errorf("failed to decode tx in lane %s: %w", l.Name(), err)
+			}
+
+			if l.Match(tx) {
+				return fmt.Errorf("misplaced bid transactions in lane %s", l.Name())
+			}
+		}
+
+		return nil
+	}
+
+	bidInfo, err := l.GetAuctionBidInfo(tx)
+	if err != nil {
+		return fmt.Errorf("failed to get bid info for lane %s: %w", l.Name(), err)
+	}
+
+	if len(txs) < len(bidInfo.Transactions)+1 {
+		return fmt.Errorf("invalid number of transactions in lane %s; expected at least %d, got %d", l.Name(), len(bidInfo.Transactions)+1, len(txs))
+	}
+
+	// Ensure that the order of transactions in the bundle is preserved.
+	for i, bundleTxBz := range txs[1 : len(bidInfo.Transactions)+1] {
+		tx, err := l.WrapBundleTransaction(bundleTxBz)
 		if err != nil {
-			return ctx, err
+			return fmt.Errorf("failed to decode bundled tx in lane %s: %w", l.Name(), err)
 		}
 
 		if l.Match(tx) {
-			// If the transaction is an auction bid, then we need to ensure that it is
-			// the first transaction in the block proposal and that the order of
-			// transactions in the block proposal follows the order of transactions in
-			// the bid.
-			if index != 0 {
-				return ctx, fmt.Errorf("block proposal did not place auction bid transaction at the top of the lane: %d", index)
-			}
+			return fmt.Errorf("multiple bid transactions in lane %s", l.Name())
+		}
 
-			bidInfo, err := l.GetAuctionBidInfo(tx)
-			if err != nil {
-				return ctx, fmt.Errorf("failed to get auction bid info for tx at index %w", err)
-			}
+		txBz, err := l.Cfg.TxEncoder(tx)
+		if err != nil {
+			return fmt.Errorf("failed to encode bundled tx in lane %s: %w", l.Name(), err)
+		}
 
-			if bidInfo != nil {
-				if len(proposalTxs) < len(bidInfo.Transactions)+1 {
-					return ctx, errors.New("block proposal does not contain enough transactions to match the bundled transactions in the auction bid")
-				}
-
-				for i, refTxRaw := range bidInfo.Transactions {
-					// Wrap and then encode the bundled transaction to ensure that the underlying
-					// reference transaction can be processed as an sdk.Tx.
-					wrappedTx, err := l.WrapBundleTransaction(refTxRaw)
-					if err != nil {
-						return ctx, err
-					}
-
-					refTxBz, err := l.cfg.TxEncoder(wrappedTx)
-					if err != nil {
-						return ctx, err
-					}
-
-					if !bytes.Equal(refTxBz, proposalTxs[i+1]) {
-						return ctx, errors.New("block proposal does not match the bundled transactions in the auction bid")
-					}
-				}
-
-				// Verify the bid transaction.
-				if err = l.VerifyTx(ctx, tx); err != nil {
-					return ctx, err
-				}
-
-				endIndex = len(bidInfo.Transactions) + 1
-			}
+		if !bytes.Equal(txBz, bidInfo.Transactions[i]) {
+			return fmt.Errorf("invalid order of transactions in lane %s", l.Name())
 		}
 	}
 
-	return next(ctx, proposalTxs[endIndex:])
+	// Ensure that there are no more bid transactions in the block proposal.
+	for _, txBz := range txs[len(bidInfo.Transactions)+1:] {
+		tx, err := l.Cfg.TxDecoder(txBz)
+		if err != nil {
+			return fmt.Errorf("failed to decode tx in lane %s: %w", l.Name(), err)
+		}
+
+		if l.Match(tx) {
+			return fmt.Errorf("multiple bid transactions in lane %s", l.Name())
+		}
+	}
+
+	return nil
 }
 
 // VerifyTx will verify that the bid transaction and all of its bundled
@@ -235,8 +257,8 @@ func (l *TOBLane) VerifyTx(ctx sdk.Context, bidTx sdk.Tx) error {
 // verifyTx will execute the ante handler on the transaction and return the
 // resulting context and error.
 func (l *TOBLane) verifyTx(ctx sdk.Context, tx sdk.Tx) (sdk.Context, error) {
-	if l.cfg.AnteHandler != nil {
-		newCtx, err := l.cfg.AnteHandler(ctx, tx, false)
+	if l.Cfg.AnteHandler != nil {
+		newCtx, err := l.Cfg.AnteHandler(ctx, tx, false)
 		return newCtx, err
 	}
 

blockbuster/lanes/auction/auction_test.go

@@ -0,0 +1,47 @@
+package auction_test
+
+import (
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/cometbft/cometbft/libs/log"
+	cmtproto "github.com/cometbft/cometbft/proto/tendermint/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/skip-mev/pob/blockbuster/lanes/auction"
+	testutils "github.com/skip-mev/pob/testutils"
+	"github.com/stretchr/testify/suite"
+)
+
+type IntegrationTestSuite struct {
+	suite.Suite
+
+	encCfg   testutils.EncodingConfig
+	config   auction.Factory
+	mempool  auction.Mempool
+	ctx      sdk.Context
+	random   *rand.Rand
+	accounts []testutils.Account
+	nonces   map[string]uint64
+}
+
+func TestMempoolTestSuite(t *testing.T) {
+	suite.Run(t, new(IntegrationTestSuite))
+}
+
+func (suite *IntegrationTestSuite) SetupTest() {
+	// Mempool setup
+	suite.encCfg = testutils.CreateTestEncodingConfig()
+	suite.config = auction.NewDefaultAuctionFactory(suite.encCfg.TxConfig.TxDecoder())
+	suite.mempool = auction.NewMempool(suite.encCfg.TxConfig.TxEncoder(), 0, suite.config)
+	suite.ctx = sdk.NewContext(nil, cmtproto.Header{}, false, log.NewNopLogger())
+
+	// Init accounts
+	suite.random = rand.New(rand.NewSource(time.Now().Unix()))
+	suite.accounts = testutils.RandomAccounts(suite.random, 10)
+
+	suite.nonces = make(map[string]uint64)
+	for _, acc := range suite.accounts {
+		suite.nonces[acc.Address.String()] = 0
+	}
+}

blockbuster/lanes/auction/factory.go

@@ -5,18 +5,10 @@ import (
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/cosmos/cosmos-sdk/x/auth/signing"
+	"github.com/skip-mev/pob/x/builder/types"
 )
 
 type (
-	// BidInfo defines the information about a bid to the auction house.
-	BidInfo struct {
-		Bidder       sdk.AccAddress
-		Bid          sdk.Coin
-		Transactions [][]byte
-		Timeout      uint64
-		Signers      []map[string]struct{}
-	}
-
 	// Factory defines the interface for processing auction transactions. It is
 	// a wrapper around all of the functionality that each application chain must implement
 	// in order for auction processing to work.
@@ -28,7 +20,7 @@ type (
 		WrapBundleTransaction(tx []byte) (sdk.Tx, error)
 
 		// GetAuctionBidInfo defines a function that returns the bid info from an auction transaction.
-		GetAuctionBidInfo(tx sdk.Tx) (*BidInfo, error)
+		GetAuctionBidInfo(tx sdk.Tx) (*types.BidInfo, error)
 	}
 
 	// DefaultAuctionFactory defines a default implmentation for the auction factory interface for processing auction transactions.
@@ -65,7 +57,7 @@ func (config *DefaultAuctionFactory) WrapBundleTransaction(tx []byte) (sdk.Tx, e
 // GetAuctionBidInfo defines a default function that returns the auction bid info from
 // an auction transaction. In the default case, the auction bid info is stored in the
 // MsgAuctionBid message.
-func (config *DefaultAuctionFactory) GetAuctionBidInfo(tx sdk.Tx) (*BidInfo, error) {
+func (config *DefaultAuctionFactory) GetAuctionBidInfo(tx sdk.Tx) (*types.BidInfo, error) {
 	msg, err := GetMsgAuctionBidFromTx(tx)
 	if err != nil {
 		return nil, err
@@ -90,7 +82,7 @@ func (config *DefaultAuctionFactory) GetAuctionBidInfo(tx sdk.Tx) (*BidInfo, err
 		return nil, err
 	}
 
-	return &BidInfo{
+	return &types.BidInfo{
 		Bid:          msg.Bid,
 		Bidder:       bidder,
 		Transactions: msg.Transactions,

blockbuster/lanes/auction/factory_test.go

@@ -0,0 +1,525 @@
+package auction_test
+
+import (
+	"crypto/rand"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	testutils "github.com/skip-mev/pob/testutils"
+)
+
+func (suite *IntegrationTestSuite) TestIsAuctionTx() {
+	testCases := []struct {
+		name          string
+		createTx      func() sdk.Tx
+		isAuctionTx   bool
+		expectedError bool
+	}{
+		{
+			"normal sdk tx",
+			func() sdk.Tx {
+				tx, err := testutils.CreateRandomTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 2, 0)
+				suite.Require().NoError(err)
+				return tx
+			},
+			false,
+			false,
+		},
+		{
+			"malformed auction bid tx",
+			func() sdk.Tx {
+				msgAuctionBid, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				msgs := testutils.CreateRandomMsgs(suite.accounts[0].Address, 2)
+				msgs = append(msgs, msgAuctionBid)
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 0, msgs)
+				suite.Require().NoError(err)
+				return tx
+			},
+			false,
+			true,
+		},
+		{
+			"valid auction bid tx",
+			func() sdk.Tx {
+				msgAuctionBid, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				msgs := []sdk.Msg{msgAuctionBid}
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 0, msgs)
+				suite.Require().NoError(err)
+				return tx
+			},
+			true,
+			false,
+		},
+		{
+			"tx with multiple MsgAuctionBid messages",
+			func() sdk.Tx {
+				bid1, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				bid2, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 1, 2)
+				suite.Require().NoError(err)
+
+				msgs := []sdk.Msg{bid1, bid2}
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 0, msgs)
+				suite.Require().NoError(err)
+				return tx
+			},
+			false,
+			true,
+		},
+	}
+
+	for _, tc := range testCases {
+		suite.Run(tc.name, func() {
+			tx := tc.createTx()
+
+			bidInfo, err := suite.config.GetAuctionBidInfo(tx)
+
+			suite.Require().Equal(tc.isAuctionTx, bidInfo != nil)
+			if tc.expectedError {
+				suite.Require().Error(err)
+			} else {
+				suite.Require().NoError(err)
+			}
+		})
+	}
+}
+
+func (suite *IntegrationTestSuite) TestGetTransactionSigners() {
+	testCases := []struct {
+		name            string
+		createTx        func() sdk.Tx
+		expectedSigners []map[string]struct{}
+		expectedError   bool
+	}{
+		{
+			"normal auction tx",
+			func() sdk.Tx {
+				tx, err := testutils.CreateAuctionTxWithSigners(
+					suite.encCfg.TxConfig,
+					suite.accounts[0],
+					sdk.NewCoin("foo", sdk.NewInt(100)),
+					1,
+					0,
+					suite.accounts[0:1],
+				)
+				suite.Require().NoError(err)
+
+				return tx
+			},
+			[]map[string]struct{}{
+				{
+					suite.accounts[0].Address.String(): {},
+				},
+			},
+			false,
+		},
+		{
+			"normal sdk tx",
+			func() sdk.Tx {
+				tx, err := testutils.CreateRandomTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 10, 0)
+				suite.Require().NoError(err)
+
+				return tx
+			},
+			nil,
+			true,
+		},
+		{
+			"multiple signers on auction tx",
+			func() sdk.Tx {
+				tx, err := testutils.CreateAuctionTxWithSigners(
+					suite.encCfg.TxConfig,
+					suite.accounts[0],
+					sdk.NewCoin("foo", sdk.NewInt(100)),
+					1,
+					0,
+					suite.accounts[0:3],
+				)
+				suite.Require().NoError(err)
+
+				return tx
+			},
+			[]map[string]struct{}{
+				{
+					suite.accounts[0].Address.String(): {},
+				},
+				{
+					suite.accounts[1].Address.String(): {},
+				},
+				{
+					suite.accounts[2].Address.String(): {},
+				},
+			},
+			false,
+		},
+	}
+
+	for _, tc := range testCases {
+		suite.Run(tc.name, func() {
+			tx := tc.createTx()
+
+			bidInfo, _ := suite.config.GetAuctionBidInfo(tx)
+			if tc.expectedError {
+				suite.Require().Nil(bidInfo)
+			} else {
+				suite.Require().Equal(tc.expectedSigners, bidInfo.Signers)
+			}
+		})
+	}
+}
+
+func (suite *IntegrationTestSuite) TestWrapBundleTransaction() {
+	testCases := []struct {
+		name           string
+		createBundleTx func() (sdk.Tx, []byte)
+		expectedError  bool
+	}{
+		{
+			"normal sdk tx",
+			func() (sdk.Tx, []byte) {
+				tx, err := testutils.CreateRandomTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 1, 0)
+				suite.Require().NoError(err)
+
+				bz, err := suite.encCfg.TxConfig.TxEncoder()(tx)
+				suite.Require().NoError(err)
+
+				return tx, bz
+			},
+			false,
+		},
+		{
+			"random bytes with expected failure",
+			func() (sdk.Tx, []byte) {
+				bz := make([]byte, 100)
+				rand.Read(bz)
+
+				return nil, bz
+			},
+			true,
+		},
+	}
+
+	for _, tc := range testCases {
+		suite.Run(tc.name, func() {
+			tx, bz := tc.createBundleTx()
+
+			wrappedTx, err := suite.config.WrapBundleTransaction(bz)
+			if tc.expectedError {
+				suite.Require().Error(err)
+			} else {
+				suite.Require().NoError(err)
+
+				txBytes, err := suite.encCfg.TxConfig.TxEncoder()(tx)
+				suite.Require().NoError(err)
+
+				wrappedTxBytes, err := suite.encCfg.TxConfig.TxEncoder()(wrappedTx)
+				suite.Require().NoError(err)
+
+				suite.Require().Equal(txBytes, wrappedTxBytes)
+			}
+		})
+	}
+}
+
+func (suite *IntegrationTestSuite) TestGetBidder() {
+	testCases := []struct {
+		name           string
+		createTx       func() sdk.Tx
+		expectedBidder string
+		expectedError  bool
+		isAuctionTx    bool
+	}{
+		{
+			"normal sdk tx",
+			func() sdk.Tx {
+				tx, err := testutils.CreateRandomTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 1, 0)
+				suite.Require().NoError(err)
+
+				return tx
+			},
+			"",
+			false,
+			false,
+		},
+		{
+			"valid auction tx",
+			func() sdk.Tx {
+				msgAuctionBid, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				msgs := []sdk.Msg{msgAuctionBid}
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 0, msgs)
+				suite.Require().NoError(err)
+				return tx
+			},
+			suite.accounts[0].Address.String(),
+			false,
+			true,
+		},
+		{
+			"invalid auction tx",
+			func() sdk.Tx {
+				msgAuctionBid, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				randomMsg := testutils.CreateRandomMsgs(suite.accounts[0].Address, 1)[0]
+				suite.Require().NoError(err)
+
+				msgs := []sdk.Msg{msgAuctionBid, randomMsg}
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 0, msgs)
+				suite.Require().NoError(err)
+				return tx
+			},
+			"",
+			true,
+			false,
+		},
+	}
+
+	for _, tc := range testCases {
+		suite.Run(tc.name, func() {
+			tx := tc.createTx()
+
+			bidInfo, err := suite.config.GetAuctionBidInfo(tx)
+			if tc.expectedError {
+				suite.Require().Error(err)
+			} else {
+				suite.Require().NoError(err)
+
+				if tc.isAuctionTx {
+					suite.Require().Equal(tc.expectedBidder, bidInfo.Bidder.String())
+				}
+			}
+		})
+	}
+}
+
+func (suite *IntegrationTestSuite) TestGetBid() {
+	testCases := []struct {
+		name          string
+		createTx      func() sdk.Tx
+		expectedBid   sdk.Coin
+		expectedError bool
+		isAuctionTx   bool
+	}{
+		{
+			"normal sdk tx",
+			func() sdk.Tx {
+				tx, err := testutils.CreateRandomTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 1, 0)
+				suite.Require().NoError(err)
+
+				return tx
+			},
+			sdk.Coin{},
+			false,
+			false,
+		},
+		{
+			"valid auction tx",
+			func() sdk.Tx {
+				msgAuctionBid, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				msgs := []sdk.Msg{msgAuctionBid}
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 0, msgs)
+				suite.Require().NoError(err)
+				return tx
+			},
+			sdk.NewInt64Coin("foo", 100),
+			false,
+			true,
+		},
+		{
+			"invalid auction tx",
+			func() sdk.Tx {
+				msgAuctionBid, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				randomMsg := testutils.CreateRandomMsgs(suite.accounts[0].Address, 1)[0]
+				suite.Require().NoError(err)
+
+				msgs := []sdk.Msg{msgAuctionBid, randomMsg}
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 0, msgs)
+				suite.Require().NoError(err)
+				return tx
+			},
+			sdk.Coin{},
+			true,
+			false,
+		},
+	}
+
+	for _, tc := range testCases {
+		suite.Run(tc.name, func() {
+			tx := tc.createTx()
+
+			bidInfo, err := suite.config.GetAuctionBidInfo(tx)
+			if tc.expectedError {
+				suite.Require().Error(err)
+			} else {
+				suite.Require().NoError(err)
+
+				if tc.isAuctionTx {
+					suite.Require().Equal(tc.expectedBid, bidInfo.Bid)
+				}
+			}
+		})
+	}
+}
+
+func (suite *IntegrationTestSuite) TestGetBundledTransactions() {
+	testCases := []struct {
+		name          string
+		createTx      func() (sdk.Tx, [][]byte)
+		expectedError bool
+		isAuctionTx   bool
+	}{
+		{
+			"normal sdk tx",
+			func() (sdk.Tx, [][]byte) {
+				tx, err := testutils.CreateRandomTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 1, 0)
+				suite.Require().NoError(err)
+
+				return tx, nil
+			},
+			false,
+			false,
+		},
+		{
+			"valid auction tx",
+			func() (sdk.Tx, [][]byte) {
+				msgAuctionBid, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				msgs := []sdk.Msg{msgAuctionBid}
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 0, msgs)
+				suite.Require().NoError(err)
+				return tx, msgAuctionBid.Transactions
+			},
+			false,
+			true,
+		},
+		{
+			"invalid auction tx",
+			func() (sdk.Tx, [][]byte) {
+				msgAuctionBid, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				randomMsg := testutils.CreateRandomMsgs(suite.accounts[0].Address, 1)[0]
+				suite.Require().NoError(err)
+
+				msgs := []sdk.Msg{msgAuctionBid, randomMsg}
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 0, msgs)
+				suite.Require().NoError(err)
+				return tx, nil
+			},
+			true,
+			false,
+		},
+	}
+
+	for _, tc := range testCases {
+		suite.Run(tc.name, func() {
+			tx, expectedBundledTxs := tc.createTx()
+
+			bidInfo, err := suite.config.GetAuctionBidInfo(tx)
+			if tc.expectedError {
+				suite.Require().Error(err)
+			} else {
+				suite.Require().NoError(err)
+
+				if tc.isAuctionTx {
+					suite.Require().Equal(expectedBundledTxs, bidInfo.Transactions)
+				}
+			}
+		})
+	}
+}
+
+func (suite *IntegrationTestSuite) TestGetTimeout() {
+	testCases := []struct {
+		name            string
+		createTx        func() sdk.Tx
+		expectedError   bool
+		isAuctionTx     bool
+		expectedTimeout uint64
+	}{
+		{
+			"normal sdk tx",
+			func() sdk.Tx {
+				tx, err := testutils.CreateRandomTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 1, 1)
+				suite.Require().NoError(err)
+
+				return tx
+			},
+			false,
+			false,
+			1,
+		},
+		{
+			"valid auction tx",
+			func() sdk.Tx {
+				msgAuctionBid, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				msgs := []sdk.Msg{msgAuctionBid}
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 10, msgs)
+				suite.Require().NoError(err)
+				return tx
+			},
+			false,
+			true,
+			10,
+		},
+		{
+			"invalid auction tx",
+			func() sdk.Tx {
+				msgAuctionBid, err := testutils.CreateMsgAuctionBid(suite.encCfg.TxConfig, suite.accounts[0], sdk.NewInt64Coin("foo", 100), 0, 2)
+				suite.Require().NoError(err)
+
+				randomMsg := testutils.CreateRandomMsgs(suite.accounts[0].Address, 1)[0]
+				suite.Require().NoError(err)
+
+				msgs := []sdk.Msg{msgAuctionBid, randomMsg}
+
+				tx, err := testutils.CreateTx(suite.encCfg.TxConfig, suite.accounts[0], 0, 10, msgs)
+				suite.Require().NoError(err)
+				return tx
+			},
+			true,
+			false,
+			10,
+		},
+	}
+
+	for _, tc := range testCases {
+		suite.Run(tc.name, func() {
+			tx := tc.createTx()
+
+			bidInfo, err := suite.config.GetAuctionBidInfo(tx)
+			if tc.expectedError {
+				suite.Require().Error(err)
+			} else {
+				suite.Require().NoError(err)
+
+				if tc.isAuctionTx {
+					suite.Require().Equal(tc.expectedTimeout, bidInfo.Timeout)
+				}
+			}
+		})
+	}
+}

blockbuster/lanes/auction/lane.go

@@ -1,17 +1,20 @@
 package auction
 
 import (
-	"github.com/cometbft/cometbft/libs/log"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/skip-mev/pob/blockbuster"
+	"github.com/skip-mev/pob/blockbuster/lanes/base"
 )
 
 const (
 	// LaneName defines the name of the top-of-block auction lane.
-	LaneName = "tob"
+	LaneName = "top-of-block"
 )
 
-var _ blockbuster.Lane = (*TOBLane)(nil)
+var (
+	_ blockbuster.Lane = (*TOBLane)(nil)
+	_ Factory          = (*TOBLane)(nil)
+)
 
 // TOBLane defines a top-of-block auction lane. The top of block auction lane
 // hosts transactions that want to bid for inclusion at the top of the next block.
@@ -24,7 +27,7 @@ type TOBLane struct {
 	Mempool
 
 	// LaneConfig defines the base lane configuration.
-	cfg blockbuster.BaseLaneConfig
+	*base.DefaultLane
 
 	// Factory defines the API/functionality which is responsible for determining
 	// if a transaction is a bid transaction and how to extract relevant
@@ -34,18 +37,18 @@ type TOBLane struct {
 
 // NewTOBLane returns a new TOB lane.
 func NewTOBLane(
-	logger log.Logger,
-	txDecoder sdk.TxDecoder,
-	txEncoder sdk.TxEncoder,
+	cfg blockbuster.BaseLaneConfig,
 	maxTx int,
-	anteHandler sdk.AnteHandler,
 	af Factory,
-	maxBlockSpace sdk.Dec,
 ) *TOBLane {
+	if err := cfg.ValidateBasic(); err != nil {
+		panic(err)
+	}
+
 	return &TOBLane{
-		Mempool: NewMempool(txEncoder, maxTx, af),
-		cfg:     blockbuster.NewBaseLaneConfig(logger, txEncoder, txDecoder, anteHandler, maxBlockSpace),
-		Factory: af,
+		Mempool:     NewMempool(cfg.TxEncoder, maxTx, af),
+		DefaultLane: base.NewDefaultLane(cfg),
+		Factory:     af,
 	}
 }
 

blockbuster/lanes/auction/mempool.go

@@ -8,7 +8,7 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkmempool "github.com/cosmos/cosmos-sdk/types/mempool"
 	"github.com/skip-mev/pob/blockbuster"
-	"github.com/skip-mev/pob/mempool"
+	"github.com/skip-mev/pob/blockbuster/utils"
 )
 
 var _ Mempool = (*TOBMempool)(nil)
@@ -48,8 +48,8 @@ type (
 
 // TxPriority returns a TxPriority over auction bid transactions only. It
 // is to be used in the auction index only.
-func TxPriority(config Factory) mempool.TxPriority[string] {
-	return mempool.TxPriority[string]{
+func TxPriority(config Factory) blockbuster.TxPriority[string] {
+	return blockbuster.TxPriority[string]{
 		GetTxPriority: func(goCtx context.Context, tx sdk.Tx) string {
 			bidInfo, err := config.GetAuctionBidInfo(tx)
 			if err != nil {
@@ -92,8 +92,8 @@ func TxPriority(config Factory) mempool.TxPriority[string] {
 // NewMempool returns a new auction mempool.
 func NewMempool(txEncoder sdk.TxEncoder, maxTx int, config Factory) *TOBMempool {
 	return &TOBMempool{
-		index: mempool.NewPriorityMempool(
-			mempool.PriorityNonceMempoolConfig[string]{
+		index: blockbuster.NewPriorityMempool(
+			blockbuster.PriorityNonceMempoolConfig[string]{
 				TxPriority: TxPriority(config),
 				MaxTx:      maxTx,
 			},
@@ -120,7 +120,7 @@ func (am *TOBMempool) Insert(ctx context.Context, tx sdk.Tx) error {
 		return fmt.Errorf("failed to insert tx into auction index: %w", err)
 	}
 
-	txHashStr, err := blockbuster.GetTxHashStr(am.txEncoder, tx)
+	_, txHashStr, err := utils.GetTxHashStr(am.txEncoder, tx)
 	if err != nil {
 		return err
 	}
@@ -167,7 +167,7 @@ func (am *TOBMempool) CountTx() int {
 
 // Contains returns true if the transaction is contained in the mempool.
 func (am *TOBMempool) Contains(tx sdk.Tx) (bool, error) {
-	txHashStr, err := blockbuster.GetTxHashStr(am.txEncoder, tx)
+	_, txHashStr, err := utils.GetTxHashStr(am.txEncoder, tx)
 	if err != nil {
 		return false, fmt.Errorf("failed to get tx hash string: %w", err)
 	}
@@ -181,7 +181,7 @@ func (am *TOBMempool) removeTx(mp sdkmempool.Mempool, tx sdk.Tx) {
 		panic(fmt.Errorf("failed to remove invalid transaction from the mempool: %w", err))
 	}
 
-	txHashStr, err := blockbuster.GetTxHashStr(am.txEncoder, tx)
+	_, txHashStr, err := utils.GetTxHashStr(am.txEncoder, tx)
 	if err != nil {
 		panic(fmt.Errorf("failed to get tx hash string: %w", err))
 	}

blockbuster/lanes/auction/utils_test.go

@@ -0,0 +1,48 @@
+package auction_test
+
+import (
+	"testing"
+
+	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
+	"github.com/skip-mev/pob/blockbuster/lanes/auction"
+	pobcodec "github.com/skip-mev/pob/codec"
+	buildertypes "github.com/skip-mev/pob/x/builder/types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetMsgAuctionBidFromTx_Valid(t *testing.T) {
+	encCfg := pobcodec.CreateEncodingConfig()
+
+	txBuilder := encCfg.TxConfig.NewTxBuilder()
+	txBuilder.SetMsgs(&buildertypes.MsgAuctionBid{})
+
+	msg, err := auction.GetMsgAuctionBidFromTx(txBuilder.GetTx())
+	require.NoError(t, err)
+	require.NotNil(t, msg)
+}
+
+func TestGetMsgAuctionBidFromTx_MultiMsgBid(t *testing.T) {
+	encCfg := pobcodec.CreateEncodingConfig()
+
+	txBuilder := encCfg.TxConfig.NewTxBuilder()
+	txBuilder.SetMsgs(
+		&buildertypes.MsgAuctionBid{},
+		&buildertypes.MsgAuctionBid{},
+		&banktypes.MsgSend{},
+	)
+
+	msg, err := auction.GetMsgAuctionBidFromTx(txBuilder.GetTx())
+	require.Error(t, err)
+	require.Nil(t, msg)
+}
+
+func TestGetMsgAuctionBidFromTx_NoBid(t *testing.T) {
+	encCfg := pobcodec.CreateEncodingConfig()
+
+	txBuilder := encCfg.TxConfig.NewTxBuilder()
+	txBuilder.SetMsgs(&banktypes.MsgSend{})
+
+	msg, err := auction.GetMsgAuctionBidFromTx(txBuilder.GetTx())
+	require.NoError(t, err)
+	require.Nil(t, msg)
+}

blockbuster/lanes/base/abci.go

@@ -5,37 +5,36 @@ import (
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/skip-mev/pob/blockbuster"
+	"github.com/skip-mev/pob/blockbuster/utils"
 )
 
 // PrepareLane will prepare a partial proposal for the base lane.
-func (l *DefaultLane) PrepareLane(ctx sdk.Context, proposal *blockbuster.Proposal, next blockbuster.PrepareLanesHandler) *blockbuster.Proposal {
+func (l *DefaultLane) PrepareLane(
+	ctx sdk.Context,
+	proposal *blockbuster.Proposal,
+	maxTxBytes int64,
+	next blockbuster.PrepareLanesHandler,
+) *blockbuster.Proposal {
 	// Define all of the info we need to select transactions for the partial proposal.
-	txs := make([][]byte, 0)
-	txsToRemove := make(map[sdk.Tx]struct{}, 0)
-	totalSize := int64(0)
-
-	// Calculate the max tx bytes for the lane and track the total size of the
-	// transactions we have selected so far.
-	maxTxBytes := blockbuster.GetMaxTxBytesForLane(proposal, l.cfg.MaxBlockSpace)
+	var (
+		totalSize   int64
+		txs         [][]byte
+		txsToRemove = make(map[sdk.Tx]struct{}, 0)
+	)
 
 	// Select all transactions in the mempool that are valid and not already in the
 	// partial proposal.
 	for iterator := l.Mempool.Select(ctx, nil); iterator != nil; iterator = iterator.Next() {
 		tx := iterator.Tx()
 
-		txBytes, err := l.cfg.TxEncoder(tx)
+		txBytes, hash, err := utils.GetTxHashStr(l.Cfg.TxEncoder, tx)
 		if err != nil {
 			txsToRemove[tx] = struct{}{}
 			continue
 		}
 
 		// if the transaction is already in the (partial) block proposal, we skip it.
-		hash, err := blockbuster.GetTxHashStr(l.cfg.TxEncoder, tx)
-		if err != nil {
-			txsToRemove[tx] = struct{}{}
-			continue
-		}
-		if _, ok := proposal.SelectedTxs[hash]; ok {
+		if _, ok := proposal.Cache[hash]; ok {
 			continue
 		}
 
@@ -56,8 +55,8 @@ func (l *DefaultLane) PrepareLane(ctx sdk.Context, proposal *blockbuster.Proposa
 	}
 
 	// Remove all transactions that were invalid during the creation of the partial proposal.
-	if err := blockbuster.RemoveTxsFromLane(txsToRemove, l.Mempool); err != nil {
-		l.cfg.Logger.Error("failed to remove txs from mempool", "lane", l.Name(), "err", err)
+	if err := utils.RemoveTxsFromLane(txsToRemove, l.Mempool); err != nil {
+		l.Cfg.Logger.Error("failed to remove txs from mempool", "lane", l.Name(), "err", err)
 		return proposal
 	}
 
@@ -69,7 +68,7 @@ func (l *DefaultLane) PrepareLane(ctx sdk.Context, proposal *blockbuster.Proposa
 // ProcessLane verifies the default lane's portion of a block proposal.
 func (l *DefaultLane) ProcessLane(ctx sdk.Context, proposalTxs [][]byte, next blockbuster.ProcessLanesHandler) (sdk.Context, error) {
 	for index, tx := range proposalTxs {
-		tx, err := l.cfg.TxDecoder(tx)
+		tx, err := l.Cfg.TxDecoder(tx)
 		if err != nil {
 			return ctx, fmt.Errorf("failed to decode tx: %w", err)
 		}
@@ -87,10 +86,37 @@ func (l *DefaultLane) ProcessLane(ctx sdk.Context, proposalTxs [][]byte, next bl
 	return ctx, nil
 }
 
+// ProcessLaneBasic does basic validation on the block proposal to ensure that
+// transactions that belong to this lane are not misplaced in the block proposal.
+func (l *DefaultLane) ProcessLaneBasic(txs [][]byte) error {
+	seenOtherLaneTx := false
+	lastSeenIndex := 0
+
+	for _, txBz := range txs {
+		tx, err := l.Cfg.TxDecoder(txBz)
+		if err != nil {
+			return fmt.Errorf("failed to decode tx in lane %s: %w", l.Name(), err)
+		}
+
+		if l.Match(tx) {
+			if seenOtherLaneTx {
+				return fmt.Errorf("the %s lane contains a transaction that belongs to another lane", l.Name())
+			}
+
+			lastSeenIndex++
+			continue
+		}
+
+		seenOtherLaneTx = true
+	}
+
+	return nil
+}
+
 // VerifyTx does basic verification of the transaction using the ante handler.
 func (l *DefaultLane) VerifyTx(ctx sdk.Context, tx sdk.Tx) error {
-	if l.cfg.AnteHandler != nil {
-		_, err := l.cfg.AnteHandler(ctx, tx, false)
+	if l.Cfg.AnteHandler != nil {
+		_, err := l.Cfg.AnteHandler(ctx, tx, false)
 		return err
 	}
 

blockbuster/lanes/base/lane.go

@@ -20,13 +20,18 @@ type DefaultLane struct {
 	Mempool
 
 	// LaneConfig defines the base lane configuration.
-	cfg blockbuster.BaseLaneConfig
+	Cfg blockbuster.BaseLaneConfig
 }
 
-func NewDefaultLane(logger log.Logger, txDecoder sdk.TxDecoder, txEncoder sdk.TxEncoder, anteHandler sdk.AnteHandler, maxBlockSpace sdk.Dec) *DefaultLane {
+// NewDefaultLane returns a new default lane.
+func NewDefaultLane(cfg blockbuster.BaseLaneConfig) *DefaultLane {
+	if err := cfg.ValidateBasic(); err != nil {
+		panic(err)
+	}
+
 	return &DefaultLane{
-		Mempool: NewDefaultMempool(txEncoder),
-		cfg:     blockbuster.NewBaseLaneConfig(logger, txEncoder, txDecoder, anteHandler, maxBlockSpace),
+		Mempool: NewDefaultMempool(cfg.TxEncoder),
+		Cfg:     cfg,
 	}
 }
 
@@ -41,3 +46,18 @@ func (l *DefaultLane) Match(sdk.Tx) bool {
 func (l *DefaultLane) Name() string {
 	return LaneName
 }
+
+// Logger returns the lane's logger.
+func (l *DefaultLane) Logger() log.Logger {
+	return l.Cfg.Logger
+}
+
+// SetAnteHandler sets the lane's antehandler.
+func (l *DefaultLane) SetAnteHandler(anteHandler sdk.AnteHandler) {
+	l.Cfg.AnteHandler = anteHandler
+}
+
+// GetMaxBlockSpace returns the maximum block space for the lane as a relative percentage.
+func (l *DefaultLane) GetMaxBlockSpace() sdk.Dec {
+	return l.Cfg.MaxBlockSpace
+}

blockbuster/lanes/base/mempool.go

@@ -8,7 +8,7 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkmempool "github.com/cosmos/cosmos-sdk/types/mempool"
 	"github.com/skip-mev/pob/blockbuster"
-	"github.com/skip-mev/pob/mempool"
+	"github.com/skip-mev/pob/blockbuster/utils"
 )
 
 var _ sdkmempool.Mempool = (*DefaultMempool)(nil)
@@ -42,8 +42,8 @@ type (
 
 func NewDefaultMempool(txEncoder sdk.TxEncoder) *DefaultMempool {
 	return &DefaultMempool{
-		index: mempool.NewPriorityMempool(
-			mempool.DefaultPriorityNonceMempoolConfig(),
+		index: blockbuster.NewPriorityMempool(
+			blockbuster.DefaultPriorityNonceMempoolConfig(),
 		),
 		txEncoder: txEncoder,
 		txIndex:   make(map[string]struct{}),
@@ -56,7 +56,7 @@ func (am *DefaultMempool) Insert(ctx context.Context, tx sdk.Tx) error {
 		return fmt.Errorf("failed to insert tx into auction index: %w", err)
 	}
 
-	txHashStr, err := blockbuster.GetTxHashStr(am.txEncoder, tx)
+	_, txHashStr, err := utils.GetTxHashStr(am.txEncoder, tx)
 	if err != nil {
 		return err
 	}
@@ -82,7 +82,7 @@ func (am *DefaultMempool) CountTx() int {
 
 // Contains returns true if the transaction is contained in the mempool.
 func (am *DefaultMempool) Contains(tx sdk.Tx) (bool, error) {
-	txHashStr, err := blockbuster.GetTxHashStr(am.txEncoder, tx)
+	_, txHashStr, err := utils.GetTxHashStr(am.txEncoder, tx)
 	if err != nil {
 		return false, fmt.Errorf("failed to get tx hash string: %w", err)
 	}
@@ -97,7 +97,7 @@ func (am *DefaultMempool) removeTx(mp sdkmempool.Mempool, tx sdk.Tx) {
 		panic(fmt.Errorf("failed to remove invalid transaction from the mempool: %w", err))
 	}
 
-	txHashStr, err := blockbuster.GetTxHashStr(am.txEncoder, tx)
+	_, txHashStr, err := utils.GetTxHashStr(am.txEncoder, tx)
 	if err != nil {
 		panic(fmt.Errorf("failed to get tx hash string: %w", err))
 	}

blockbuster/lanes/terminator/lane.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/cometbft/cometbft/libs/log"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkmempool "github.com/cosmos/cosmos-sdk/types/mempool"
 	"github.com/skip-mev/pob/blockbuster"
@@ -34,7 +35,7 @@ type Terminator struct{}
 var _ blockbuster.Lane = (*Terminator)(nil)
 
 // PrepareLane is a no-op
-func (t Terminator) PrepareLane(_ sdk.Context, proposal *blockbuster.Proposal, _ blockbuster.PrepareLanesHandler) *blockbuster.Proposal {
+func (t Terminator) PrepareLane(_ sdk.Context, proposal *blockbuster.Proposal, _ int64, _ blockbuster.PrepareLanesHandler) *blockbuster.Proposal {
 	return proposal
 }
 
@@ -82,3 +83,21 @@ func (t Terminator) Remove(sdk.Tx) error {
 func (t Terminator) Select(context.Context, [][]byte) sdkmempool.Iterator {
 	return nil
 }
+
+// ValidateLaneBasic is a no-op
+func (t Terminator) ProcessLaneBasic([][]byte) error {
+	return nil
+}
+
+// SetLaneConfig is a no-op
+func (t Terminator) SetAnteHandler(sdk.AnteHandler) {}
+
+// Logger is a no-op
+func (t Terminator) Logger() log.Logger {
+	return log.NewNopLogger()
+}
+
+// GetMaxBlockSpace is a no-op
+func (t Terminator) GetMaxBlockSpace() sdk.Dec {
+	return sdk.ZeroDec()
+}

blockbuster/mempool.go

@@ -19,6 +19,9 @@ type (
 
 		// Contains returns true if the transaction is contained in the mempool.
 		Contains(tx sdk.Tx) (bool, error)
+
+		// GetTxDistribution returns the number of transactions in each lane.
+		GetTxDistribution() map[string]int
 	}
 
 	// Mempool defines the Blockbuster mempool implement. It contains a registry
@@ -34,24 +37,27 @@ func NewMempool(lanes ...Lane) *BBMempool {
 	}
 }
 
-// TODO: Consider using a tx cache in Mempool and returning the length of that
-// cache instead of relying on lane count tracking.
+// CountTx returns the total number of transactions in the mempool.
 func (m *BBMempool) CountTx() int {
 	var total int
 	for _, lane := range m.registry {
-		// TODO: If a global lane exists, we assume that lane has all transactions
-		// and we return the total.
-		//
-		// if lane.Name() == LaneNameGlobal {
-		// 	return lane.CountTx()
-		// }
-
 		total += lane.CountTx()
 	}
 
 	return total
 }
 
+// GetTxDistribution returns the number of transactions in each lane.
+func (m *BBMempool) GetTxDistribution() map[string]int {
+	counts := make(map[string]int, len(m.registry))
+
+	for _, lane := range m.registry {
+		counts[lane.Name()] = lane.CountTx()
+	}
+
+	return counts
+}
+
 // Insert inserts a transaction into every lane that it matches. Insertion will
 // be attempted on all lanes, even if an error is encountered.
 func (m *BBMempool) Insert(ctx context.Context, tx sdk.Tx) error {
@@ -74,8 +80,8 @@ func (m *BBMempool) Select(_ context.Context, _ [][]byte) sdkmempool.Iterator {
 	return nil
 }
 
-// Remove removes a transaction from every lane that it matches. Removal will be
-// attempted on all lanes, even if an error is encountered.
+// Remove removes a transaction from the mempool. It removes the transaction
+// from the first lane that it matches.
 func (m *BBMempool) Remove(tx sdk.Tx) error {
 	for _, lane := range m.registry {
 		if lane.Match(tx) {

blockbuster/priority_nonce.go

@@ -0,0 +1,477 @@
+package blockbuster
+
+import (
+	"context"
+	"fmt"
+	"math"
+
+	"github.com/huandu/skiplist"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdkmempool "github.com/cosmos/cosmos-sdk/types/mempool"
+	"github.com/cosmos/cosmos-sdk/x/auth/signing"
+)
+
+var (
+	_ sdkmempool.Mempool  = (*PriorityNonceMempool[int64])(nil)
+	_ sdkmempool.Iterator = (*PriorityNonceIterator[int64])(nil)
+)
+
+type (
+	// PriorityNonceMempoolConfig defines the configuration used to configure the
+	// PriorityNonceMempool.
+	PriorityNonceMempoolConfig[C comparable] struct {
+		// TxPriority defines the transaction priority and comparator.
+		TxPriority TxPriority[C]
+
+		// OnRead is a callback to be called when a tx is read from the mempool.
+		OnRead func(tx sdk.Tx)
+
+		// TxReplacement is a callback to be called when duplicated transaction nonce
+		// detected during mempool insert. An application can define a transaction
+		// replacement rule based on tx priority or certain transaction fields.
+		TxReplacement func(op, np C, oTx, nTx sdk.Tx) bool
+
+		// MaxTx sets the maximum number of transactions allowed in the mempool with
+		// the semantics:
+		// - if MaxTx == 0, there is no cap on the number of transactions in the mempool
+		// - if MaxTx > 0, the mempool will cap the number of transactions it stores,
+		//   and will prioritize transactions by their priority and sender-nonce
+		//   (sequence number) when evicting transactions.
+		// - if MaxTx < 0, `Insert` is a no-op.
+		MaxTx int
+	}
+
+	// PriorityNonceMempool is a mempool implementation that stores txs
+	// in a partially ordered set by 2 dimensions: priority, and sender-nonce
+	// (sequence number). Internally it uses one priority ordered skip list and one
+	// skip list per sender ordered by sender-nonce (sequence number). When there
+	// are multiple txs from the same sender, they are not always comparable by
+	// priority to other sender txs and must be partially ordered by both sender-nonce
+	// and priority.
+	PriorityNonceMempool[C comparable] struct {
+		priorityIndex  *skiplist.SkipList
+		priorityCounts map[C]int
+		senderIndices  map[string]*skiplist.SkipList
+		scores         map[txMeta[C]]txMeta[C]
+		cfg            PriorityNonceMempoolConfig[C]
+	}
+
+	// PriorityNonceIterator defines an iterator that is used for mempool iteration
+	// on Select().
+	PriorityNonceIterator[C comparable] struct {
+		mempool       *PriorityNonceMempool[C]
+		priorityNode  *skiplist.Element
+		senderCursors map[string]*skiplist.Element
+		sender        string
+		nextPriority  C
+	}
+
+	// TxPriority defines a type that is used to retrieve and compare transaction
+	// priorities. Priorities must be comparable.
+	TxPriority[C comparable] struct {
+		// GetTxPriority returns the priority of the transaction. A priority must be
+		// comparable via Compare.
+		GetTxPriority func(ctx context.Context, tx sdk.Tx) C
+
+		// CompareTxPriority compares two transaction priorities. The result should be
+		// 0 if a == b, -1 if a < b, and +1 if a > b.
+		Compare func(a, b C) int
+
+		// MinValue defines the minimum priority value, e.g. MinInt64. This value is
+		// used when instantiating a new iterator and comparing weights.
+		MinValue C
+	}
+
+	// txMeta stores transaction metadata used in indices
+	txMeta[C comparable] struct {
+		// nonce is the sender's sequence number
+		nonce uint64
+		// priority is the transaction's priority
+		priority C
+		// sender is the transaction's sender
+		sender string
+		// weight is the transaction's weight, used as a tiebreaker for transactions
+		// with the same priority
+		weight C
+		// senderElement is a pointer to the transaction's element in the sender index
+		senderElement *skiplist.Element
+	}
+)
+
+// NewDefaultTxPriority returns a TxPriority comparator using ctx.Priority as
+// the defining transaction priority.
+func NewDefaultTxPriority() TxPriority[int64] {
+	return TxPriority[int64]{
+		GetTxPriority: func(goCtx context.Context, _ sdk.Tx) int64 {
+			return sdk.UnwrapSDKContext(goCtx).Priority()
+		},
+		Compare: func(a, b int64) int {
+			return skiplist.Int64.Compare(a, b)
+		},
+		MinValue: math.MinInt64,
+	}
+}
+
+func DefaultPriorityNonceMempoolConfig() PriorityNonceMempoolConfig[int64] {
+	return PriorityNonceMempoolConfig[int64]{
+		TxPriority: NewDefaultTxPriority(),
+	}
+}
+
+// skiplistComparable is a comparator for txKeys that first compares priority,
+// then weight, then sender, then nonce, uniquely identifying a transaction.
+//
+// Note, skiplistComparable is used as the comparator in the priority index.
+func skiplistComparable[C comparable](txPriority TxPriority[C]) skiplist.Comparable {
+	return skiplist.LessThanFunc(func(a, b any) int {
+		keyA := a.(txMeta[C])
+		keyB := b.(txMeta[C])
+
+		res := txPriority.Compare(keyA.priority, keyB.priority)
+		if res != 0 {
+			return res
+		}
+
+		// Weight is used as a tiebreaker for transactions with the same priority.
+		// Weight is calculated in a single pass in .Select(...) and so will be 0
+		// on .Insert(...).
+		res = txPriority.Compare(keyA.weight, keyB.weight)
+		if res != 0 {
+			return res
+		}
+
+		// Because weight will be 0 on .Insert(...), we must also compare sender and
+		// nonce to resolve priority collisions. If we didn't then transactions with
+		// the same priority would overwrite each other in the priority index.
+		res = skiplist.String.Compare(keyA.sender, keyB.sender)
+		if res != 0 {
+			return res
+		}
+
+		return skiplist.Uint64.Compare(keyA.nonce, keyB.nonce)
+	})
+}
+
+// NewPriorityMempool returns the SDK's default mempool implementation which
+// returns txs in a partial order by 2 dimensions; priority, and sender-nonce.
+func NewPriorityMempool[C comparable](cfg PriorityNonceMempoolConfig[C]) *PriorityNonceMempool[C] {
+	mp := &PriorityNonceMempool[C]{
+		priorityIndex:  skiplist.New(skiplistComparable(cfg.TxPriority)),
+		priorityCounts: make(map[C]int),
+		senderIndices:  make(map[string]*skiplist.SkipList),
+		scores:         make(map[txMeta[C]]txMeta[C]),
+		cfg:            cfg,
+	}
+
+	return mp
+}
+
+// DefaultPriorityMempool returns a priorityNonceMempool with no options.
+func DefaultPriorityMempool() *PriorityNonceMempool[int64] {
+	return NewPriorityMempool(DefaultPriorityNonceMempoolConfig())
+}
+
+// NextSenderTx returns the next transaction for a given sender by nonce order,
+// i.e. the next valid transaction for the sender. If no such transaction exists,
+// nil will be returned.
+func (mp *PriorityNonceMempool[C]) NextSenderTx(sender string) sdk.Tx {
+	senderIndex, ok := mp.senderIndices[sender]
+	if !ok {
+		return nil
+	}
+
+	cursor := senderIndex.Front()
+	return cursor.Value.(sdk.Tx)
+}
+
+// Insert attempts to insert a Tx into the app-side mempool in O(log n) time,
+// returning an error if unsuccessful. Sender and nonce are derived from the
+// transaction's first signature.
+//
+// Transactions are unique by sender and nonce. Inserting a duplicate tx is an
+// O(log n) no-op.
+//
+// Inserting a duplicate tx with a different priority overwrites the existing tx,
+// changing the total order of the mempool.
+func (mp *PriorityNonceMempool[C]) Insert(ctx context.Context, tx sdk.Tx) error {
+	if mp.cfg.MaxTx > 0 && mp.CountTx() >= mp.cfg.MaxTx {
+		return sdkmempool.ErrMempoolTxMaxCapacity
+	} else if mp.cfg.MaxTx < 0 {
+		return nil
+	}
+
+	sigs, err := tx.(signing.SigVerifiableTx).GetSignaturesV2()
+	if err != nil {
+		return err
+	}
+	if len(sigs) == 0 {
+		return fmt.Errorf("tx must have at least one signer")
+	}
+
+	sig := sigs[0]
+	sender := sdk.AccAddress(sig.PubKey.Address()).String()
+	priority := mp.cfg.TxPriority.GetTxPriority(ctx, tx)
+	nonce := sig.Sequence
+	key := txMeta[C]{nonce: nonce, priority: priority, sender: sender}
+
+	senderIndex, ok := mp.senderIndices[sender]
+	if !ok {
+		senderIndex = skiplist.New(skiplist.LessThanFunc(func(a, b any) int {
+			return skiplist.Uint64.Compare(b.(txMeta[C]).nonce, a.(txMeta[C]).nonce)
+		}))
+
+		// initialize sender index if not found
+		mp.senderIndices[sender] = senderIndex
+	}
+
+	// Since mp.priorityIndex is scored by priority, then sender, then nonce, a
+	// changed priority will create a new key, so we must remove the old key and
+	// re-insert it to avoid having the same tx with different priorityIndex indexed
+	// twice in the mempool.
+	//
+	// This O(log n) remove operation is rare and only happens when a tx's priority
+	// changes.
+	sk := txMeta[C]{nonce: nonce, sender: sender}
+	if oldScore, txExists := mp.scores[sk]; txExists {
+		if mp.cfg.TxReplacement != nil && !mp.cfg.TxReplacement(oldScore.priority, priority, senderIndex.Get(key).Value.(sdk.Tx), tx) {
+			return fmt.Errorf(
+				"tx doesn't fit the replacement rule, oldPriority: %v, newPriority: %v, oldTx: %v, newTx: %v",
+				oldScore.priority,
+				priority,
+				senderIndex.Get(key).Value.(sdk.Tx),
+				tx,
+			)
+		}
+
+		mp.priorityIndex.Remove(txMeta[C]{
+			nonce:    nonce,
+			sender:   sender,
+			priority: oldScore.priority,
+			weight:   oldScore.weight,
+		})
+		mp.priorityCounts[oldScore.priority]--
+	}
+
+	mp.priorityCounts[priority]++
+
+	// Since senderIndex is scored by nonce, a changed priority will overwrite the
+	// existing key.
+	key.senderElement = senderIndex.Set(key, tx)
+
+	mp.scores[sk] = txMeta[C]{priority: priority}
+	mp.priorityIndex.Set(key, tx)
+
+	return nil
+}
+
+func (i *PriorityNonceIterator[C]) iteratePriority() sdkmempool.Iterator {
+	// beginning of priority iteration
+	if i.priorityNode == nil {
+		i.priorityNode = i.mempool.priorityIndex.Front()
+	} else {
+		i.priorityNode = i.priorityNode.Next()
+	}
+
+	// end of priority iteration
+	if i.priorityNode == nil {
+		return nil
+	}
+
+	i.sender = i.priorityNode.Key().(txMeta[C]).sender
+
+	nextPriorityNode := i.priorityNode.Next()
+	if nextPriorityNode != nil {
+		i.nextPriority = nextPriorityNode.Key().(txMeta[C]).priority
+	} else {
+		i.nextPriority = i.mempool.cfg.TxPriority.MinValue
+	}
+
+	return i.Next()
+}
+
+func (i *PriorityNonceIterator[C]) Next() sdkmempool.Iterator {
+	if i.priorityNode == nil {
+		return nil
+	}
+
+	cursor, ok := i.senderCursors[i.sender]
+	if !ok {
+		// beginning of sender iteration
+		cursor = i.mempool.senderIndices[i.sender].Front()
+	} else {
+		// middle of sender iteration
+		cursor = cursor.Next()
+	}
+
+	// end of sender iteration
+	if cursor == nil {
+		return i.iteratePriority()
+	}
+
+	key := cursor.Key().(txMeta[C])
+
+	// We've reached a transaction with a priority lower than the next highest
+	// priority in the pool.
+	if i.mempool.cfg.TxPriority.Compare(key.priority, i.nextPriority) < 0 {
+		return i.iteratePriority()
+	} else if i.mempool.cfg.TxPriority.Compare(key.priority, i.nextPriority) == 0 {
+		// Weight is incorporated into the priority index key only (not sender index)
+		// so we must fetch it here from the scores map.
+		weight := i.mempool.scores[txMeta[C]{nonce: key.nonce, sender: key.sender}].weight
+		if i.mempool.cfg.TxPriority.Compare(weight, i.priorityNode.Next().Key().(txMeta[C]).weight) < 0 {
+			return i.iteratePriority()
+		}
+	}
+
+	i.senderCursors[i.sender] = cursor
+	return i
+}
+
+func (i *PriorityNonceIterator[C]) Tx() sdk.Tx {
+	return i.senderCursors[i.sender].Value.(sdk.Tx)
+}
+
+// Select returns a set of transactions from the mempool, ordered by priority
+// and sender-nonce in O(n) time. The passed in list of transactions are ignored.
+// This is a readonly operation, the mempool is not modified.
+//
+// The maxBytes parameter defines the maximum number of bytes of transactions to
+// return.
+func (mp *PriorityNonceMempool[C]) Select(_ context.Context, _ [][]byte) sdkmempool.Iterator {
+	if mp.priorityIndex.Len() == 0 {
+		return nil
+	}
+
+	mp.reorderPriorityTies()
+
+	iterator := &PriorityNonceIterator[C]{
+		mempool:       mp,
+		senderCursors: make(map[string]*skiplist.Element),
+	}
+
+	return iterator.iteratePriority()
+}
+
+type reorderKey[C comparable] struct {
+	deleteKey txMeta[C]
+	insertKey txMeta[C]
+	tx        sdk.Tx
+}
+
+func (mp *PriorityNonceMempool[C]) reorderPriorityTies() {
+	node := mp.priorityIndex.Front()
+
+	var reordering []reorderKey[C]
+	for node != nil {
+		key := node.Key().(txMeta[C])
+		if mp.priorityCounts[key.priority] > 1 {
+			newKey := key
+			newKey.weight = senderWeight(mp.cfg.TxPriority, key.senderElement)
+			reordering = append(reordering, reorderKey[C]{deleteKey: key, insertKey: newKey, tx: node.Value.(sdk.Tx)})
+		}
+
+		node = node.Next()
+	}
+
+	for _, k := range reordering {
+		mp.priorityIndex.Remove(k.deleteKey)
+		delete(mp.scores, txMeta[C]{nonce: k.deleteKey.nonce, sender: k.deleteKey.sender})
+		mp.priorityIndex.Set(k.insertKey, k.tx)
+		mp.scores[txMeta[C]{nonce: k.insertKey.nonce, sender: k.insertKey.sender}] = k.insertKey
+	}
+}
+
+// senderWeight returns the weight of a given tx (t) at senderCursor. Weight is
+// defined as the first (nonce-wise) same sender tx with a priority not equal to
+// t. It is used to resolve priority collisions, that is when 2 or more txs from
+// different senders have the same priority.
+func senderWeight[C comparable](txPriority TxPriority[C], senderCursor *skiplist.Element) C {
+	if senderCursor == nil {
+		return txPriority.MinValue
+	}
+
+	weight := senderCursor.Key().(txMeta[C]).priority
+	senderCursor = senderCursor.Next()
+	for senderCursor != nil {
+		p := senderCursor.Key().(txMeta[C]).priority
+		if txPriority.Compare(p, weight) != 0 {
+			weight = p
+		}
+
+		senderCursor = senderCursor.Next()
+	}
+
+	return weight
+}
+
+// CountTx returns the number of transactions in the mempool.
+func (mp *PriorityNonceMempool[C]) CountTx() int {
+	return mp.priorityIndex.Len()
+}
+
+// Remove removes a transaction from the mempool in O(log n) time, returning an
+// error if unsuccessful.
+func (mp *PriorityNonceMempool[C]) Remove(tx sdk.Tx) error {
+	sigs, err := tx.(signing.SigVerifiableTx).GetSignaturesV2()
+	if err != nil {
+		return err
+	}
+	if len(sigs) == 0 {
+		return fmt.Errorf("attempted to remove a tx with no signatures")
+	}
+
+	sig := sigs[0]
+	sender := sdk.AccAddress(sig.PubKey.Address()).String()
+	nonce := sig.Sequence
+
+	scoreKey := txMeta[C]{nonce: nonce, sender: sender}
+	score, ok := mp.scores[scoreKey]
+	if !ok {
+		return sdkmempool.ErrTxNotFound
+	}
+	tk := txMeta[C]{nonce: nonce, priority: score.priority, sender: sender, weight: score.weight}
+
+	senderTxs, ok := mp.senderIndices[sender]
+	if !ok {
+		return fmt.Errorf("sender %s not found", sender)
+	}
+
+	mp.priorityIndex.Remove(tk)
+	senderTxs.Remove(tk)
+	delete(mp.scores, scoreKey)
+	mp.priorityCounts[score.priority]--
+
+	return nil
+}
+
+func IsEmpty[C comparable](mempool sdkmempool.Mempool) error {
+	mp := mempool.(*PriorityNonceMempool[C])
+	if mp.priorityIndex.Len() != 0 {
+		return fmt.Errorf("priorityIndex not empty")
+	}
+
+	countKeys := make([]C, 0, len(mp.priorityCounts))
+	for k := range mp.priorityCounts {
+		countKeys = append(countKeys, k)
+	}
+
+	for _, k := range countKeys {
+		if mp.priorityCounts[k] != 0 {
+			return fmt.Errorf("priorityCounts not zero at %v, got %v", k, mp.priorityCounts[k])
+		}
+	}
+
+	senderKeys := make([]string, 0, len(mp.senderIndices))
+	for k := range mp.senderIndices {
+		senderKeys = append(senderKeys, k)
+	}
+
+	for _, k := range senderKeys {
+		if mp.senderIndices[k].Len() != 0 {
+			return fmt.Errorf("senderIndex not empty for sender %v", k)
+		}
+	}
+
+	return nil
+}

blockbuster/utils/utils.go

@@ -1,4 +1,4 @@
-package blockbuster
+package utils
 
 import (
 	"crypto/sha256"
@@ -7,19 +7,21 @@ import (
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkmempool "github.com/cosmos/cosmos-sdk/types/mempool"
+	"github.com/skip-mev/pob/blockbuster"
 )
 
-// GetTxHashStr returns the hex-encoded hash of the transaction.
-func GetTxHashStr(txEncoder sdk.TxEncoder, tx sdk.Tx) (string, error) {
+// GetTxHashStr returns the hex-encoded hash of the transaction alongside the
+// transaction bytes.
+func GetTxHashStr(txEncoder sdk.TxEncoder, tx sdk.Tx) ([]byte, string, error) {
 	txBz, err := txEncoder(tx)
 	if err != nil {
-		return "", fmt.Errorf("failed to encode transaction: %w", err)
+		return nil, "", fmt.Errorf("failed to encode transaction: %w", err)
 	}
 
 	txHash := sha256.Sum256(txBz)
 	txHashStr := hex.EncodeToString(txHash[:])
 
-	return txHashStr, nil
+	return txBz, txHashStr, nil
 }
 
 // RemoveTxsFromLane removes the transactions from the given lane's mempool.
@@ -35,7 +37,7 @@ func RemoveTxsFromLane(txs map[sdk.Tx]struct{}, mempool sdkmempool.Mempool) erro
 
 // GetMaxTxBytesForLane returns the maximum number of bytes that can be included in the proposal
 // for the given lane.
-func GetMaxTxBytesForLane(proposal *Proposal, ratio sdk.Dec) int64 {
+func GetMaxTxBytesForLane(proposal *blockbuster.Proposal, ratio sdk.Dec) int64 {
 	// In the case where the ratio is zero, we return the max tx bytes remaining. Note, the only
 	// lane that should have a ratio of zero is the default lane. This means the default lane
 	// will have no limit on the number of transactions it can include in a block and is only