rename of constructor to base
This commit is contained in:
David Terpay
@@ -0,0 +1,452 @@
|
||||
# 🎨 Base Lane
|
||||
|
||||
> 🏗️ Build your own lane in less than 10 minutes using the Base Lane
|
||||
|
||||
## 💡 Overview
|
||||
|
||||
The Base Lane is a generic implementation of a lane. It comes out of the
|
||||
box with default implementations for all the required interfaces. It is meant to
|
||||
be used as a starting point for building your own lane.
|
||||
|
||||
## 🤔 How to use it
|
||||
|
||||
> **Default Implementations**
|
||||
>
|
||||
> There are default implementations for all of the below which can be found in
|
||||
> the `block/base` package. It is highly recommended that developers overview
|
||||
> the default implementations before building their own lane.
|
||||
|
||||
There are **three** critical components to building a custom lane using the lane
|
||||
constructor:
|
||||
|
||||
1. `LaneConfig` - The lane configuration which determines the basic properties
|
||||
of the lane including the maximum block space that the lane can fill up.
|
||||
2. `LaneMempool` - The lane mempool which is responsible for storing
|
||||
transactions that have been verified and are waiting to be included in proposals.
|
||||
3. `MatchHandler` - This is responsible for determining whether a transaction should
|
||||
belong to this lane.
|
||||
4. [**OPTIONAL**] `PrepareLaneHandler` - Allows developers to define their own
|
||||
handler to customize the how transactions are verified and ordered before they
|
||||
are included into a proposal.
|
||||
5. [**OPTIONAL**] `CheckOrderHandler` - Allows developers to define their own
|
||||
handler that will run any custom checks on whether transactions included in
|
||||
block proposals are in the correct order (respecting the ordering rules of the
|
||||
lane and the ordering rules of the other lanes).
|
||||
6. [**OPTIONAL**] `ProcessLaneHandler` - Allows developers to define their own
|
||||
handler for processing transactions that are included in block proposals.
|
||||
|
||||
|
||||
### 1. 📝 Lane Config
|
||||
|
||||
The lane config (`LaneConfig`) is a simple configuration
|
||||
object that defines the desired amount of block space the lane should
|
||||
utilize when building a proposal, an antehandler that is used to verify
|
||||
transactions as they are added/verified to/in a proposal, and more. By default,
|
||||
we recommend that user's pass in all of the base apps configurations (txDecoder,
|
||||
logger, etc.). A sample `LaneConfig` might look like the following:
|
||||
|
||||
```golang
|
||||
config := block.LaneConfig{
|
||||
Logger: app.Logger(),
|
||||
TxDecoder: app.TxDecoder(),
|
||||
TxEncoder: app.TxEncoder(),
|
||||
AnteHandler: app.AnteHandler(),
|
||||
MaxTxs: 0,
|
||||
MaxBlockSpace: math.LegacyZeroDec(),
|
||||
IgnoreList: []block.Lane{},
|
||||
}
|
||||
```
|
||||
|
||||
The three most important parameters to set are the `AnteHandler`, `MaxTxs`, and
|
||||
`MaxBlockSpace`.
|
||||
|
||||
#### **AnteHandler**
|
||||
|
||||
With the default implementation, the `AnteHandler` is responsible for verifying
|
||||
transactions as they are being considered for a new proposal or are being processed
|
||||
in a proposed block. We recommend user's utilize the same antehandler chain that
|
||||
is used in the base app. If developers want a certain `AnteDecorator` to be
|
||||
ignored if it qualifies for a given lane, they can do so by using the `NewIgnoreDecorator`
|
||||
defined in `block/utils/ante.go`.
|
||||
|
||||
For example, a free lane might want to ignore the `DeductFeeDecorator` so that it's
|
||||
transactions are not charged any fees. Where ever the `AnteHandler` is defined,
|
||||
we could add the following to ignore the `DeductFeeDecorator`:
|
||||
|
||||
```golang
|
||||
anteDecorators := []sdk.AnteDecorator{
|
||||
ante.NewSetUpContextDecorator(),
|
||||
...,
|
||||
utils.NewIgnoreDecorator(
|
||||
ante.NewDeductFeeDecorator(
|
||||
options.BaseOptions.AccountKeeper,
|
||||
options.BaseOptions.BankKeeper,
|
||||
options.BaseOptions.FeegrantKeeper,
|
||||
options.BaseOptions.TxFeeChecker,
|
||||
),
|
||||
options.FreeLane,
|
||||
),
|
||||
...,
|
||||
}
|
||||
```
|
||||
|
||||
Anytime a transaction that qualifies for the free lane is being processed, the
|
||||
`DeductFeeDecorator` will be ignored and no fees will be deducted!
|
||||
|
||||
|
||||
#### **MaxTxs**
|
||||
|
||||
This sets the maximum number of transactions allowed in the mempool with
|
||||
the semantics:
|
||||
|
||||
* if `MaxTxs` == 0, there is no cap on the number of transactions in the mempool
|
||||
* if `MaxTxs` > 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 `MaxTxs` < 0, `Insert` is a no-op.
|
||||
|
||||
#### **MaxBlockSpace**
|
||||
|
||||
MaxBlockSpace is the maximum amount of block space that the lane will attempt to
|
||||
fill when building a proposal. This parameter may be useful lanes that should be
|
||||
limited (such as a free or onboarding lane) in space usage. Setting this to 0
|
||||
will allow the lane to fill the block with as many transactions as possible.
|
||||
|
||||
If a block proposal request has a `MaxTxBytes` of 1000 and the lane has a
|
||||
`MaxBlockSpace` of 0.5, the lane will attempt to fill the block with 500 bytes.
|
||||
|
||||
#### **[OPTIONAL] IgnoreList**
|
||||
|
||||
`IgnoreList` defines the list of lanes to ignore when processing transactions.
|
||||
For example, say there are two lanes: default and free. The free lane is
|
||||
processed after the default lane. In this case, the free lane should be added
|
||||
to the ignore list of the default lane. Otherwise, the transactions that belong
|
||||
to the free lane will be processed by the default lane (which accepts all
|
||||
transactions by default).
|
||||
|
||||
|
||||
### 2. 🗄️ LaneMempool
|
||||
|
||||
This is the data structure that is responsible for storing transactions
|
||||
as they are being verified and are waiting to be included in proposals. `block/base/mempool.go`
|
||||
provides an out-of-the-box implementation that should be used as a starting
|
||||
point for building out the mempool and should cover most use cases. To
|
||||
utilize the mempool, you must implement a `TxPriority[C]` struct that does the
|
||||
following:
|
||||
|
||||
* Implements a `GetTxPriority` method that returns the priority (as defined
|
||||
by the type `[C]`) of a given transaction.
|
||||
* Implements a `Compare` method that returns the relative priority of two
|
||||
transactions. If the first transaction has a higher priority, the method
|
||||
should return -1, if the second transaction has a higher priority the method
|
||||
should return 1, otherwise the method should return 0.
|
||||
* Implements a `MinValue` method that returns the minimum priority value
|
||||
that a transaction can have.
|
||||
|
||||
The default implementation can be found in `block/base/mempool.go`. What
|
||||
if we wanted to prioritize transactions by the amount they have staked on a chain?
|
||||
Well we could do something like the following:
|
||||
|
||||
```golang
|
||||
// CustomTxPriority returns a TxPriority that prioritizes transactions by the
|
||||
// amount they have staked on chain. This means that transactions with a higher
|
||||
// amount staked will be prioritized over transactions with a lower amount staked.
|
||||
func (p *CustomTxPriority) CustomTxPriority() TxPriority[string] {
|
||||
return TxPriority[string]{
|
||||
GetTxPriority: func(ctx context.Context, tx sdk.Tx) string {
|
||||
// Get the signer of the transaction.
|
||||
signer := p.getTransactionSigner(tx)
|
||||
|
||||
// Get the total amount staked by the signer on chain.
|
||||
// This is abstracted away in the example, but you can
|
||||
// implement this using the staking keeper.
|
||||
totalStake, err := p.getTotalStake(ctx, signer)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
|
||||
return totalStake.String()
|
||||
},
|
||||
Compare: func(a, b string) int {
|
||||
aCoins, _ := sdk.ParseCoinsNormalized(a)
|
||||
bCoins, _ := sdk.ParseCoinsNormalized(b)
|
||||
|
||||
switch {
|
||||
case aCoins == nil && bCoins == nil:
|
||||
return 0
|
||||
|
||||
case aCoins == nil:
|
||||
return -1
|
||||
|
||||
case bCoins == nil:
|
||||
return 1
|
||||
|
||||
default:
|
||||
switch {
|
||||
case aCoins.IsAllGT(bCoins):
|
||||
return 1
|
||||
|
||||
case aCoins.IsAllLT(bCoins):
|
||||
return -1
|
||||
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
},
|
||||
MinValue: "",
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### Using a Custom TxPriority
|
||||
|
||||
To utilize this new priority configuration in a lane, all you have to then do
|
||||
is pass in the `TxPriority[C]` to the `NewLaneMempool` function.
|
||||
|
||||
```golang
|
||||
// Create the lane config
|
||||
laneCfg := NewLaneConfig(
|
||||
...
|
||||
MaxTxs: 100,
|
||||
...
|
||||
)
|
||||
|
||||
// Pseudocode for creating the custom tx priority
|
||||
priorityCfg := NewPriorityConfig(
|
||||
stakingKeeper,
|
||||
accountKeeper,
|
||||
...
|
||||
)
|
||||
|
||||
|
||||
// define your mempool that orders transactions by on-chain stake
|
||||
mempool := constructor.NewMempool[string](
|
||||
priorityCfg.CustomTxPriority(),
|
||||
laneCfg.TxEncoder,
|
||||
laneCfg.MaxTxs,
|
||||
)
|
||||
|
||||
// Initialize your lane with the mempool
|
||||
lane := constructor.NewBaseLane(
|
||||
laneCfg,
|
||||
LaneName,
|
||||
mempool,
|
||||
constructor.DefaultMatchHandler(),
|
||||
)
|
||||
```
|
||||
|
||||
### 3. 🤝 MatchHandler
|
||||
|
||||
`MatchHandler` is utilized to determine if a transaction should be included in
|
||||
the lane. This function can be a stateless or stateful check on the transaction.
|
||||
The default implementation can be found in `block/base/handlers.go`.
|
||||
|
||||
The match handler can be as custom as desired. Following the example above, if
|
||||
we wanted to make a lane that only accepts transactions if they have a large
|
||||
amount staked, we could do the following:
|
||||
|
||||
```golang
|
||||
// CustomMatchHandler returns a custom implementation of the MatchHandler. It
|
||||
// matches transactions that have a large amount staked. These transactions
|
||||
// will then be charged no fees at execution time.
|
||||
//
|
||||
// NOTE: This is a stateful check on the transaction. The details of how to
|
||||
// implement this are abstracted away in the example, but you can implement
|
||||
// this using the staking keeper.
|
||||
func (h *Handler) CustomMatchHandler() block.MatchHandler {
|
||||
return func(ctx sdk.Context, tx sdk.Tx) bool {
|
||||
if !h.IsStakingTx(tx) {
|
||||
return false
|
||||
}
|
||||
|
||||
signer, err := getTxSigner(tx)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
stakedAmount, err := h.GetStakedAmount(signer)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// The transaction can only be considered for inclusion if the amount
|
||||
// staked is greater than some predetermined threshold.
|
||||
return stakeAmount.GT(h.Threshold)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### Using a Custom MatchHandler
|
||||
|
||||
If we wanted to create the lane using the custom match handler along with the
|
||||
custom mempool, we could do the following:
|
||||
|
||||
```golang
|
||||
// Pseudocode for creating the custom match handler
|
||||
handler := NewHandler(
|
||||
stakingKeeper,
|
||||
accountKeeper,
|
||||
...
|
||||
)
|
||||
|
||||
// define your mempool that orders transactions by on chain stake
|
||||
mempool := constructor.NewMempool[string](
|
||||
priorityCfg.CustomTxPriority(),
|
||||
cfg.TxEncoder,
|
||||
cfg.MaxTxs,
|
||||
)
|
||||
|
||||
// Initialize your lane with the mempool
|
||||
lane := constructor.NewBaseLane(
|
||||
cfg,
|
||||
LaneName,
|
||||
mempool,
|
||||
handler.CustomMatchHandler(),
|
||||
)
|
||||
```
|
||||
|
||||
### Summary on Steps 1-3
|
||||
|
||||
The following is a summary of the steps above:
|
||||
|
||||
1. Create a custom `LaneConfig` struct that defines the configuration of the lane.
|
||||
2. Create a custom `TxPriority[C]` struct to have a custom mempool that orders
|
||||
transactions via a custom priority mechanism.
|
||||
3. Create a custom `MatchHandler` that implements the `block.MatchHandler` to
|
||||
have a custom lane that only accepts transactions that match a custom criteria.
|
||||
|
||||
### [OPTIONAL] Steps 4-6
|
||||
|
||||
The remaining steps walk through the process of creating custom block
|
||||
building/verification logic. The default implementation found in `block/base/handlers.go`
|
||||
should fit most use cases. Please reference that file for more details on
|
||||
the default implementation and whether it fits your use case.
|
||||
|
||||
Implementing custom block building/verification logic is a bit more involved
|
||||
than the previous steps and is a all or nothing approach. This means that if
|
||||
you implement any of the handlers, you must implement all of them in most cases.
|
||||
If you do not implement all of them, the lane may have unintended behavior.
|
||||
|
||||
### 4. 🛠️ PrepareLaneHandler
|
||||
|
||||
The `PrepareLaneHandler` is an optional field you can set on the lane constructor.
|
||||
This handler is responsible for the transaction selection logic when a new proposal
|
||||
is requested.
|
||||
|
||||
The handler should return the following for a given lane:
|
||||
|
||||
1. The transactions to be included in the block proposal.
|
||||
2. The transactions to be removed from the lane's mempool.
|
||||
3. An error if the lane is unable to prepare a block proposal.
|
||||
|
||||
```golang
|
||||
// PrepareLaneHandler is responsible for preparing transactions to be included
|
||||
// in the block from a given lane. Given a lane, this function should return
|
||||
// the transactions to include in the block, the transactions that must be
|
||||
// removed from the lane, and an error if one occurred.
|
||||
PrepareLaneHandler func(ctx sdk.Context,proposal BlockProposal,maxTxBytes int64)
|
||||
(txsToInclude [][]byte, txsToRemove []sdk.Tx, err error)
|
||||
```
|
||||
|
||||
The default implementation is simple. It will continue to select transactions
|
||||
from its mempool under the following criteria:
|
||||
|
||||
1. The transactions is not already included in the block proposal.
|
||||
2. The transaction is valid and passes the AnteHandler check.
|
||||
3. The transaction is not too large to be included in the block.
|
||||
|
||||
If a more involved selection process is required, you can implement your own
|
||||
`PrepareLaneHandler` and and set it after creating the lane constructor.
|
||||
|
||||
```golang
|
||||
// Pseudocode for creating the custom prepare lane handler
|
||||
// This assumes that the CustomLane inherits from the constructor
|
||||
// lane.
|
||||
customLane := constructor.NewCustomLane(
|
||||
cfg,
|
||||
LaneName,
|
||||
mempool,
|
||||
handler.CustomMatchHandler(),
|
||||
)
|
||||
|
||||
// Set the custom PrepareLaneHandler on the lane
|
||||
customLane.SetPrepareLaneHandler(customlane.PrepareLaneHandler())
|
||||
```
|
||||
|
||||
### 5. ✅ CheckOrderHandler
|
||||
|
||||
The `CheckOrderHandler` is an optional field you can set on the lane constructor.
|
||||
This handler is responsible for verifying the ordering of the transactions in the
|
||||
block proposal that belong to the lane.
|
||||
|
||||
```golang
|
||||
// CheckOrderHandler is responsible for checking the order of transactions that
|
||||
// belong to a given lane. This handler should be used to verify that the
|
||||
// ordering of transactions passed into the function respect the ordering logic
|
||||
// of the lane (if any transactions from the lane are included). This function
|
||||
// should also ensure that transactions that belong to this lane are contiguous
|
||||
// and do not have any transactions from other lanes in between them.
|
||||
CheckOrderHandler func(ctx sdk.Context, txs []sdk.Tx) error
|
||||
```
|
||||
|
||||
The default implementation is simple and utilizes the same `TxPriority` struct
|
||||
that the mempool uses to determine if transactions are in order. The criteria
|
||||
for determining if transactions are in order is as follows:
|
||||
|
||||
1. The transactions are in order according to the `TxPriority` struct. i.e. any
|
||||
two transactions (that match to the lane) `tx1` and `tx2` where `tx1` has a
|
||||
higher priority than `tx2` should be ordered before `tx2`.
|
||||
2. The transactions are contiguous. i.e. there are no transactions from other
|
||||
lanes in between the transactions that belong to this lane. i.e. if `tx1` and
|
||||
`tx2` belong to the lane, there should be no transactions from other lanes in
|
||||
between `tx1` and `tx2`.
|
||||
|
||||
If a more involved ordering process is required, you can implement your own
|
||||
`CheckOrderHandler` and and set it after creating the lane constructor.
|
||||
|
||||
```golang
|
||||
// Pseudocode for creating the custom check order handler
|
||||
// This assumes that the CustomLane inherits from the constructor
|
||||
// lane.
|
||||
customLane := constructor.NewCustomLane(
|
||||
cfg,
|
||||
LaneName,
|
||||
mempool,
|
||||
handler.CustomMatchHandler(),
|
||||
)
|
||||
|
||||
// Set the custom CheckOrderHandler on the lane
|
||||
customLane.SetCheckOrderHandler(customlane.CheckOrderHandler())
|
||||
```
|
||||
|
||||
|
||||
### 6. 🆗 ProcessLaneHandler
|
||||
|
||||
The `ProcessLaneHandler` is an optional field you can set on the lane constructor.
|
||||
This handler is responsible for verifying the transactions in the block proposal
|
||||
that belong to the lane. This handler is executed after the `CheckOrderHandler`
|
||||
so the transactions passed into this function SHOULD already be in order
|
||||
respecting the ordering rules of the lane and respecting the ordering rules of
|
||||
mempool relative to the lanes it has. This means that if the first transaction
|
||||
does not belong to the lane, the remaining transactions should not belong to the
|
||||
lane either.
|
||||
|
||||
|
||||
```golang
|
||||
// ProcessLaneHandler is responsible for processing transactions that are
|
||||
// included in a block and belong to a given lane. ProcessLaneHandler is
|
||||
// executed after CheckOrderHandler so the transactions passed into this
|
||||
// function SHOULD already be in order respecting the ordering rules of the
|
||||
// lane and respecting the ordering rules of mempool relative to the lanes it has.
|
||||
ProcessLaneHandler func(ctx sdk.Context, txs []sdk.Tx) ([]sdk.Tx, error)
|
||||
```
|
||||
|
||||
Given the invarients above, the default implementation is simple. It will
|
||||
continue to verify transactions in the block proposal under the following
|
||||
criteria:
|
||||
|
||||
1. If a transaction matches to this lane, verify it and continue. If it is not
|
||||
valid, return an error.
|
||||
2. If a transaction does not match to this lane, return the remaining transactions
|
||||
to the next lane to process.
|
||||
@@ -0,0 +1,68 @@
|
||||
package base
|
||||
|
||||
import (
|
||||
sdk "github.com/cosmos/cosmos-sdk/types"
|
||||
"github.com/skip-mev/pob/block"
|
||||
"github.com/skip-mev/pob/block/utils"
|
||||
)
|
||||
|
||||
// PrepareLane will prepare a partial proposal for the lane. It will select transactions from the
|
||||
// lane respecting the selection logic of the prepareLaneHandler. It will then update the partial
|
||||
// proposal with the selected transactions. If the proposal is unable to be updated, we return an
|
||||
// error. The proposal will only be modified if it passes all of the invarient checks.
|
||||
func (l *BaseLane) PrepareLane(
|
||||
ctx sdk.Context,
|
||||
proposal block.BlockProposal,
|
||||
maxTxBytes int64,
|
||||
next block.PrepareLanesHandler,
|
||||
) (block.BlockProposal, error) {
|
||||
txs, txsToRemove, err := l.prepareLaneHandler(ctx, proposal, maxTxBytes)
|
||||
if err != nil {
|
||||
return proposal, err
|
||||
}
|
||||
|
||||
// Remove all transactions that were invalid during the creation of the partial proposal.
|
||||
if err := utils.RemoveTxsFromLane(txsToRemove, l); err != nil {
|
||||
l.Logger().Error(
|
||||
"failed to remove transactions from lane",
|
||||
"lane", l.Name(),
|
||||
"err", err,
|
||||
)
|
||||
}
|
||||
|
||||
// Update the proposal with the selected transactions.
|
||||
if err := proposal.UpdateProposal(l, txs); err != nil {
|
||||
return proposal, err
|
||||
}
|
||||
|
||||
return next(ctx, proposal)
|
||||
}
|
||||
|
||||
// CheckOrder checks that the ordering logic of the lane is respected given the set of transactions
|
||||
// in the block proposal. If the ordering logic is not respected, we return an error.
|
||||
func (l *BaseLane) CheckOrder(ctx sdk.Context, txs []sdk.Tx) error {
|
||||
return l.checkOrderHandler(ctx, txs)
|
||||
}
|
||||
|
||||
// ProcessLane verifies that the transactions included in the block proposal are valid respecting
|
||||
// the verification logic of the lane (processLaneHandler). If the transactions are valid, we
|
||||
// return the transactions that do not belong to this lane to the next lane. If the transactions
|
||||
// are invalid, we return an error.
|
||||
func (l *BaseLane) ProcessLane(ctx sdk.Context, txs []sdk.Tx, next block.ProcessLanesHandler) (sdk.Context, error) {
|
||||
remainingTxs, err := l.processLaneHandler(ctx, txs)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
}
|
||||
|
||||
return next(ctx, remainingTxs)
|
||||
}
|
||||
|
||||
// AnteVerifyTx verifies that the transaction is valid respecting the ante verification logic of
|
||||
// of the antehandler chain.
|
||||
func (l *BaseLane) AnteVerifyTx(ctx sdk.Context, tx sdk.Tx, simulate bool) (sdk.Context, error) {
|
||||
if l.cfg.AnteHandler != nil {
|
||||
return l.cfg.AnteHandler(ctx, tx, simulate)
|
||||
}
|
||||
|
||||
return ctx, nil
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package base
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"cosmossdk.io/log"
|
||||
"cosmossdk.io/math"
|
||||
sdk "github.com/cosmos/cosmos-sdk/types"
|
||||
"github.com/skip-mev/pob/block"
|
||||
)
|
||||
|
||||
// LaneConfig defines the basic functionality needed for a lane.
|
||||
type LaneConfig struct {
|
||||
Logger log.Logger
|
||||
TxEncoder sdk.TxEncoder
|
||||
TxDecoder sdk.TxDecoder
|
||||
AnteHandler sdk.AnteHandler
|
||||
|
||||
// MaxBlockSpace defines the relative percentage of block space that can be
|
||||
// used by this lane. NOTE: If this is set to zero, then there is no limit
|
||||
// on the number of transactions that can be included in the block for this
|
||||
// lane (up to maxTxBytes as provided by the request). This is useful for the default lane.
|
||||
MaxBlockSpace math.LegacyDec
|
||||
|
||||
// IgnoreList defines the list of lanes to ignore when processing transactions. This
|
||||
// is useful for when you want lanes to exist after the default lane. For example,
|
||||
// say there are two lanes: default and free. The free lane should be processed after
|
||||
// the default lane. In this case, the free lane should be added to the ignore list
|
||||
// of the default lane. Otherwise, the transactions that belong to the free lane
|
||||
// will be processed by the default lane (which accepts all transactions by default).
|
||||
IgnoreList []block.Lane
|
||||
|
||||
// MaxTxs 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.
|
||||
MaxTxs int
|
||||
}
|
||||
|
||||
// NewLaneConfig returns a new LaneConfig. This will be embedded in a lane.
|
||||
func NewLaneConfig(
|
||||
logger log.Logger,
|
||||
txEncoder sdk.TxEncoder,
|
||||
txDecoder sdk.TxDecoder,
|
||||
anteHandler sdk.AnteHandler,
|
||||
maxBlockSpace math.LegacyDec,
|
||||
) LaneConfig {
|
||||
return LaneConfig{
|
||||
Logger: logger,
|
||||
TxEncoder: txEncoder,
|
||||
TxDecoder: txDecoder,
|
||||
AnteHandler: anteHandler,
|
||||
MaxBlockSpace: maxBlockSpace,
|
||||
}
|
||||
}
|
||||
|
||||
// ValidateBasic validates the lane configuration.
|
||||
func (c *LaneConfig) 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(math.LegacyOneDec()) {
|
||||
return fmt.Errorf("max block space must be set to a value between 0 and 1")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,156 @@
|
||||
package base
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
sdk "github.com/cosmos/cosmos-sdk/types"
|
||||
"github.com/skip-mev/pob/block"
|
||||
"github.com/skip-mev/pob/block/utils"
|
||||
)
|
||||
|
||||
// DefaultPrepareLaneHandler returns a default implementation of the PrepareLaneHandler. It
|
||||
// selects all transactions in the mempool that are valid and not already in the partial
|
||||
// proposal. It will continue to reap transactions until the maximum block space for this
|
||||
// lane has been reached. Additionally, any transactions that are invalid will be returned.
|
||||
func (l *BaseLane) DefaultPrepareLaneHandler() block.PrepareLaneHandler {
|
||||
return func(ctx sdk.Context, proposal block.BlockProposal, maxTxBytes int64) ([][]byte, []sdk.Tx, error) {
|
||||
var (
|
||||
totalSize int64
|
||||
txs [][]byte
|
||||
txsToRemove []sdk.Tx
|
||||
)
|
||||
|
||||
// Select all transactions in the mempool that are valid and not already in the
|
||||
// partial proposal.
|
||||
for iterator := l.Select(ctx, nil); iterator != nil; iterator = iterator.Next() {
|
||||
tx := iterator.Tx()
|
||||
|
||||
txBytes, hash, err := utils.GetTxHashStr(l.TxEncoder(), tx)
|
||||
if err != nil {
|
||||
l.Logger().Info("failed to get hash of tx", "err", err)
|
||||
|
||||
txsToRemove = append(txsToRemove, tx)
|
||||
continue
|
||||
}
|
||||
|
||||
// Double check that the transaction belongs to this lane.
|
||||
if !l.Match(ctx, tx) {
|
||||
l.Logger().Info(
|
||||
"failed to select tx for lane; tx does not belong to lane",
|
||||
"tx_hash", hash,
|
||||
"lane", l.Name(),
|
||||
)
|
||||
|
||||
txsToRemove = append(txsToRemove, tx)
|
||||
continue
|
||||
}
|
||||
|
||||
// if the transaction is already in the (partial) block proposal, we skip it.
|
||||
if proposal.Contains(txBytes) {
|
||||
l.Logger().Info(
|
||||
"failed to select tx for lane; tx is already in proposal",
|
||||
"tx_hash", hash,
|
||||
"lane", l.Name(),
|
||||
)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
// If the transaction is too large, we break and do not attempt to include more txs.
|
||||
txSize := int64(len(txBytes))
|
||||
if updatedSize := totalSize + txSize; updatedSize > maxTxBytes {
|
||||
l.Logger().Info(
|
||||
"tx bytes above the maximum allowed",
|
||||
"lane", l.Name(),
|
||||
"tx_size", txSize,
|
||||
"total_size", totalSize,
|
||||
"max_tx_bytes", maxTxBytes,
|
||||
"tx_hash", hash,
|
||||
)
|
||||
|
||||
break
|
||||
}
|
||||
|
||||
// Verify the transaction.
|
||||
if ctx, err = l.AnteVerifyTx(ctx, tx, false); err != nil {
|
||||
l.Logger().Info(
|
||||
"failed to verify tx",
|
||||
"tx_hash", hash,
|
||||
"err", err,
|
||||
)
|
||||
|
||||
txsToRemove = append(txsToRemove, tx)
|
||||
continue
|
||||
}
|
||||
|
||||
totalSize += txSize
|
||||
txs = append(txs, txBytes)
|
||||
}
|
||||
|
||||
return txs, txsToRemove, nil
|
||||
}
|
||||
}
|
||||
|
||||
// DefaultProcessLaneHandler returns a default implementation of the ProcessLaneHandler. It
|
||||
// verifies all transactions in the lane that matches to the lane. If any transaction
|
||||
// fails to verify, the entire proposal is rejected. If the handler comes across a transaction
|
||||
// that does not match the lane's matcher, it will return the remaining transactions in the
|
||||
// proposal.
|
||||
func (l *BaseLane) DefaultProcessLaneHandler() block.ProcessLaneHandler {
|
||||
return func(ctx sdk.Context, txs []sdk.Tx) ([]sdk.Tx, error) {
|
||||
var err error
|
||||
|
||||
// Process all transactions that match the lane's matcher.
|
||||
for index, tx := range txs {
|
||||
if l.Match(ctx, tx) {
|
||||
if ctx, err = l.AnteVerifyTx(ctx, tx, false); err != nil {
|
||||
return nil, fmt.Errorf("failed to verify tx: %w", err)
|
||||
}
|
||||
} else {
|
||||
return txs[index:], nil
|
||||
}
|
||||
}
|
||||
|
||||
// This means we have processed all transactions in the proposal.
|
||||
return nil, nil
|
||||
}
|
||||
}
|
||||
|
||||
// DefaultCheckOrderHandler returns a default implementation of the CheckOrderHandler. It
|
||||
// ensures the following invariants:
|
||||
//
|
||||
// 1. All transactions that belong to this lane respect the ordering logic defined by the
|
||||
// lane.
|
||||
// 2. Transactions that belong to other lanes cannot be interleaved with transactions that
|
||||
// belong to this lane.
|
||||
func (l *BaseLane) DefaultCheckOrderHandler() block.CheckOrderHandler {
|
||||
return func(ctx sdk.Context, txs []sdk.Tx) error {
|
||||
seenOtherLaneTx := false
|
||||
|
||||
for index, tx := range txs {
|
||||
if l.Match(ctx, tx) {
|
||||
if seenOtherLaneTx {
|
||||
return fmt.Errorf("the %s lane contains a transaction that belongs to another lane", l.Name())
|
||||
}
|
||||
|
||||
// If the transactions do not respect the priority defined by the mempool, we consider the proposal
|
||||
// to be invalid
|
||||
if index > 0 && l.Compare(ctx, txs[index-1], tx) == -1 {
|
||||
return fmt.Errorf("transaction at index %d has a higher priority than %d", index, index-1)
|
||||
}
|
||||
} else {
|
||||
seenOtherLaneTx = true
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// DefaultMatchHandler returns a default implementation of the MatchHandler. It matches all
|
||||
// transactions.
|
||||
func DefaultMatchHandler() block.MatchHandler {
|
||||
return func(ctx sdk.Context, tx sdk.Tx) bool {
|
||||
return true
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,199 @@
|
||||
package base
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"cosmossdk.io/log"
|
||||
"cosmossdk.io/math"
|
||||
sdk "github.com/cosmos/cosmos-sdk/types"
|
||||
"github.com/skip-mev/pob/block"
|
||||
)
|
||||
|
||||
var _ block.Lane = (*BaseLane)(nil)
|
||||
|
||||
// BaseLane is a generic implementation of a lane. It is meant to be used
|
||||
// as a base for other lanes to be built on top of. It provides a default
|
||||
// implementation of the MatchHandler, PrepareLaneHandler, ProcessLaneHandler,
|
||||
// and CheckOrderHandler. To extend this lane, you must either utilize the default
|
||||
// handlers or construct your own that you pass into the base/setters.
|
||||
type BaseLane struct {
|
||||
// cfg stores functionality required to encode/decode transactions, maintains how
|
||||
// many transactions are allowed in this lane's mempool, and the amount of block
|
||||
// space this lane is allowed to consume.
|
||||
cfg LaneConfig
|
||||
|
||||
// laneName is the name of the lane.
|
||||
laneName string
|
||||
|
||||
// LaneMempool is the mempool that is responsible for storing transactions
|
||||
// that are waiting to be processed.
|
||||
block.LaneMempool
|
||||
|
||||
// matchHandler is the function that determines whether or not a transaction
|
||||
// should be processed by this lane.
|
||||
matchHandler block.MatchHandler
|
||||
|
||||
// prepareLaneHandler is the function that is called when a new proposal is being
|
||||
// requested and the lane needs to submit transactions it wants included in the block.
|
||||
prepareLaneHandler block.PrepareLaneHandler
|
||||
|
||||
// checkOrderHandler is the function that is called when a new proposal is being
|
||||
// verified and the lane needs to verify that the transactions included in the proposal
|
||||
// respect the ordering rules of the lane and does not interleave transactions from other lanes.
|
||||
checkOrderHandler block.CheckOrderHandler
|
||||
|
||||
// processLaneHandler is the function that is called when a new proposal is being
|
||||
// verified and the lane needs to verify that the transactions included in the proposal
|
||||
// are valid respecting the verification logic of the lane.
|
||||
processLaneHandler block.ProcessLaneHandler
|
||||
}
|
||||
|
||||
// NewBaseLane returns a new lane base. When creating this lane, the type
|
||||
// of the lane must be specified. The type of the lane is directly associated with the
|
||||
// type of the mempool that is used to store transactions that are waiting to be processed.
|
||||
func NewBaseLane(
|
||||
cfg LaneConfig,
|
||||
laneName string,
|
||||
laneMempool block.LaneMempool,
|
||||
matchHandlerFn block.MatchHandler,
|
||||
) *BaseLane {
|
||||
lane := &BaseLane{
|
||||
cfg: cfg,
|
||||
laneName: laneName,
|
||||
LaneMempool: laneMempool,
|
||||
matchHandler: matchHandlerFn,
|
||||
}
|
||||
|
||||
if err := lane.ValidateBasic(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
return lane
|
||||
}
|
||||
|
||||
// ValidateBasic ensures that the lane was constructed properly. In the case that
|
||||
// the lane was not constructed with proper handlers, default handlers are set.
|
||||
func (l *BaseLane) ValidateBasic() error {
|
||||
if err := l.cfg.ValidateBasic(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if l.laneName == "" {
|
||||
return fmt.Errorf("lane name cannot be empty")
|
||||
}
|
||||
|
||||
if l.LaneMempool == nil {
|
||||
return fmt.Errorf("lane mempool cannot be nil")
|
||||
}
|
||||
|
||||
if l.matchHandler == nil {
|
||||
return fmt.Errorf("match handler cannot be nil")
|
||||
}
|
||||
|
||||
if l.prepareLaneHandler == nil {
|
||||
l.prepareLaneHandler = l.DefaultPrepareLaneHandler()
|
||||
}
|
||||
|
||||
if l.processLaneHandler == nil {
|
||||
l.processLaneHandler = l.DefaultProcessLaneHandler()
|
||||
}
|
||||
|
||||
if l.checkOrderHandler == nil {
|
||||
l.checkOrderHandler = l.DefaultCheckOrderHandler()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetPrepareLaneHandler sets the prepare lane handler for the lane. This handler
|
||||
// is called when a new proposal is being requested and the lane needs to submit
|
||||
// transactions it wants included in the block.
|
||||
func (l *BaseLane) SetPrepareLaneHandler(prepareLaneHandler block.PrepareLaneHandler) {
|
||||
if prepareLaneHandler == nil {
|
||||
panic("prepare lane handler cannot be nil")
|
||||
}
|
||||
|
||||
l.prepareLaneHandler = prepareLaneHandler
|
||||
}
|
||||
|
||||
// SetProcessLaneHandler sets the process lane handler for the lane. This handler
|
||||
// is called when a new proposal is being verified and the lane needs to verify
|
||||
// that the transactions included in the proposal are valid respecting the verification
|
||||
// logic of the lane.
|
||||
func (l *BaseLane) SetProcessLaneHandler(processLaneHandler block.ProcessLaneHandler) {
|
||||
if processLaneHandler == nil {
|
||||
panic("process lane handler cannot be nil")
|
||||
}
|
||||
|
||||
l.processLaneHandler = processLaneHandler
|
||||
}
|
||||
|
||||
// SetCheckOrderHandler sets the check order handler for the lane. This handler
|
||||
// is called when a new proposal is being verified and the lane needs to verify
|
||||
// that the transactions included in the proposal respect the ordering rules of
|
||||
// the lane and does not include transactions from other lanes.
|
||||
func (l *BaseLane) SetCheckOrderHandler(checkOrderHandler block.CheckOrderHandler) {
|
||||
if checkOrderHandler == nil {
|
||||
panic("check order handler cannot be nil")
|
||||
}
|
||||
|
||||
l.checkOrderHandler = checkOrderHandler
|
||||
}
|
||||
|
||||
// Match returns true if the transaction should be processed by this lane. This
|
||||
// function first determines if the transaction matches the lane and then checks
|
||||
// if the transaction is on the ignore list. If the transaction is on the ignore
|
||||
// list, it returns false.
|
||||
func (l *BaseLane) Match(ctx sdk.Context, tx sdk.Tx) bool {
|
||||
return l.matchHandler(ctx, tx) && !l.CheckIgnoreList(ctx, tx)
|
||||
}
|
||||
|
||||
// CheckIgnoreList returns true if the transaction is on the ignore list. The ignore
|
||||
// list is utilized to prevent transactions that should be considered in other lanes
|
||||
// from being considered from this lane.
|
||||
func (l *BaseLane) CheckIgnoreList(ctx sdk.Context, tx sdk.Tx) bool {
|
||||
for _, lane := range l.cfg.IgnoreList {
|
||||
if lane.Match(ctx, tx) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// Name returns the name of the lane.
|
||||
func (l *BaseLane) Name() string {
|
||||
return l.laneName
|
||||
}
|
||||
|
||||
// SetIgnoreList sets the ignore list for the lane. The ignore list is a list
|
||||
// of lanes that the lane should ignore when processing transactions.
|
||||
func (l *BaseLane) SetIgnoreList(lanes []block.Lane) {
|
||||
l.cfg.IgnoreList = lanes
|
||||
}
|
||||
|
||||
// SetAnteHandler sets the ante handler for the lane.
|
||||
func (l *BaseLane) SetAnteHandler(anteHandler sdk.AnteHandler) {
|
||||
l.cfg.AnteHandler = anteHandler
|
||||
}
|
||||
|
||||
// Logger returns the logger for the lane.
|
||||
func (l *BaseLane) Logger() log.Logger {
|
||||
return l.cfg.Logger
|
||||
}
|
||||
|
||||
// TxDecoder returns the tx decoder for the lane.
|
||||
func (l *BaseLane) TxDecoder() sdk.TxDecoder {
|
||||
return l.cfg.TxDecoder
|
||||
}
|
||||
|
||||
// TxEncoder returns the tx encoder for the lane.
|
||||
func (l *BaseLane) TxEncoder() sdk.TxEncoder {
|
||||
return l.cfg.TxEncoder
|
||||
}
|
||||
|
||||
// GetMaxBlockSpace returns the maximum amount of block space that the lane is
|
||||
// allowed to consume as a percentage of the total block space.
|
||||
func (l *BaseLane) GetMaxBlockSpace() math.LegacyDec {
|
||||
return l.cfg.MaxBlockSpace
|
||||
}
|
||||
@@ -0,0 +1,159 @@
|
||||
package base
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
sdk "github.com/cosmos/cosmos-sdk/types"
|
||||
sdkmempool "github.com/cosmos/cosmos-sdk/types/mempool"
|
||||
"github.com/skip-mev/pob/block/utils"
|
||||
)
|
||||
|
||||
type (
|
||||
// ConstructorMempool defines a mempool that orders transactions based on the
|
||||
// txPriority. The mempool is a wrapper on top of the SDK's Priority Nonce mempool.
|
||||
// It include's additional helper functions that allow users to determine if a
|
||||
// transaction is already in the mempool and to compare the priority of two
|
||||
// transactions.
|
||||
Mempool[C comparable] struct {
|
||||
// index defines an index of transactions.
|
||||
index sdkmempool.Mempool
|
||||
|
||||
// txPriority defines the transaction priority function. It is used to
|
||||
// retrieve the priority of a given transaction and to compare the priority
|
||||
// of two transactions. The index utilizes this struct to order transactions
|
||||
// in the mempool.
|
||||
txPriority TxPriority[C]
|
||||
|
||||
// txEncoder defines the sdk.Tx encoder that allows us to encode transactions
|
||||
// to bytes.
|
||||
txEncoder sdk.TxEncoder
|
||||
|
||||
// txCache is a map of all transactions in the mempool. It is used
|
||||
// to quickly check if a transaction is already in the mempool.
|
||||
txCache map[string]struct{}
|
||||
}
|
||||
)
|
||||
|
||||
// DefaultTxPriority returns a default implementation of the TxPriority. It prioritizes
|
||||
// transactions by their fee.
|
||||
func DefaultTxPriority() TxPriority[string] {
|
||||
return TxPriority[string]{
|
||||
GetTxPriority: func(goCtx context.Context, tx sdk.Tx) string {
|
||||
feeTx, ok := tx.(sdk.FeeTx)
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
|
||||
return feeTx.GetFee().String()
|
||||
},
|
||||
Compare: func(a, b string) int {
|
||||
aCoins, _ := sdk.ParseCoinsNormalized(a)
|
||||
bCoins, _ := sdk.ParseCoinsNormalized(b)
|
||||
|
||||
switch {
|
||||
case aCoins == nil && bCoins == nil:
|
||||
return 0
|
||||
|
||||
case aCoins == nil:
|
||||
return -1
|
||||
|
||||
case bCoins == nil:
|
||||
return 1
|
||||
|
||||
default:
|
||||
switch {
|
||||
case aCoins.IsAllGT(bCoins):
|
||||
return 1
|
||||
|
||||
case aCoins.IsAllLT(bCoins):
|
||||
return -1
|
||||
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
},
|
||||
MinValue: "",
|
||||
}
|
||||
}
|
||||
|
||||
// NewMempool returns a new ConstructorMempool.
|
||||
func NewMempool[C comparable](txPriority TxPriority[C], txEncoder sdk.TxEncoder, maxTx int) *Mempool[C] {
|
||||
return &Mempool[C]{
|
||||
index: NewPriorityMempool(
|
||||
PriorityNonceMempoolConfig[C]{
|
||||
TxPriority: txPriority,
|
||||
MaxTx: maxTx,
|
||||
},
|
||||
),
|
||||
txPriority: txPriority,
|
||||
txEncoder: txEncoder,
|
||||
txCache: make(map[string]struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// Insert inserts a transaction into the mempool.
|
||||
func (cm *Mempool[C]) Insert(ctx context.Context, tx sdk.Tx) error {
|
||||
if err := cm.index.Insert(ctx, tx); err != nil {
|
||||
return fmt.Errorf("failed to insert tx into auction index: %w", err)
|
||||
}
|
||||
|
||||
_, txHashStr, err := utils.GetTxHashStr(cm.txEncoder, tx)
|
||||
if err != nil {
|
||||
cm.Remove(tx)
|
||||
return err
|
||||
}
|
||||
|
||||
cm.txCache[txHashStr] = struct{}{}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Remove removes a transaction from the mempool.
|
||||
func (cm *Mempool[C]) Remove(tx sdk.Tx) error {
|
||||
if err := cm.index.Remove(tx); err != nil && !errors.Is(err, sdkmempool.ErrTxNotFound) {
|
||||
return fmt.Errorf("failed to remove transaction from the mempool: %w", err)
|
||||
}
|
||||
|
||||
_, txHashStr, err := utils.GetTxHashStr(cm.txEncoder, tx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get tx hash string: %w", err)
|
||||
}
|
||||
|
||||
delete(cm.txCache, txHashStr)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Select returns an iterator of all transactions in the mempool. NOTE: If you
|
||||
// remove a transaction from the mempool while iterating over the transactions,
|
||||
// the iterator will not be aware of the removal and will continue to iterate
|
||||
// over the removed transaction. Be sure to reset the iterator if you remove a transaction.
|
||||
func (cm *Mempool[C]) Select(ctx context.Context, txs [][]byte) sdkmempool.Iterator {
|
||||
return cm.index.Select(ctx, txs)
|
||||
}
|
||||
|
||||
// CountTx returns the number of transactions in the mempool.
|
||||
func (cm *Mempool[C]) CountTx() int {
|
||||
return cm.index.CountTx()
|
||||
}
|
||||
|
||||
// Contains returns true if the transaction is contained in the mempool.
|
||||
func (cm *Mempool[C]) Contains(tx sdk.Tx) bool {
|
||||
_, txHashStr, err := utils.GetTxHashStr(cm.txEncoder, tx)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
_, ok := cm.txCache[txHashStr]
|
||||
return ok
|
||||
}
|
||||
|
||||
// Compare determines the relative priority of two transactions belonging in the same lane.
|
||||
func (cm *Mempool[C]) Compare(ctx sdk.Context, this sdk.Tx, other sdk.Tx) int {
|
||||
firstPriority := cm.txPriority.GetTxPriority(ctx, this)
|
||||
secondPriority := cm.txPriority.GetTxPriority(ctx, other)
|
||||
return cm.txPriority.Compare(firstPriority, secondPriority)
|
||||
}
|
||||
@@ -0,0 +1,491 @@
|
||||
package base
|
||||
|
||||
// ------------------------------------------------------------------------------ //
|
||||
// ------------------------------------------------------------------------------ //
|
||||
// ------------------------------------------------------------------------------ //
|
||||
// ------------------------------------------------------------------------------ //
|
||||
// NOTE: THIS IS A COPY OF THE PRIORITY NONCE MEMPOOL FROM COSMOS-SDK. IT HAS BEEN
|
||||
// MODIFIED FOR OUR USE CASE. THIS CODE WILL BE DEPRECATED ONCE THE COSMOS-SDK
|
||||
// CUTS A FINAL v0.50.0 RELEASE.
|
||||
// ------------------------------------------------------------------------------ //
|
||||
// ------------------------------------------------------------------------------ //
|
||||
// ------------------------------------------------------------------------------ //
|
||||
// ------------------------------------------------------------------------------ //
|
||||
|
||||
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.priorityNode.Next() != nil {
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user