diff --git a/core/txpool/localpool/blockchain.go b/core/txpool/localpool/blockchain.go new file mode 100644 index 0000000000..df5d069baa --- /dev/null +++ b/core/txpool/localpool/blockchain.go @@ -0,0 +1,24 @@ +package localpool + +import ( + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/params" +) + +// BlockChain defines the minimal set of methods needed to back a tx pool with +// a chain. Exists to allow mocking the live chain out of tests. +type BlockChain interface { + // Config retrieves the chain's fork configuration. + Config() *params.ChainConfig + + // CurrentBlock returns the current head of the chain. + CurrentBlock() *types.Header + + // GetBlock retrieves a specific block, used during pool resets. + GetBlock(hash common.Hash, number uint64) *types.Block + + // StateAt returns a state database for a given root hash (generally the head). + StateAt(root common.Hash) (*state.StateDB, error) +} diff --git a/core/txpool/localpool/localpool.go b/core/txpool/localpool/localpool.go new file mode 100644 index 0000000000..2181266da7 --- /dev/null +++ b/core/txpool/localpool/localpool.go @@ -0,0 +1,250 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Package localpool implements a transaction pool only for local transactions. +package localpool + +import ( + "errors" + "math/big" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/core/txpool" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/log" +) + +var ( + errNotLocal = errors.New("non-local transaction added to local txpool") +) + +var _ txpool.SubPool = new(LocalPool) + +type LocalPool struct { + allTxs map[common.Hash]*types.Transaction + allAccounts map[common.Address]nonceOrderedList + + signer types.Signer + currentState *state.StateDB // Current state in the blockchain head + reserver txpool.AddressReserver + chain BlockChain +} + +func NewLocalPool(chain BlockChain, signer types.Signer) (*LocalPool, error) { + head := chain.CurrentBlock() + currentState, err := chain.StateAt(head.Root) + if err != nil { + return nil, err + } + return &LocalPool{ + chain: chain, + currentState: currentState, + signer: signer, + }, nil +} + +// Filter is a selector used to decide whether a transaction whould be added +// to this particular subpool. +func (l *LocalPool) Filter(tx *types.Transaction) bool { + // Only disallow blob txs, all other txs should be allowed + return tx.Type() != types.BlobTxType +} + +// Init sets the base parameters of the subpool, allowing it to load any saved +// transactions from disk and also permitting internal maintenance routines to +// start up. +// +// These should not be passed as a constructor argument - nor should the pools +// start by themselves - in order to keep multiple subpools in lockstep with +// one another. +func (l *LocalPool) Init(gasTip *big.Int, head *types.Header, reserve txpool.AddressReserver) error { + l.reserver = reserve + // TODO load transactions.rlp + // todo run reorg procedure + l.Reset(nil, head) + return errors.New("not implemented") +} + +func (l *LocalPool) Close() error { + // todo shut down subscription + return nil +} + +// Reset retrieves the current state of the blockchain and ensures the content +// of the transaction pool is valid with regard to the chain state. +func (l *LocalPool) Reset(oldHead, newHead *types.Header) { + newState, err := l.chain.StateAt(newHead.Root) + if err != nil { + log.Error("Could not get new state in LocalPool", "head", newHead.Hash()) + } + l.currentState = newState + l.runReorg() + // todo should I reinsert local transactions that have been mined? +} + +func (l *LocalPool) runReorg() { + for addr, list := range l.allAccounts { + reorged := list.reorg(l.currentState.GetNonce(addr)) + for _, hash := range reorged { + delete(l.allTxs, hash) + } + } +} + +// SetGasTip updates the minimum price, since all transactions should +// be retained, we do nothing here. +func (l *LocalPool) SetGasTip(tip *big.Int) {} + +func (l *LocalPool) Has(hash common.Hash) bool { + _, ok := l.allTxs[hash] + return ok +} + +func (l *LocalPool) Get(hash common.Hash) *types.Transaction { + return l.allTxs[hash] +} + +func (l *LocalPool) Add(txs []*types.Transaction, local bool, sync bool) []error { + errs := make([]error, len(txs)) + if !local { + // If the transactions are not local, reject all + for i := 0; i < len(txs); i++ { + errs[i] = errNotLocal + } + return errs + } + for i, tx := range txs { + errs[i] = l.add(tx) + } + return nil +} + +func (l *LocalPool) add(tx *types.Transaction) error { + // Ignore already known transactions + if _, ok := l.allTxs[tx.Hash()]; ok { + log.Info("Ignoring already known transaction", "hash", tx.Hash()) + return nil + } + sender, err := l.signer.Sender(tx) + if err != nil { + return err + } + if _, ok := l.allAccounts[sender]; !ok { + if err := l.reserver(sender, true); err != nil { + log.Warn("Could not reserve account", "account", sender) + return err + } + l.allAccounts[sender] = make(nonceOrderedList) + } + l.allTxs[tx.Hash()] = tx + l.allAccounts[sender].enqueue(tx) + return nil +} + +// Pending retrieves all currently processable transactions, grouped by origin +// account and sorted by nonce. +func (l *LocalPool) Pending(enforceTips bool) map[common.Address][]*txpool.LazyTransaction { + pending := make(map[common.Address][]*txpool.LazyTransaction) + for addr, list := range l.allAccounts { + var txs []*txpool.LazyTransaction + list.forAllPending(l.currentState.GetNonce(addr), func(tx *types.Transaction) { + txs = append(txs, &txpool.LazyTransaction{ + Pool: l, + Hash: tx.Hash(), + Tx: tx, + Time: tx.Time(), + GasFeeCap: tx.GasFeeCap(), + GasTipCap: tx.GasTipCap(), + Gas: tx.Gas(), + BlobGas: tx.BlobGas(), + }) + }) + pending[addr] = txs + } + return nil +} + +// SubscribeTransactions subscribes to new transaction events. The subscriber +// can decide whether to receive notifications only for newly seen transactions +// or also for reorged out ones. +func (l *LocalPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool) event.Subscription { + // todo impl + return nil +} + +// Nonce returns the next nonce of an account, with all transactions executable +// by the pool already applied on top. +func (l *LocalPool) Nonce(addr common.Address) uint64 { + nextNonce := l.currentState.GetNonce(addr) + 1 + l.allAccounts[addr].forAllPending(l.currentState.GetNonce(addr), func(tx *types.Transaction) { + nextNonce++ + }) + return nextNonce +} + +// Stats retrieves the current pool stats, namely the number of pending and the +// number of queued (non-executable) transactions. +func (l *LocalPool) Stats() (int, int) { + var pending, queued = 0, 0 + for addr, list := range l.allAccounts { + var pCount int + list.forAllPending(l.currentState.GetNonce(addr), func(tx *types.Transaction) { + pCount++ + }) + pending += pCount + queued += len(list) - pCount + } + return pending, queued +} + +// Content retrieves the data content of the transaction pool, returning all the +// pending as well as queued transactions, grouped by account and sorted by nonce. +func (l *LocalPool) Content() (map[common.Address][]*types.Transaction, map[common.Address][]*types.Transaction) { + var ( + queued = make(map[common.Address][]*types.Transaction) + pending = make(map[common.Address][]*types.Transaction) + ) + for addr, list := range l.allAccounts { + p, q := list.content(l.currentState.GetNonce(addr)) + queued[addr] = q + pending[addr] = p + } + return pending, queued +} + +// ContentFrom retrieves the data content of the transaction pool, returning the +// pending as well as queued transactions of this address, grouped by nonce. +func (l *LocalPool) ContentFrom(addr common.Address) ([]*types.Transaction, []*types.Transaction) { + return l.allAccounts[addr].content(l.currentState.GetNonce(addr)) +} + +// Locals retrieves the accounts currently considered local by the pool. +func (l *LocalPool) Locals() []common.Address { + var addresses []common.Address + for addr := range l.allAccounts { + addresses = append(addresses, addr) + } + return addresses +} + +// Status returns the known status (unknown/pending/queued) of a transaction +// identified by their hashes. +func (l *LocalPool) Status(hash common.Hash) txpool.TxStatus { + return txpool.TxStatusUnknown +} diff --git a/core/txpool/localpool/localpool_test.go b/core/txpool/localpool/localpool_test.go new file mode 100644 index 0000000000..2bacb0f0f6 --- /dev/null +++ b/core/txpool/localpool/localpool_test.go @@ -0,0 +1,27 @@ +package localpool + +import "errors" + +func (l *LocalPool) verifyConsistency() error { + for _, list := range l.allAccounts { + for _, tx := range list { + if _, ok := l.allTxs[tx.Hash()]; !ok { + return errors.New("tx in nonceOrderedList but not in all txs") + } + } + } + for _, tx := range l.allTxs { + found := 0 + for _, list := range l.allAccounts { + if tx2, ok := list[tx.Nonce()]; ok { + if tx.Hash() == tx2.Hash() { + found++ + } + } + } + if found != 1 { + return errors.New("tx in all txs but not in nonceOrderedList") + } + } + return nil +} diff --git a/core/txpool/localpool/noncelist.go b/core/txpool/localpool/noncelist.go new file mode 100644 index 0000000000..2419580aae --- /dev/null +++ b/core/txpool/localpool/noncelist.go @@ -0,0 +1,65 @@ +package localpool + +import ( + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/log" +) + +type nonceOrderedList map[uint64]*types.Transaction + +func (n nonceOrderedList) enqueue(tx *types.Transaction) { + if old, ok := n[tx.Nonce()]; ok { + log.Warn("Replacing old local transaction with new", "old", old.Hash(), "new", tx.Hash()) + } + n[tx.Nonce()] = tx +} + +func (n nonceOrderedList) forAllPending(initialNonce uint64, call func(tx *types.Transaction)) { + currentNonce := initialNonce + for { + currentNonce++ + tx, ok := n[currentNonce] + if !ok { + break + } + call(tx) + } +} + +func (n nonceOrderedList) content(initialNonce uint64) ([]*types.Transaction, []*types.Transaction) { + var ( + pending []*types.Transaction + isPending = make(map[common.Hash]struct{}) + queued []*types.Transaction + ) + // collect all pending transactions + n.forAllPending(initialNonce, func(tx *types.Transaction) { + pending = append(pending, tx) + isPending[tx.Hash()] = struct{}{} + }) + // mark all transactions as queued that are not pending + for _, tx := range n { + if _, ok := isPending[tx.Hash()]; !ok { + queued = append(queued, tx) + } + } + return pending, queued +} + +func (n nonceOrderedList) reorg(currentNonce uint64) []common.Hash { + var ( + toDelete []uint64 + hashes []common.Hash + ) + for nonce := range n { + if nonce <= currentNonce { + toDelete = append(toDelete, nonce) + } + } + for _, nonce := range toDelete { + hashes = append(hashes, n[nonce].Hash()) + delete(n, nonce) + } + return hashes +}