txpool

package
v1.9.7-0...-3c1a9ee Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 7, 2025 License: LGPL-3.0 Imports: 70 Imported by: 0

README

Functionality of Transaction Pool

Starting point of the design is the description of various functions that transaction pool needs to perform

Respond to tx fetch requests from network peers

As part of eth/65 protocol, there are messages GetPooledTransactions (there is also variant GetPooledTransactions66 that contains RequestID) that other peers can send to our node. These messages contain a list of transaction hashes. Our node is expected to reply with the list of transaction "bodies" (up to certain limit of reply message size). In order to compose such reply, tx pool needs to have function:

tx_hash => transaction

JSON RPC request asking for transaction by hash

RPC method eth_getTransactionByHash first looks the given hash up in the database, in the TxLookup index. If nothing found, it queries the tx pool for this hash. To respond to such query, tx pool needs the same function as above:

tx_hash => transaction

Send notifications about new pending transactions

There are RPC methods that allow subscribing to new pending transactions. First of all, we need to define what "new pending" transaction means. Our current understanding of this terminology is illustrated below.

Out of all transaction arriving at the pool, there are those that are valid on the current state. Good test for those is: "If we were to produce a block with unlimited block gas limit, as large as possible, which transactions would we include?". This set of transactions is shown as light green circle.

Transactions that are not valid on the current state, can be split further into three categories. Firstly, those that will never become valid unless there is some chain reorg. Secondly, those that may potentially become valid later. It can be because of their nonce that is too large, or because the balance currently on the sender account is deemed insufficient to pay for gas. These "potentially valid in the future" transactions are shown as pink circle. The rest can be thought if the space inside the "all transactions" circle but not within light green or pink circles, shown in light blue colour.

Due to limited amount of resources on the node (memory, CPU, network bandwidth), there is normally a limit on the total number or total size of the transactions that are actually stored by the node. This is shown by dark green (pending) and red (queued) circles. The rest of the transactions in the light green and pink areas are discarded.

Now the definition of a "new pending" transaction can be figured out by thinking about what happens when a new "remote" transaction arrives into the pool. It is assessed, firstly whether:

  1. It can never be valid - discarded immediately
  2. Is not valid currently, but can potentially be valid in the future - assigned to the pink circle
  3. Is valid currently - assigned to the light green circle

Then, if the transaction is assigned into the light green circle, we need to check whether it can be included into the "pending" dark green circle. Usually, if it is more attractive (in terms of perceived mining reward) than some other transactions in the dark green circle, it can be included, and some other transactions are evicted. There might be some exceptions to that, if the pending circle is limited by total size of the transactions rather than total number of transaction. If the new transaction is quite large, and even by pushing some other, less attractive transactions, it cannot fit, it would not be included.

At the end, if the new transaction is included into the dark green "pending" circle, a notification needs to be sent.

We can express this function this way perhaps:

transaction, pending_pool, queued_pool => pending_pool', queued_pool' bool

Given a transaction and pending pool, this function produces modified pending pool or modified queued pool, as well as indication whether this transaction was added to the pending pool, in which case a notification needs to be sent.

Retrieve pending and queued transaction

There are couple of RPC requests, such as txpool_content and txpool_inspect in go-ethereum, and parity_pendingTransactions, and parity_allTransactions open ethereum that require enumeration of pending and queued pools (dark green and red circles respectively).

We can express this as a function

=> pending_pool, queued_pool

Select transaction for mining

Before a new block gets mined, local and "remote" pending transactions are queried from the transaction pool. Local transactions (injected by JSON RPC methods into the node) get priority over "remote" transactions. If there are no local transactions left and there is still "space" left in the block, "remote" transactions are selected from the pending pool, in order of their priority, which is usually defined by much Wei/eth miner would receive for each unit of gas by including that transaction.

Note that although in the current implementation, ALL pending transactions are first retrieved, then sorted by priority, and then selected one-by-one, this is not a functional requirement. A different implementation that selects pending transactions in the order of priority, without fully sorting them, would be suitable as well.

We can describe this function as:

=> local_transactions, sorted_iterator(pending_pool)

Relaying transactions to other network peers

Here, the requirement is that transactions received from other peers via "gossip" messages, which are Transactions and NewPooledTransactionHashes, are relayed further. However, not all received transactions should be relayed. In order to prevent the node from becoming a part traffic amplification attacks on the network (where non-viable transactions are produced and relayed around the network, consuming bandwidth but never getting mined), only transactions from pending pool should be relayed. This requirement makes this particular function similar to the function of sending notifications about new pending transactions, i.e:

transaction, pending_pool, queued_pool => pending_pool', queued_pool' bool

Remove all mined transactions from the pool

When a block gets completely imported and it is deemed as "the best" block at its block height, all the transactions in that block need to be removed from the transaction pool, because they cannot be included twice (it will be prevented by the incrementing nonce rule). This functionality can be described as:

tx_hashes, pending_pool, queued_pool, local_transactions => pending_pool`, queued_pool`, local_transactions`

where modified pending pool and queued pool, and local transactions do not contain given transaction hashes anymore.

In code can find production of such event by p.newPendingTxns <-, and handling of this event by announcements := <-newTxns here

Reinject transactions into the transaction pool on unwinding a block

This can be thought of a reverse operation from the one described before. When a block that was deemed "the best" of its height, is no longer deemed "the best", the transactions contained in it, are now viable for inclusion in other blocks, and therefore should be returned into the transaction pool. We can describe this function as:

transactions, pending_pool, queued_pool => pending_pool`, queued_pool`

An interesting note here is that if the block contained any transactions local to the node, by being first removed from the pool (from the "local" part of it), and then re-injected, they effective lose their priority over the "remote" transactions. In order to prevent that, somehow the fact that certain transactions were local, needs to be remembered for some time (up to some "immutability threshold")

Question about persistence of the transactions

In the current implementation, transactions from the pool are persisted upon graceful shutdown. Upon restart, they are recovered and used for reducing the amount of reconciliation that is done via eth/65 pooled transactions. However, if we assume that transactions in the pool get refreshed relatively quickly, then the benefit of persistence for this particular purpose becomes uncertain.

On the other hand, if valid transactions are persisted in an append-only storage with some extra metadata, for example, time of arrival, this could be a powerful feature for historical data analysis. If such functionality is easy to enable, it should be provided.

In any event, local transactions need to be persisted, so that they do not lose their priority when they get included into a block but then reverted.

Another challenge (solvable) of persisting transactions (for example, local transactions) is that on the start up, we need to go through the recent blocks to figure out which ones have been already included. We cannot rely on the TxLookup index (mapping transaction hash to block number), because it is an optional index.

Changes for EIP-1559

With the EIP-1559, there are a few changes to the transaction pool model described above. Now it is possible for transactions to become executable or not without any changes to the sender's balance or nonce. This is because there is a rule that feeCap of a transaction needs to be no less than baseFee of the block, for the transaction to be includable. Another change is that under rules of EIP-1559, transaction needs to have the balance of at least feeCap x gasLimit, to be includable.

Design proposal

Given the functional analysis above and also some known challenges (for example, ensuring that transactions from the same senders are included in the strict order of their nonces, without gaps), the proposal is to have the following elements in the design: ordering function and subordinate pools, most consisting of a pair of priority queues. Description of the element follows

Ordering function

Ordering function is a relation that introduces total order over the set of all syntactically valid transactions. This means, for any two transaction, it tells us in which order their need to be sorted. The main idea of what such an ordering function should achieve is the stratification of transactions into sub-pools, like the ones shown above (pending, queued, etc.). For example, transactions belonging to pending sub-pool must be ordered above transactions in other sub-pools. Within pending sub-pool, the nonce increment rule needs to be enforced. Even looking at these two examples of requirements to the ordering function, we can determine that it is not enough to just look at the usual transaction attributes to order them. We'd need some supplemental data structures, and some extra "ephemeral" fields in the transactions.

First ephemeral field is an integer that acts as a bitfield, we call it SubPool. The bits in this field are (in the order from most significant to least significant):

  1. Minimum fee requirement. Set to 1 if feeCap of the transaction is no less than in-protocol parameter of minimal base fee. Set to 0 if feeCap is less than minimum base fee, which means this transaction will never be included into this particular chain.
  2. Absence of nonce gaps. Set to 1 for transactions whose nonce is N, state nonce for the sender is M, and there are transactions for all nonces between M and N from the same sender. Set to 0 is the transaction's nonce is divided from the state nonce by one or more nonce gaps.
  3. Sufficient balance for gas. Set to 1 if the balance of sender's account in the state is B, nonce of the sender in the state is M, nonce of the transaction is N, and the sum of feeCap x gasLimit + transferred_value of all transactions from this sender with nonces N+1 ... M is no more than B. Set to 0 otherwise. In other words, this bit is set if there is currently a guarantee that the transaction and all its required prior transactions will be able to pay for gas.
  4. Dynamic fee requirement. Set to 1 if feeCap of the transaction is no less than baseFee of the currently pending block. Set to 0 otherwise.
  5. Local transaction. Set to 1 if transaction is local.

Currently there are five bits in the SubPool ephemeral field, so it is an integer 0..31. These integers can be directly compared, and the largest value means a transaction is more preferable. This ephemeral field allows us to stratify all transactions into three sub pools (hence the name of the field).

There are three subpools - pending (green), basefee (yellow), and queued (red). Each subpool consists of two sorting data structures, one of which sorts transactions in the order of decreasing priority ("best" structure), another - in the order of increasing priority ("worst" structure). For yellow and red subpools, the sorting data structures are priority queues. For the green pool, the "best" structure could be priority queue, but it may be more convenient to have it as a list, to avoid frequent popping and pushing elements in and out of priority queue. Worst data structures serve to restrict the size of a subpool. The restriction can be defined either in terms of number of transactions, or in terms of total size of transactions (in bytes). Worst data structure allows taking the "worst" elements from the pool (and either demoting them into lower pools, or discarding them), that is why it needs to order transactions from the lowest priority to the highest. Best data structures for yellow and red subpools serve to promote transactions to higher pools. Best data structure in the green pool allows selection of a sub-set of highest priority transactions for the purposes of composing a block (mining or pending block API).

[RULES need to be updated for the additions of the local transaction bit]

The "tubes" (thick arrows) between the pools show how transactions move between the pools. Below are the rules of such movements:

  1. If top element in the worst green queue has SubPool != 0b1111 (binary), it needs to be removed from the green pool. If SubPool < 0b1000 (not satisfying minimum fee), discard. If SubPool == 0b1110, demote to the yellow pool, otherwise demote to the red pool.
  2. If top element in the worst green queue has SubPool == 0b1111, but there is not enough room in the pool, discard.
  3. If the top element in the best yellow queue has SubPool == 0b1111, promote to the green pool.
  4. If the top element in the worst yellow queue has SubPool != 0x1110, it needs to be removed from the yellow pool. If SubPool < 0b1000 (not satisfying minimum fee), discard. Otherwise, demote to the red pool.
  5. If the top element in the worst yellow queue has SubPool == 0x1110, but there is not enough room in the pool, discard.
  6. If the top element in the best red queue has SubPool == 0x1110, promote to the yellow pool. If SubPool == 0x1111, promote to the green pool.
  7. If the top element in the worst red queue has SubPool < 0b1000 (not satisfying minimum fee), discard.
  8. If the top element in the worst red queue has SubPool >= 0b1000, but there is not enough room in the pool, discard.

The rules above need to be checked once either a new transaction has appeared (it needs to be sorted into a sub-pool, then it might pop up at the top of the worst queue to be discarded, or push out another transaction), or if baseFee of the pending block changes. In the latter case, all priority queues need to have their invariants re-established (in Go, it is done by heap.Init function, which has linear algorithmic complexity). Also in the latter case, or if the new transaction ends up inhabiting the green pool, the best structure of the green pool needs to be re-sorted.

How is SubPool ephemeral field calculated?

It is quite easy to imagine how to calculate the bits 1 and 4 of the SubPool (minimum fee requirement and dynamic fee requirement). But it is not obvious how to calculate bits 2 and 3 (absence of nonce gaps and sufficient balance).

We introduce another couple of data structures alongside the sub pools. First will simply map address of an account to a senderId. Here, senderId would be uint64 for convenience:

senderIds :: address => senderId

Another data structure is a mapping of sender Id to:

  1. Account balance in the current state (before pending block)
  2. Account nonce in the current state (before pending block)
  3. Sorted map of nonce => transaction_record

So this structure looks like:

senders :: senderId => { state_balance; state_nonce; nonce =>(sorted) transaction_record }

Here, transaction_record is a pointer to a piece of data that represent a transaction in one of the sub-pools. Data structure senders allows us to react on the following events and modify the ephemeral field SubPool inside transaction_record:

  1. New best block arrives, which potentially changes the balance and the nonce of some senders. We use senderIds data structure to find relevant senderId values, and then use senders data structure to modify state_balance and state_nonce, potentially remove some elements (if transaction with some nonce is included into a block), and finally, walk over the transaction records and update SubPool fields depending on the actual presence of nonce gaps and what the balance is.
  2. New transaction arrives, and it may potentially replace existing one, shifting the balances required in all subsequent transactions from the same sender. Also, newly arrived transaction may fill the nonce gap. In both of these cases, we walk over the transaction records and update SubPool fields.
  3. Transaction is discarded. As it will be mentioned later, the ordering function is designed in such a way that transaction with a nonce N can only be discarded from the sub pools if all transactions with nonces higher than N have been discarded. Therefore, the only reaction on discarding a transaction is the deletion of corresponding entry from the mapping nonce => transaction_record.
Other parts of the ordering function

The field SubPool is only one part of the ordering function, the part that can be used to stratify transactions into sub pools, each having certain resource limit. Within every sub pool, there is a finer grained ordering. This ordering needs to server two objectives:

  1. In the green sub pool, any part of the sorted "best" structure, as long as it starts from the best transaction, can be taken and inserted in the block. Given sufficient gas limit in that block, all transactions taken in that way, must be includable. In particular, it means that strict nonce ordering of transactions from the same sender must be observed. Also, there needs to be enough balance in the sender's account to guarantee payment for gasLimit x feeCap + transferred_value for every transaction.
  2. In all sub pools, transactions with least priority need to show up on the top of the "worst" priority queues.

To satisfy the first objective, we need to another three ephemeral fields in each transactions.

Firstly, the cumulative required balance. It is the sum of gasLimit x feeCap + transferred_value for the transaction in question and all transactions from the sender that are required to be included before. The reason to have this field is to make sure we do not need to update every transaction when their sender balance change.

Secondly, the minimum feeCap for the transaction in question and all transactions from the sender that are required to be included before. This minimum value, and not the feeCap of the transaction in question is used for ordering.

Thirdly, the minimum tip for the transaction in question and all transactions from the sender that are required to be included before. This minimum value, and not the tip of the transaction in question, is used for ordering.

To illustrate the use of the fields described above, let us take an example. Given set of transactions:

TxId SenderId Nonce FeeCap Tip
1 A 2 23 12
2 A 3 45 10
3 A 4 22 15
4 B 1 30 14

We first compute the minimum of FeeCap and Tip for each of them, resulting in the following values:

TxId SenderId Nonce FeeCap Tip min(FeeCap) min(Tip)
1 A 2 23 12 min{23}=23 min{12}=12
2 A 3 45 10 min{45,23}=23 min{10,12}=10
3 A 4 22 15 min{22,45,23}=22 min{15,10,12}=10
4 B 1 30 14 min{30}=30 min{14}=14

Now, to demonstrate the use of these ephemeral fields in the ordering function, we need to pick baseFee. For first case, let us pick it so that all transactions can just pay full Tip to the miner. That value is baseFee = 11. In this case, so-called effectiveTip will be equal to min(Tip) for all transactions, and the ordering is as follows (from highest priority to lowest priority):

Tx 4 (effectiveTip 14)
Tx 1 (effectiveTip 12)
Tx 2 (effectiveTip 10, chosen over Tx 3 because of the nonce)
Tx 3 (effectiveTip 10)

If we ordered these transactions by Tip instead of min(Tip), the order would have been:

Tx 3, Tip 15, sender A, nonce 4
Tx 4, Tip 14, sender B, nonce 1
Tx 1, Tip 12, sender A, nonce 2
Tx 2, Tip 10, sender A, nonce 3

this ordering is incorrect, because Tx 3 cannot be included before both Tx 1 and Tx 2 are included. We could have enforced this relative ordering by comparing their nonces, like this:

Tx 1, Tip 12, sender A, nonce 2
Tx 2, Tip 10, sender A, nonce 3
Tx 4, Tip 14, sender B, nonce 1
Tx 3, Tip 15, sender A, nonce 4

but that would have been insufficient, because it would have violated the transitive property of the ordering function, which it must have to be correctly used in sorting structures. In this example we are lucky it did not cause inconsistency. But if the Tip of Tx 4 were 11 instead of 14, then we could have inserted it either before Tx 1 (because 11 < 12), or after Tx 2 (because 11 > 10). Using min(Tip) removes this inconsistency and is compatible with enforcing the strict ordering of the nonces.

If we now choose a slightly higher baseFee, lets say, 13, for transactions Tx 1 and Tx 3, effectiveTip becomes min(FeeCap) - baseFee instead of min(Tip) and the ordering is as follows:

Tx 4 (effectiveTip 14)
Tx 1 (effectiveTip 10, chosen over Tx 2 because of the nonce)
Tx 2 (effectiveTip 10)
Tx 3 (effectiveTip 9)

In fact, the ordering stays the same, but the effectiveTip changes. Still this ordering is consistent, and is compatible with enforcing strict ordering of nonces.

All three of these ephemeral fields: cumulative required balance, minimum fee cap and minimum tip, are recalculated upon arrival of new transactions from a certain sender, by walking the corresponding sorted map in the senders data structure.

Nuances of ordering in the yellow sub pool

We need to consider separately the ordering of the transactions by effectiveTip inside the yellow sub pool. Since transactions in the yellow sub pool have their FeeCap < baseFee, it means that their values of effectiveTip calculated as above, would be negative. To avoid negative numbers, we may instead modify the ordering function for this case to order by reverse order of min(FeeCap) (in yellow pool, the value of Tip is irrelevant). Also, having learnt about using min(FeeCap) instead of FeeCap, we can now adjust the condition for the "Dynamic Fee requirement" in the SubPool ephemeral field. It is set to 1 when min(FeeCap) is no less than baseFee of the pending block. This means that if a transaction satisfies the dynamic fee requirement, but its required predecessors do not, it will still be sorted into the yellow pool instead of the green sub pool.

Ordering in the red sub pool

Red sub pool contains transactions that may never be includable. And, since red sub pool has a limited amount of space, it needs to order transactions to maintain some notion of fairness to the users. Customary limitation so far in many implementation is how many transactions per unique sender can be queued (or, in our terminology, in the red sub pool). In order to approximate this limitation, but also make it more flexible (for example, if the red sub pool has plenty of space, why not keep more transactions per sender?), we should order transactions based on how far their nonces are from the state's nonce for the sender.

For example, we have two senders, A and B, and their nonces in the state are 13 and 20 respectively. Then, for these transactions:

TxId SenderId Nonce
1 A 18
2 A 20
3 B 26

we first calculate the distance from their state nonce:

TxId SenderId Nonce Distance from the state nonce
1 A 18 18-13 = 5
2 A 20 20-13 = 7
3 B 26 26-20 = 6

Then, the ordering becomes:

Tx 1, sender A, nonce 18
Tx 3, sender B, nonce 20
Tx 2, sender A, nonce 26

Using the combination of the ephemeral field for "cumulative required balance" and balance of the sender in the state, we can also introduce secondary ordering (for the case where distance from the state nonce is equal). If the nonce distances are equal, we would prefer transactions for which the gap between "cumulative required balance and the balance of the sender in the state", is lower. One small complication is that such gap may well be a negative number. In such cases, we can simply take the gap to be zero.

Testing strategy

Given the complexity of the ordering function and variety of rules for movements of transactions between the sub pools, it is likely that manually writing test cases would not be a very productive way of testing the implementation of the design proposed above. Therefore, we shall try to use fuzz testing. Generate a set of transaction pseudo-randomly. Use algorithms to sort them into the sub pools. Check invariants on the sub pools.

Coherence of the state cache and the state reads

This applies not just to transaction pool, but also to the RPC daemon, because it will benefits from the state cache. One of the main issues with the state cache (as it is presented in the KV interface) is the coherence between what is cached and what is queried from the remote database (over gRPC). Here is the suggestion on how to resolve it.

Firstly, it will require a change in the KV interface. Currently, the StateChange is defined like this:

message StateChange {
  Direction direction = 1;
  uint64 blockHeight = 2;
  types.H256 blockHash = 3;
...
}

and if we look at the implementation, we find that for the forward changes, we generate one StateChange message every single block. This means we can easily detect any missing blocks if gRPC transport loses a message for some reason. But for the unwind changes, everything is delivered in one "lump". For example, if unwind goes over 3 blocks, it will produce a single StateChange message and it won't be possible to detect any missing messages. Therefore, this is a suggested modification:

message StateChange {
  uint64 prevBlockHeight = 1;
  uint64 blockHeight = 2;
  types.H256 prevBlockHash = 3;
  types.H256 blockHash = 4;
...
}

Firstly, we now specified what was the expected block height and the block hash on top of which this state change should be applied. This makes sense because any such StateChange leads to valid (uncorrupted) state only if applied to specific previous state. Secondly, the indication of "direction" is not required anymore, because it should be obvious from the relationship between prevBlockHeight and blockHeight.

Having this modified interface, it would be possible to batch the forward updates for multiple blocks into one. But perhaps more importantly, it allows the "client side" (tx pool, or RPC daemon, or other system where state cache is used) to detect the lost messages, and invalidate the state cache if it happens.

State Cache works on top of Database Transaction and pair Cache+ReadTransaction must provide "Serializable Isolation Level" semantic: all data form consistent db view at moment when read transaction started, read data are immutable until end of read transaction, reader can't see newer updates.

When the stream of state updates come, we maintain an ordered list of their identifiers (can be combination/concatenation of block height and block hash). For each element in such list, we keep the pointer to the cache (for example, if the cache is represented as google B-Tree, we can keep *BTree pointer for each identifier, even though the majority of cache content will be shared). To ensure the synchronisation between the cache and the remote state query (via read-only transactions), we also have a map of conditional variables, with the keys being the identifiers. These conditional variables are placed into the map by read-only transactions (to wait for corresponding cache entry), and notified and removed from the map by the arrival of state updates.

Every time a new state change comes, we do the following:

  1. Check that prevBlockHeight and prevBlockHash match what is the top values we have, and if they don't we invalidate the cache, because we missed some messages and cannot consider the cache coherent anymore.
  2. Clone the cache pointer (such that the previous pointer is still accessible, but new one shared the content with it), apply state updates to the cloned cache pointer and save under the new identified made from blockHeight and blockHash.
  3. If there is a conditional variable corresponding to the identifier, remove it from the map and notify conditional variable, waking up the read-only transaction waiting on it.

On the other hand, whenever we have a cache miss (by looking at the top cache), we do the following:

  1. Once read the current block height and block hash (canonical) from underlying db transaction (for example by golang.org/x/sync/singleflight, or remoteKv server may send us this data when we open Tx)
  2. Construct the identifier from the current block height and block hash
  3. Look for the constructed identifier in the cache. If the identifier is found, use the corresponding cache in conjunction with this read-only transaction (it will be consistent with it). If the identifier is not found, it means that the transaction has been committed in Erigon, but the state update has not arrived yet (as shown in the picture on the right). Insert conditional variable for this identifier and wait on it until either cache with the given identifier appears, or timeout (indicating that the cache update mechanism is broken and cache is likely invalidated).

TODO: Describe the cleanup of the cache identifiers, describe exact the locking of the map of cache pointers and the map of conditional variables. Perhaps there just need to be a single map of structs like that:

type CacheRoot struct {
   cache *btree.BTree
   lock sync.RWMutex
   cond *sync.Cond
}
var roots map[string]CacheRoot
var rootsLock sync.RWMutex

where the keys in the map are concatenation of blockHeight (as [8]byte) and blockHash.

Documentation

Overview

Package txpool is a generated GoMock package.

Index

Constants

View Source 
const (
	LegacyTxnType     byte = 0
	AccessListTxnType byte = 1 // EIP-2930
	DynamicFeeTxnType byte = 2 // EIP-1559
	BlobTxnType       byte = 3 // EIP-4844
	SetCodeTxnType    byte = 4 // EIP-7702
)
View Source 
const DefaultBlockGasLimit = uint64(30000000)

Variables

View Source 
var ErrAlreadyKnown = errors.New("already known")
View Source 
var ErrParseTxn = fmt.Errorf("%w transaction", rlp.ErrParse)
View Source 
var ErrPoolDisabled = errors.New("TxPool Disabled")
View Source 
var ErrRejected = errors.New("rejected")
View Source 
var ErrRlpTooBig = errors.New("txn rlp too big")
View Source 
var PoolChainConfigKey = []byte("chain_config")
View Source 
var PoolLastSeenBlockKey = []byte("last_seen_block")
View Source 
var PoolPendingBaseFeeKey = []byte("pending_base_fee")
View Source 
var PoolPendingBlobFeeKey = []byte("pending_blob_fee")
View Source 
var PoolStateVersion = []byte("state_version")
View Source 
var TxPoolAPIVersion = &typesproto.VersionReply{Major: 1, Minor: 0, Patch: 0}

TxPoolAPIVersion

Functions

func ChainConfig 

func ChainConfig(tx kv.Getter) (*chain.Config, error)

func EncodeGetPooledTransactions66 

func EncodeGetPooledTransactions66(hashes []byte, requestID uint64, encodeBuf []byte) ([]byte, error)

EncodeGetPooledTransactions66 produces encoding of GetPooledTransactions66 packet

func EncodeHashes 

func EncodeHashes(hashes []byte, encodeBuf []byte) []byte

EncodeHashes produces RLP encoding of given number of hashes, as RLP list It appends encoding to the given given slice (encodeBuf), reusing the space there is enough capacity. The first returned value is the slice where encodinfg

func EncodePooledTransactions66 

func EncodePooledTransactions66(txnsRlp [][]byte, requestID uint64, encodeBuf []byte) []byte

TODO(eip-4844) wrappedWithBlobs = true?

func EncodeSender 

func EncodeSender(nonce uint64, balance uint256.Int, buffer []byte)

Encode the details of txn sender into the given "buffer" byte-slice that should be big enough

func EncodeSenderLengthForStorage 

func EncodeSenderLengthForStorage(nonce uint64, balance uint256.Int) uint

func EncodeTransactions 

func EncodeTransactions(txnsRlp [][]byte, encodeBuf []byte) []byte

TODO(eip-4844) wrappedWithBlobs = false?

func LastSeenBlock 

func LastSeenBlock(tx kv.Getter) (uint64, error)

func ParseGetPooledTransactions66 

func ParseGetPooledTransactions66(payload []byte, pos int, hashbuf []byte) (requestID uint64, hashes []byte, newPos int, err error)

func ParseHash 

func ParseHash(payload []byte, pos int, hashbuf []byte) ([]byte, int, error)

ParseHash extracts the next hash from the RLP encoding (payload) from a given position. It appends the hash to the given slice, reusing the space if there is enough capacity The first returned value is the slice where hash is appended to. The second returned value is the new position in the RLP payload after the extraction of the hash.

func ParseHashesCount 

func ParseHashesCount(payload []byte, pos int) (count int, dataPos int, err error)

ParseHashesCount looks at the RLP length Prefix for list of 32-byte hashes and returns number of hashes in the list to expect

func ParsePooledTransactions66 

func ParsePooledTransactions66(payload []byte, pos int, ctx *TxnParseContext, txnSlots *TxnSlots, validateHash func([]byte) error) (requestID uint64, newPos int, err error)

func ParseTransactions 

func ParseTransactions(payload []byte, pos int, ctx *TxnParseContext, txnSlots *TxnSlots, validateHash func([]byte) error) (newPos int, err error)

func PeekTransactionType 

func PeekTransactionType(serialized []byte) (byte, error)

func PutChainConfig 

func PutChainConfig(tx kv.Putter, cc *chain.Config, buf []byte) error

func PutLastSeenBlock 

func PutLastSeenBlock(tx kv.Putter, n uint64, buf []byte) error

func SaveChainConfigIfNeed 

func SaveChainConfigIfNeed(
	ctx context.Context,
	coreDB kv.RoDB,
	poolDB kv.RwDB,
	force bool,
	logger log.Logger,
) (cc *chain.Config, blockNum uint64, err error)

func SortByNonceLess 

func SortByNonceLess(a, b *metaTxn) bool

func StartGrpc 

func StartGrpc(txPoolServer txpool_proto.TxpoolServer, miningServer txpool_proto.MiningServer, addr string, creds *credentials.TransportCredentials, logger log.Logger) (*grpc.Server, error)

Types

type Addresses 

type Addresses []byte // flatten list of 20-byte addresses

func (Addresses) AddressAt 

func (h Addresses) AddressAt(i int) common.Address

AddressAt returns an address at the given index in the flattened list. Use this method if you want to reduce memory allocations

func (Addresses) At 

func (h Addresses) At(i int) []byte

func (Addresses) Len 

func (h Addresses) Len() int

type Announcements 

type Announcements struct {
	// contains filtered or unexported fields
}

func (*Announcements) Append 

func (a *Announcements) Append(t byte, size uint32, hash []byte)

func (*Announcements) AppendOther 

func (a *Announcements) AppendOther(other Announcements)

func (Announcements) At 

func (a Announcements) At(i int) (byte, uint32, []byte)

func (Announcements) Copy 

func (a Announcements) Copy() Announcements

func (Announcements) DedupCopy 

func (a Announcements) DedupCopy() Announcements

DedupCopy sorts hashes, and creates deduplicated copy

func (Announcements) DedupHashes 

func (a Announcements) DedupHashes() Hashes

func (Announcements) Hashes 

func (a Announcements) Hashes() Hashes

func (Announcements) Len 

func (a Announcements) Len() int

func (Announcements) Less 

func (a Announcements) Less(i, j int) bool

func (*Announcements) Reset 

func (a *Announcements) Reset()

func (Announcements) Swap 

func (a Announcements) Swap(i, j int)

type BestQueue 

type BestQueue struct {
	// contains filtered or unexported fields
}

func (*BestQueue) Len 

func (p *BestQueue) Len() int

func (*BestQueue) Less 

func (p *BestQueue) Less(i, j int) bool

func (*BestQueue) Pop 

func (p *BestQueue) Pop() interface{}

func (*BestQueue) Push 

func (p *BestQueue) Push(x interface{})

func (*BestQueue) Swap 

func (p *BestQueue) Swap(i, j int)

type BySenderAndNonce 

type BySenderAndNonce struct {
	// contains filtered or unexported fields
}

BySenderAndNonce - designed to perform most expensive operation in TxPool: "recalculate all ephemeral fields of all transactions" by algo

  • for all senders - iterate over all transactions in nonce growing order

Performances decisions:

  • All senders stored inside 1 large BTree - because iterate over 1 BTree is faster than over map[senderId]BTree
  • sortByNonce used as non-pointer wrapper - because iterate over BTree of pointers is 2x slower

type FeeCalculator 

type FeeCalculator interface {
	CurrentFees(chainConfig *chain.Config, db kv.Getter) (baseFee uint64, blobFee uint64, minBlobGasPrice, blockGasLimit uint64, err error)
}

type Fetch 

type Fetch struct {
	// contains filtered or unexported fields
}

Fetch connects to sentry and implements eth/66 protocol regarding the transaction messages. It tries to "prime" the sentry with StatusData message containing given genesis hash and list of forks, but with zero max block and total difficulty Sentry should have a logic not to overwrite statusData with messages from txn pool

func NewFetch 

func NewFetch(
	ctx context.Context,
	sentryClients []sentry.SentryClient,
	pool Pool,
	stateChangesClient StateChangesClient,
	db kv.RwDB,
	chainID uint256.Int,
	logger log.Logger,
	opts ...Option,
) *Fetch

NewFetch creates a new fetch object that will work with given sentry clients. Since the SentryClient here is an interface, it is suitable for mocking in tests (mock will need to implement all the functions of the SentryClient interface).

func (*Fetch) ConnectCore 

func (f *Fetch) ConnectCore()

func (*Fetch) ConnectSentries 

func (f *Fetch) ConnectSentries()

ConnectSentries initialises connection to the sentry

type GrpcDisabled 

type GrpcDisabled struct {
	txpool_proto.UnimplementedTxpoolServer
}

func (*GrpcDisabled) Add 

func (*GrpcDisabled) Add(ctx context.Context, request *txpool_proto.AddRequest) (*txpool_proto.AddReply, error)

func (*GrpcDisabled) All 

func (*GrpcDisabled) All(ctx context.Context, request *txpool_proto.AllRequest) (*txpool_proto.AllReply, error)

func (*GrpcDisabled) FindUnknown 

func (*GrpcDisabled) FindUnknown(ctx context.Context, hashes *txpool_proto.TxHashes) (*txpool_proto.TxHashes, error)

func (*GrpcDisabled) Nonce 

func (*GrpcDisabled) Nonce(ctx context.Context, request *txpool_proto.NonceRequest) (*txpool_proto.NonceReply, error)

func (*GrpcDisabled) OnAdd 

func (*GrpcDisabled) OnAdd(request *txpool_proto.OnAddRequest, server txpool_proto.Txpool_OnAddServer) error

func (*GrpcDisabled) Pending 

func (*GrpcDisabled) Pending(ctx context.Context, empty *emptypb.Empty) (*txpool_proto.PendingReply, error)

func (*GrpcDisabled) Status 

func (*GrpcDisabled) Status(ctx context.Context, request *txpool_proto.StatusRequest) (*txpool_proto.StatusReply, error)

func (*GrpcDisabled) Transactions 

func (*GrpcDisabled) Transactions(ctx context.Context, request *txpool_proto.TransactionsRequest) (*txpool_proto.TransactionsReply, error)

func (*GrpcDisabled) Version 

func (*GrpcDisabled) Version(ctx context.Context, empty *emptypb.Empty) (*typesproto.VersionReply, error)

type GrpcServer 

type GrpcServer struct {
	txpool_proto.UnimplementedTxpoolServer
	// contains filtered or unexported fields
}

func NewGrpcServer 

func NewGrpcServer(ctx context.Context, txPool txPool, db kv.RoDB, newSlotsStreams *NewSlotsStreams, chainID uint256.Int, logger log.Logger) *GrpcServer

func (*GrpcServer) Add 

func (s *GrpcServer) Add(ctx context.Context, in *txpool_proto.AddRequest) (*txpool_proto.AddReply, error)

func (*GrpcServer) All 

func (s *GrpcServer) All(ctx context.Context, _ *txpool_proto.AllRequest) (*txpool_proto.AllReply, error)

func (*GrpcServer) FindUnknown 

func (s *GrpcServer) FindUnknown(ctx context.Context, in *txpool_proto.TxHashes) (*txpool_proto.TxHashes, error)

func (*GrpcServer) Nonce 

func (s *GrpcServer) Nonce(ctx context.Context, in *txpool_proto.NonceRequest) (*txpool_proto.NonceReply, error)

returns nonce for address

func (*GrpcServer) OnAdd 

func (s *GrpcServer) OnAdd(req *txpool_proto.OnAddRequest, stream txpool_proto.Txpool_OnAddServer) error

func (*GrpcServer) Pending 

func (s *GrpcServer) Pending(ctx context.Context, _ *emptypb.Empty) (*txpool_proto.PendingReply, error)

func (*GrpcServer) Status 

func (s *GrpcServer) Status(_ context.Context, _ *txpool_proto.StatusRequest) (*txpool_proto.StatusReply, error)

func (*GrpcServer) Transactions 

func (s *GrpcServer) Transactions(ctx context.Context, in *txpool_proto.TransactionsRequest) (*txpool_proto.TransactionsReply, error)

func (*GrpcServer) Version 

func (s *GrpcServer) Version(context.Context, *emptypb.Empty) (*typesproto.VersionReply, error)

type Hashes 

type Hashes []byte // flatten list of 32-byte hashes

func (Hashes) At 

func (h Hashes) At(i int) []byte

func (Hashes) DedupCopy 

func (h Hashes) DedupCopy() Hashes

DedupCopy sorts hashes, and creates deduplicated copy

func (Hashes) Len 

func (h Hashes) Len() int

func (Hashes) Less 

func (h Hashes) Less(i, j int) bool

func (Hashes) Swap 

func (h Hashes) Swap(i, j int)

type MockPool 

type MockPool struct {
	// contains filtered or unexported fields
}

MockPool is a mock of Pool interface.

func NewMockPool 

func NewMockPool(ctrl *gomock.Controller) *MockPool

NewMockPool creates a new mock instance.

func (*MockPool) AddLocalTxns 

func (m *MockPool) AddLocalTxns(ctx context.Context, newTxns TxnSlots) ([]txpoolcfg.DiscardReason, error)

AddLocalTxns mocks base method.

func (*MockPool) AddNewGoodPeer 

func (m *MockPool) AddNewGoodPeer(peerID PeerID)

AddNewGoodPeer mocks base method.

func (*MockPool) AddRemoteTxns 

func (m *MockPool) AddRemoteTxns(ctx context.Context, newTxns TxnSlots)

AddRemoteTxns mocks base method.

func (*MockPool) EXPECT 

func (m *MockPool) EXPECT() *MockPoolMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockPool) FilterKnownIdHashes 

func (m *MockPool) FilterKnownIdHashes(tx kv.Tx, hashes Hashes) (Hashes, error)

FilterKnownIdHashes mocks base method.

func (*MockPool) GetRlp 

func (m *MockPool) GetRlp(tx kv.Tx, hash []byte) ([]byte, error)

GetRlp mocks base method.

func (*MockPool) IdHashKnown 

func (m *MockPool) IdHashKnown(tx kv.Tx, hash []byte) (bool, error)

IdHashKnown mocks base method.

func (*MockPool) OnNewBlock 

func (m *MockPool) OnNewBlock(ctx context.Context, stateChanges *remoteproto.StateChangeBatch, unwindTxns, unwindBlobTxns, minedTxns TxnSlots) error

OnNewBlock mocks base method.

func (*MockPool) Started 

func (m *MockPool) Started() bool

Started mocks base method.

func (*MockPool) ValidateSerializedTxn 

func (m *MockPool) ValidateSerializedTxn(serializedTxn []byte) error

ValidateSerializedTxn mocks base method.

type MockPoolAddLocalTxnsCall 

type MockPoolAddLocalTxnsCall struct {
	*gomock.Call
}

MockPoolAddLocalTxnsCall wrap *gomock.Call

func (*MockPoolAddLocalTxnsCall) Do 

func (c *MockPoolAddLocalTxnsCall) Do(f func(context.Context, TxnSlots) ([]txpoolcfg.DiscardReason, error)) *MockPoolAddLocalTxnsCall

Do rewrite *gomock.Call.Do

func (*MockPoolAddLocalTxnsCall) DoAndReturn 

func (c *MockPoolAddLocalTxnsCall) DoAndReturn(f func(context.Context, TxnSlots) ([]txpoolcfg.DiscardReason, error)) *MockPoolAddLocalTxnsCall

DoAndReturn rewrite *gomock.Call.DoAndReturn

func (*MockPoolAddLocalTxnsCall) Return 

func (c *MockPoolAddLocalTxnsCall) Return(arg0 []txpoolcfg.DiscardReason, arg1 error) *MockPoolAddLocalTxnsCall

Return rewrite *gomock.Call.Return

type MockPoolAddNewGoodPeerCall 

type MockPoolAddNewGoodPeerCall struct {
	*gomock.Call
}

MockPoolAddNewGoodPeerCall wrap *gomock.Call

func (*MockPoolAddNewGoodPeerCall) Do 

func (c *MockPoolAddNewGoodPeerCall) Do(f func(PeerID)) *MockPoolAddNewGoodPeerCall

Do rewrite *gomock.Call.Do

func (*MockPoolAddNewGoodPeerCall) DoAndReturn 

func (c *MockPoolAddNewGoodPeerCall) DoAndReturn(f func(PeerID)) *MockPoolAddNewGoodPeerCall

DoAndReturn rewrite *gomock.Call.DoAndReturn

func (*MockPoolAddNewGoodPeerCall) Return 

func (c *MockPoolAddNewGoodPeerCall) Return() *MockPoolAddNewGoodPeerCall

Return rewrite *gomock.Call.Return

type MockPoolAddRemoteTxnsCall 

type MockPoolAddRemoteTxnsCall struct {
	*gomock.Call
}

MockPoolAddRemoteTxnsCall wrap *gomock.Call

func (*MockPoolAddRemoteTxnsCall) Do 

func (c *MockPoolAddRemoteTxnsCall) Do(f func(context.Context, TxnSlots)) *MockPoolAddRemoteTxnsCall

Do rewrite *gomock.Call.Do

func (*MockPoolAddRemoteTxnsCall) DoAndReturn 

func (c *MockPoolAddRemoteTxnsCall) DoAndReturn(f func(context.Context, TxnSlots)) *MockPoolAddRemoteTxnsCall

DoAndReturn rewrite *gomock.Call.DoAndReturn

func (*MockPoolAddRemoteTxnsCall) Return 

func (c *MockPoolAddRemoteTxnsCall) Return() *MockPoolAddRemoteTxnsCall

Return rewrite *gomock.Call.Return

type MockPoolFilterKnownIdHashesCall 

type MockPoolFilterKnownIdHashesCall struct {
	*gomock.Call
}

MockPoolFilterKnownIdHashesCall wrap *gomock.Call

func (*MockPoolFilterKnownIdHashesCall) Do 

func (c *MockPoolFilterKnownIdHashesCall) Do(f func(kv.Tx, Hashes) (Hashes, error)) *MockPoolFilterKnownIdHashesCall

Do rewrite *gomock.Call.Do

func (*MockPoolFilterKnownIdHashesCall) DoAndReturn 

func (c *MockPoolFilterKnownIdHashesCall) DoAndReturn(f func(kv.Tx, Hashes) (Hashes, error)) *MockPoolFilterKnownIdHashesCall

DoAndReturn rewrite *gomock.Call.DoAndReturn

func (*MockPoolFilterKnownIdHashesCall) Return 

func (c *MockPoolFilterKnownIdHashesCall) Return(unknownHashes Hashes, err error) *MockPoolFilterKnownIdHashesCall

Return rewrite *gomock.Call.Return

type MockPoolGetRlpCall 

type MockPoolGetRlpCall struct {
	*gomock.Call
}

MockPoolGetRlpCall wrap *gomock.Call

func (*MockPoolGetRlpCall) Do 

func (c *MockPoolGetRlpCall) Do(f func(kv.Tx, []byte) ([]byte, error)) *MockPoolGetRlpCall

Do rewrite *gomock.Call.Do

func (*MockPoolGetRlpCall) DoAndReturn 

func (c *MockPoolGetRlpCall) DoAndReturn(f func(kv.Tx, []byte) ([]byte, error)) *MockPoolGetRlpCall

DoAndReturn rewrite *gomock.Call.DoAndReturn

func (*MockPoolGetRlpCall) Return 

func (c *MockPoolGetRlpCall) Return(arg0 []byte, arg1 error) *MockPoolGetRlpCall

Return rewrite *gomock.Call.Return

type MockPoolIdHashKnownCall 

type MockPoolIdHashKnownCall struct {
	*gomock.Call
}

MockPoolIdHashKnownCall wrap *gomock.Call

func (*MockPoolIdHashKnownCall) Do 

func (c *MockPoolIdHashKnownCall) Do(f func(kv.Tx, []byte) (bool, error)) *MockPoolIdHashKnownCall

Do rewrite *gomock.Call.Do

func (*MockPoolIdHashKnownCall) DoAndReturn 

func (c *MockPoolIdHashKnownCall) DoAndReturn(f func(kv.Tx, []byte) (bool, error)) *MockPoolIdHashKnownCall

DoAndReturn rewrite *gomock.Call.DoAndReturn

func (*MockPoolIdHashKnownCall) Return 

func (c *MockPoolIdHashKnownCall) Return(arg0 bool, arg1 error) *MockPoolIdHashKnownCall

Return rewrite *gomock.Call.Return

type MockPoolMockRecorder 

type MockPoolMockRecorder struct {
	// contains filtered or unexported fields
}

MockPoolMockRecorder is the mock recorder for MockPool.

func (*MockPoolMockRecorder) AddLocalTxns 

func (mr *MockPoolMockRecorder) AddLocalTxns(ctx, newTxns any) *MockPoolAddLocalTxnsCall

AddLocalTxns indicates an expected call of AddLocalTxns.

func (*MockPoolMockRecorder) AddNewGoodPeer 

func (mr *MockPoolMockRecorder) AddNewGoodPeer(peerID any) *MockPoolAddNewGoodPeerCall

AddNewGoodPeer indicates an expected call of AddNewGoodPeer.

func (*MockPoolMockRecorder) AddRemoteTxns 

func (mr *MockPoolMockRecorder) AddRemoteTxns(ctx, newTxns any) *MockPoolAddRemoteTxnsCall

AddRemoteTxns indicates an expected call of AddRemoteTxns.

func (*MockPoolMockRecorder) FilterKnownIdHashes 

func (mr *MockPoolMockRecorder) FilterKnownIdHashes(tx, hashes any) *MockPoolFilterKnownIdHashesCall

FilterKnownIdHashes indicates an expected call of FilterKnownIdHashes.

func (*MockPoolMockRecorder) GetRlp 

func (mr *MockPoolMockRecorder) GetRlp(tx, hash any) *MockPoolGetRlpCall

GetRlp indicates an expected call of GetRlp.

func (*MockPoolMockRecorder) IdHashKnown 

func (mr *MockPoolMockRecorder) IdHashKnown(tx, hash any) *MockPoolIdHashKnownCall

IdHashKnown indicates an expected call of IdHashKnown.

func (*MockPoolMockRecorder) OnNewBlock 

func (mr *MockPoolMockRecorder) OnNewBlock(ctx, stateChanges, unwindTxns, unwindBlobTxns, minedTxns any) *MockPoolOnNewBlockCall

OnNewBlock indicates an expected call of OnNewBlock.

func (*MockPoolMockRecorder) Started 

func (mr *MockPoolMockRecorder) Started() *MockPoolStartedCall

Started indicates an expected call of Started.

func (*MockPoolMockRecorder) ValidateSerializedTxn 

func (mr *MockPoolMockRecorder) ValidateSerializedTxn(serializedTxn any) *MockPoolValidateSerializedTxnCall

ValidateSerializedTxn indicates an expected call of ValidateSerializedTxn.

type MockPoolOnNewBlockCall 

type MockPoolOnNewBlockCall struct {
	*gomock.Call
}

MockPoolOnNewBlockCall wrap *gomock.Call

func (*MockPoolOnNewBlockCall) Do 

func (c *MockPoolOnNewBlockCall) Do(f func(context.Context, *remoteproto.StateChangeBatch, TxnSlots, TxnSlots, TxnSlots) error) *MockPoolOnNewBlockCall

Do rewrite *gomock.Call.Do

func (*MockPoolOnNewBlockCall) DoAndReturn 

func (c *MockPoolOnNewBlockCall) DoAndReturn(f func(context.Context, *remoteproto.StateChangeBatch, TxnSlots, TxnSlots, TxnSlots) error) *MockPoolOnNewBlockCall

DoAndReturn rewrite *gomock.Call.DoAndReturn

func (*MockPoolOnNewBlockCall) Return 

func (c *MockPoolOnNewBlockCall) Return(arg0 error) *MockPoolOnNewBlockCall

Return rewrite *gomock.Call.Return

type MockPoolStartedCall 

type MockPoolStartedCall struct {
	*gomock.Call
}

MockPoolStartedCall wrap *gomock.Call

func (*MockPoolStartedCall) Do 

func (c *MockPoolStartedCall) Do(f func() bool) *MockPoolStartedCall

Do rewrite *gomock.Call.Do

func (*MockPoolStartedCall) DoAndReturn 

func (c *MockPoolStartedCall) DoAndReturn(f func() bool) *MockPoolStartedCall

DoAndReturn rewrite *gomock.Call.DoAndReturn

func (*MockPoolStartedCall) Return 

func (c *MockPoolStartedCall) Return(arg0 bool) *MockPoolStartedCall

Return rewrite *gomock.Call.Return

type MockPoolValidateSerializedTxnCall 

type MockPoolValidateSerializedTxnCall struct {
	*gomock.Call
}

MockPoolValidateSerializedTxnCall wrap *gomock.Call

func (*MockPoolValidateSerializedTxnCall) Do 

func (c *MockPoolValidateSerializedTxnCall) Do(f func([]byte) error) *MockPoolValidateSerializedTxnCall

Do rewrite *gomock.Call.Do

func (*MockPoolValidateSerializedTxnCall) DoAndReturn 

func (c *MockPoolValidateSerializedTxnCall) DoAndReturn(f func([]byte) error) *MockPoolValidateSerializedTxnCall

DoAndReturn rewrite *gomock.Call.DoAndReturn

func (*MockPoolValidateSerializedTxnCall) Return 

func (c *MockPoolValidateSerializedTxnCall) Return(arg0 error) *MockPoolValidateSerializedTxnCall

Return rewrite *gomock.Call.Return

type NewPooledTransactionHashesPacket 

type NewPooledTransactionHashesPacket [][length.Hash]byte

type NewSlotsStreams 

type NewSlotsStreams struct {
	// contains filtered or unexported fields
}

NewSlotsStreams - it's safe to use this class as non-pointer

func (*NewSlotsStreams) Add 

func (s *NewSlotsStreams) Add(stream txpool_proto.Txpool_OnAddServer) (remove func())

func (*NewSlotsStreams) Broadcast 

func (s *NewSlotsStreams) Broadcast(reply *txpool_proto.OnAddReply, logger log.Logger)

type Option 

type Option func(*options)

func WithFeeCalculator 

func WithFeeCalculator(f FeeCalculator) Option

func WithP2PFetcherWg 

func WithP2PFetcherWg(wg *sync.WaitGroup) Option

func WithP2PSenderWg 

func WithP2PSenderWg(wg *sync.WaitGroup) Option

func WithPoolDBInitializer 

func WithPoolDBInitializer(init poolDBInitializer) Option

type PeerID 

type PeerID *typesproto.H512

type PendingPool 

type PendingPool struct {
	// contains filtered or unexported fields
}

PendingPool - is different from other pools - it's best is Slice instead of Heap It's more expensive to maintain "slice sort" invariant, but it allow do cheap copy of pending.best slice for mining (because we consider txns and metaTxn are immutable)

func NewPendingSubPool 

func NewPendingSubPool(t SubPoolType, limit int) *PendingPool

func (*PendingPool) Add 

func (p *PendingPool) Add(i *metaTxn, logger log.Logger)

func (*PendingPool) Best 

func (p *PendingPool) Best() *metaTxn

func (*PendingPool) DebugPrint 

func (p *PendingPool) DebugPrint(prefix string)

func (*PendingPool) EnforceBestInvariants 

func (p *PendingPool) EnforceBestInvariants()

func (*PendingPool) EnforceWorstInvariants 

func (p *PendingPool) EnforceWorstInvariants()

func (*PendingPool) Len 

func (p *PendingPool) Len() int

func (*PendingPool) PopWorst 

func (p *PendingPool) PopWorst() *metaTxn

func (*PendingPool) Remove 

func (p *PendingPool) Remove(i *metaTxn, reason string, logger log.Logger)

func (*PendingPool) Updated 

func (p *PendingPool) Updated(mt *metaTxn)

func (*PendingPool) Worst 

func (p *PendingPool) Worst() *metaTxn

type Pool 

type Pool interface {
	ValidateSerializedTxn(serializedTxn []byte) error

	// Handle 3 main events - new remote txns from p2p, new local txns from RPC, new blocks from execution layer
	AddRemoteTxns(ctx context.Context, newTxns TxnSlots)
	AddLocalTxns(ctx context.Context, newTxns TxnSlots) ([]txpoolcfg.DiscardReason, error)
	OnNewBlock(ctx context.Context, stateChanges *remote.StateChangeBatch, unwindTxns, unwindBlobTxns, minedTxns TxnSlots) error
	// IdHashKnown check whether transaction with given Id hash is known to the pool
	IdHashKnown(tx kv.Tx, hash []byte) (bool, error)
	FilterKnownIdHashes(tx kv.Tx, hashes Hashes) (unknownHashes Hashes, err error)
	Started() bool
	GetRlp(tx kv.Tx, hash []byte) ([]byte, error)

	AddNewGoodPeer(peerID PeerID)
}

Pool is interface for the transaction pool This interface exists for the convenience of testing, and not yet because there are multiple implementations

type Send 

type Send struct {
	// contains filtered or unexported fields
}

Send - does send concrete P2P messages to Sentry. Same as Fetch but for outbound traffic does not initiate any messages by self

func NewSend 

func NewSend(ctx context.Context, sentryClients []sentryproto.SentryClient, logger log.Logger, opts ...Option) *Send

func (*Send) AnnouncePooledTxns 

func (f *Send) AnnouncePooledTxns(types []byte, sizes []uint32, hashes Hashes, maxPeers uint64) (hashSentTo []int)

func (*Send) BroadcastPooledTxns 

func (f *Send) BroadcastPooledTxns(rlps [][]byte, maxPeers uint64) (txnSentTo []int)

Broadcast given RLPs to random peers

func (*Send) PropagatePooledTxnsToPeersList 

func (f *Send) PropagatePooledTxnsToPeersList(peers []PeerID, types []byte, sizes []uint32, hashes []byte)

type Signature 

type Signature struct {
	ChainID uint256.Int
	V       uint256.Int
	R       uint256.Int
	S       uint256.Int
}

type StateChangesClient 

type StateChangesClient interface {
	StateChanges(ctx context.Context, in *remote.StateChangeRequest, opts ...grpc.CallOption) (remote.KV_StateChangesClient, error)
}

type SubPool 

type SubPool struct {
	// contains filtered or unexported fields
}

func NewSubPool 

func NewSubPool(t SubPoolType, limit int) *SubPool

func (*SubPool) Add 

func (p *SubPool) Add(i *metaTxn, reason string, logger log.Logger)

func (*SubPool) Best 

func (p *SubPool) Best() *metaTxn

func (*SubPool) DebugPrint 

func (p *SubPool) DebugPrint(prefix string)

func (*SubPool) EnforceInvariants 

func (p *SubPool) EnforceInvariants()

func (*SubPool) Len 

func (p *SubPool) Len() int

func (*SubPool) PopBest 

func (p *SubPool) PopBest() *metaTxn

func (*SubPool) PopWorst 

func (p *SubPool) PopWorst() *metaTxn

func (*SubPool) Remove 

func (p *SubPool) Remove(i *metaTxn, reason string, logger log.Logger)

func (*SubPool) Updated 

func (p *SubPool) Updated(i *metaTxn)

func (*SubPool) Worst 

func (p *SubPool) Worst() *metaTxn

type SubPoolMarker 

type SubPoolMarker uint8

SubPoolMarker is an ordered bitset of five bits that's used to sort transactions into sub-pools. Bits meaning: 1. Absence of nonce gaps. Set to 1 for transactions whose nonce is N, state nonce for the sender is M, and there are transactions for all nonces between M and N from the same sender. Set to 0 is the transaction's nonce is divided from the state nonce by one or more nonce gaps. 2. Sufficient balance for gas. Set to 1 if the balance of sender's account in the state is B, nonce of the sender in the state is M, nonce of the transaction is N, and the sum of feeCap x gasLimit + transferred_value of all transactions from this sender with nonces N+1 ... M is no more than B. Set to 0 otherwise. In other words, this bit is set if there is currently a guarantee that the transaction and all its required prior transactions will be able to pay for gas. 3. Not too much gas: Set to 1 if the transaction doesn't use too much gas 4. Dynamic fee requirement. Set to 1 if feeCap of the transaction is no less than baseFee of the currently pending block. Set to 0 otherwise. 5. Local transaction. Set to 1 if transaction is local. 

const (
	NoNonceGaps       SubPoolMarker = 0b010000
	EnoughBalance     SubPoolMarker = 0b001000
	NotTooMuchGas     SubPoolMarker = 0b000100
	EnoughFeeCapBlock SubPoolMarker = 0b000010
	IsLocal           SubPoolMarker = 0b000001

	BaseFeePoolBits = NoNonceGaps + EnoughBalance + NotTooMuchGas
)

type SubPoolType 

type SubPoolType uint8
const BaseFeeSubPool SubPoolType = 2
const PendingSubPool SubPoolType = 1
const QueuedSubPool SubPoolType = 3

func (SubPoolType) String 

func (sp SubPoolType) String() string

type TxPool 

type TxPool struct {
	// contains filtered or unexported fields
}

TxPool - holds all pool-related data structures and lock-based tiny methods most of logic implemented by pure tests-friendly functions

txpool doesn't start any goroutines - "leave concurrency to user" design txpool has no DB-TX fields - "leave db transactions management to user" design txpool has _chainDB field - but it must maximize local state cache hit-rate - and perform minimum _chainDB transactions

It preserve TxnSlot objects immutable

func Assemble 

func Assemble(
	ctx context.Context,
	cfg txpoolcfg.Config,
	chainDB kv.RwDB,
	cache kvcache.Cache,
	sentryClients []sentryproto.SentryClient,
	stateChangesClient StateChangesClient,
	builderNotifyNewTxns func(),
	logger log.Logger,
	opts ...Option,
) (*TxPool, txpoolproto.TxpoolServer, error)

func New 

func New(
	ctx context.Context,
	newTxns chan Announcements,
	poolDB kv.RwDB,
	chainDB kv.RoDB,
	cfg txpoolcfg.Config,
	cache kvcache.Cache,
	chainID uint256.Int,
	shanghaiTime *big.Int,
	agraBlock *big.Int,
	cancunTime *big.Int,
	pragueTime *big.Int,
	blobSchedule *chain.BlobSchedule,
	sentryClients []sentryproto.SentryClient,
	stateChangesClient StateChangesClient,
	builderNotifyNewTxns func(),
	newSlotsStreams *NewSlotsStreams,
	logger log.Logger,
	opts ...Option,
) (*TxPool, error)

func (*TxPool) AddLocalTxns 

func (p *TxPool) AddLocalTxns(ctx context.Context, newTxns TxnSlots) ([]txpoolcfg.DiscardReason, error)

func (*TxPool) AddNewGoodPeer 

func (p *TxPool) AddNewGoodPeer(peerID PeerID)

func (*TxPool) AddRemoteTxns 

func (p *TxPool) AddRemoteTxns(_ context.Context, newTxns TxnSlots)

func (*TxPool) AppendAllAnnouncements 

func (p *TxPool) AppendAllAnnouncements(types []byte, sizes []uint32, hashes []byte) ([]byte, []uint32, []byte)

func (*TxPool) AppendLocalAnnouncements 

func (p *TxPool) AppendLocalAnnouncements(types []byte, sizes []uint32, hashes []byte) ([]byte, []uint32, []byte)

func (*TxPool) AppendRemoteAnnouncements 

func (p *TxPool) AppendRemoteAnnouncements(types []byte, sizes []uint32, hashes []byte) ([]byte, []uint32, []byte)

func (*TxPool) CountContent 

func (p *TxPool) CountContent() (int, int, int)

func (*TxPool) FilterKnownIdHashes 

func (p *TxPool) FilterKnownIdHashes(tx kv.Tx, hashes Hashes) (unknownHashes Hashes, err error)

func (*TxPool) GetMaxBlobsPerBlock 

func (p *TxPool) GetMaxBlobsPerBlock() uint64

func (*TxPool) GetRlp 

func (p *TxPool) GetRlp(tx kv.Tx, hash []byte) ([]byte, error)

func (*TxPool) IdHashKnown 

func (p *TxPool) IdHashKnown(tx kv.Tx, hash []byte) (bool, error)

func (*TxPool) IsLocal 

func (p *TxPool) IsLocal(idHash []byte) bool

func (*TxPool) NonceFromAddress 

func (p *TxPool) NonceFromAddress(addr [20]byte) (nonce uint64, inPool bool)

func (*TxPool) OnNewBlock 

func (p *TxPool) OnNewBlock(ctx context.Context, stateChanges *remote.StateChangeBatch, unwindTxns, unwindBlobTxns, minedTxns TxnSlots) error

func (*TxPool) PeekBest 

func (p *TxPool) PeekBest(ctx context.Context, n int, txns *TxnsRlp, onTopOf, availableGas, availableBlobGas uint64) (bool, error)

func (*TxPool) ProvideTxns 

func (p *TxPool) ProvideTxns(ctx context.Context, opts ...txnprovider.ProvideOption) ([]types.Transaction, error)

func (*TxPool) Run 

func (p *TxPool) Run(ctx context.Context) error

Run - does: send pending byHash to p2p:

  • new byHash
  • all pooled byHash to recently connected peers
  • all local pooled byHash to random peers periodically

promote/demote transactions reorgs

func (*TxPool) Started 

func (p *TxPool) Started() bool

func (*TxPool) ValidateSerializedTxn 

func (p *TxPool) ValidateSerializedTxn(serializedTxn []byte) error

Check that the serialized txn should not exceed a certain max size

func (*TxPool) YieldBest 

func (p *TxPool) YieldBest(ctx context.Context, n int, txns *TxnsRlp, onTopOf, availableGas, availableBlobGas uint64, toSkip mapset.Set[[32]byte]) (bool, int, error)

type TxnParseConfig 

type TxnParseConfig struct {
	ChainID uint256.Int
}

type TxnParseContext 

type TxnParseContext struct {
	Signature
	Keccak2 hash.Hash
	Keccak1 hash.Hash

	Sig     [65]byte
	Sighash [32]byte
	// contains filtered or unexported fields
}

TxnParseContext is object that is required to parse transactions and turn transaction payload into TxnSlot objects usage of TxContext helps avoid extra memory allocations

func NewTxnParseContext 

func NewTxnParseContext(chainID uint256.Int) *TxnParseContext

func (*TxnParseContext) ChainIDRequired 

func (ctx *TxnParseContext) ChainIDRequired() *TxnParseContext

Set ChainID-Required flag in the Parse context and return it

func (*TxnParseContext) ParseTransaction 

func (ctx *TxnParseContext) ParseTransaction(payload []byte, pos int, slot *TxnSlot, sender []byte, hasEnvelope, wrappedWithBlobs bool, validateHash func([]byte) error) (p int, err error)

ParseTransaction extracts all the information from the transactions's payload (RLP) necessary to build TxnSlot. It also performs syntactic validation of the transactions. wrappedWithBlobs means that for blob (type 3) transactions the full version with blobs/commitments/proofs is expected (see https://eips.ethereum.org/EIPS/eip-4844#networking).

func (*TxnParseContext) ValidateRLP 

func (ctx *TxnParseContext) ValidateRLP(f func(txnRlp []byte) error)

Set the RLP validate function

func (*TxnParseContext) WithAllowPreEip2s 

func (ctx *TxnParseContext) WithAllowPreEip2s(v bool)

Set the AllowPreEIP2s flag

func (*TxnParseContext) WithSender 

func (ctx *TxnParseContext) WithSender(v bool)

Set the with sender flag

type TxnSlot 

type TxnSlot struct {
	Rlp                 []byte      // Is set to nil after flushing to db, frees memory, later we look for it in the db, if needed
	Value               uint256.Int // Value transferred by the transaction
	Tip                 uint256.Int // Maximum tip that transaction is giving to miner/block proposer
	FeeCap              uint256.Int // Maximum fee that transaction burns and gives to the miner/block proposer
	SenderID            uint64      // SenderID - require external mapping to it's address
	Nonce               uint64      // Nonce of the transaction
	DataLen             int         // Length of transaction's data (for calculation of intrinsic gas)
	DataNonZeroLen      int
	AccessListAddrCount int      // Number of addresses in the access list
	AccessListStorCount int      // Number of storage keys in the access list
	Gas                 uint64   // Gas limit of the transaction
	IDHash              [32]byte // Transaction hash for the purposes of using it as a transaction Id
	Traced              bool     // Whether transaction needs to be traced throughout transaction pool code and generate debug printing
	Creation            bool     // Set to true if "To" field of the transaction is not set
	Type                byte     // Transaction type
	Size                uint32   // Size of the payload (without the RLP string envelope for typed transactions)

	// EIP-4844: Shard Blob Transactions
	BlobFeeCap  uint256.Int // max_fee_per_blob_gas
	BlobHashes  []common.Hash
	Blobs       [][]byte
	Commitments []gokzg4844.KZGCommitment
	Proofs      []gokzg4844.KZGProof

	// EIP-7702: set code tx
	Authorizations []Signature
	AuthRaw        [][]byte // rlp encoded chainID+address+nonce, used to recover authorization address in txpool
}

TxnSlot contains information extracted from an Ethereum transaction, which is enough to manage it inside the transaction. Also, it contains some auxiliary information, like ephemeral fields, and indices within priority queues

func (*TxnSlot) PrintDebug 

func (tx *TxnSlot) PrintDebug(prefix string)

nolint

type TxnSlots 

type TxnSlots struct {
	Txns    []*TxnSlot
	Senders Addresses
	IsLocal []bool
}

func (*TxnSlots) Append 

func (s *TxnSlots) Append(slot *TxnSlot, sender []byte, isLocal bool)

func (*TxnSlots) Resize 

func (s *TxnSlots) Resize(targetSize uint)

Resize internal arrays to len=targetSize, shrinks if need. It rely on `append` algorithm to realloc

func (*TxnSlots) Valid 

func (s *TxnSlots) Valid() error

type TxnsRlp 

type TxnsRlp struct {
	Txns    [][]byte
	Senders Addresses
	IsLocal []bool
}

func (*TxnsRlp) Resize 

func (r *TxnsRlp) Resize(targetSize uint)

Resize internal arrays to len=targetSize, shrinks if need. It rely on `append` algorithm to realloc

type WorstQueue 

type WorstQueue struct {
	// contains filtered or unexported fields
}

func (*WorstQueue) Len 

func (p *WorstQueue) Len() int

func (*WorstQueue) Less 

func (p *WorstQueue) Less(i, j int) bool

func (*WorstQueue) Pop 

func (p *WorstQueue) Pop() interface{}

func (*WorstQueue) Push 

func (p *WorstQueue) Push(x interface{})

func (*WorstQueue) Swap 

func (p *WorstQueue) Swap(i, j int)

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.