lbcd/mempool.go
Dave Collins 7c174620f7 indexers: Implement optional tx/address indexes.
This introduces a new indexing infrastructure for supporting optional
indexes using the new database and blockchain infrastructure along with
two concrete indexer implementations which provide both a
transaction-by-hash and a transaction-by-address index.

The new infrastructure is mostly separated into a package named indexers
which is housed under the blockchain package.  In order to support this,
a new interface named IndexManager has been introduced in the blockchain
package which provides methods to be notified when the chain has been
initialized and when blocks are connected and disconnected from the main
chain.  A concrete implementation of an index manager is provided by the
new indexers package.

The new indexers package also provides a new interface named Indexer
which allows the index manager to manage concrete index implementations
which conform to the interface.

The following is high level overview of the main index infrastructure
changes:

- Define a new IndexManager interface in the blockchain package and
  modify the package to make use of the interface when specified
- Create a new indexers package
  - Provides an Index interface which allows concrete indexes to plugin
    to an index manager
  - Provides a concrete IndexManager implementation
    - Handles the lifecycle of all indexes it manages
    - Tracks the index tips
    - Handles catching up disabled indexes that have been reenabled
    - Handles reorgs while the index was disabled
    - Invokes the appropriate methods for all managed indexes to allow
      them to index and deindex the blocks and transactions
  - Implement a transaction-by-hash index
    - Makes use of internal block IDs to save a significant amount of
      space and indexing costs over the old transaction index format
  - Implement a transaction-by-address index
    - Makes use of a leveling scheme in order to provide a good tradeoff
      between space required and indexing costs
- Supports enabling and disabling indexes at will
- Support the ability to drop indexes if they are no longer desired

The following is an overview of the btcd changes:

- Add a new index logging subsystem
- Add new options --txindex and --addrindex in order to enable the
  optional indexes
  - NOTE: The transaction index will automatically be enabled when the
    address index is enabled because it depends on it
- Add new options --droptxindex and --dropaddrindex to allow the indexes
  to be removed
  - NOTE: The address index will also be removed when the transaction
    index is dropped because it depends on it
- Update getrawtransactions RPC to make use of the transaction index
- Reimplement the searchrawtransaction RPC that makes use of the address
  index
- Update sample-btcd.conf to include sample usage for the new optional
  index flags
2016-04-11 17:16:42 -05:00

991 lines
33 KiB
Go

// Copyright (c) 2013-2016 The btcsuite developers
// Use of this source code is governed by an ISC
// license that can be found in the LICENSE file.
package main
import (
"container/list"
"crypto/rand"
"fmt"
"math"
"math/big"
"sync"
"sync/atomic"
"time"
"github.com/btcsuite/btcd/blockchain"
"github.com/btcsuite/btcd/blockchain/indexers"
"github.com/btcsuite/btcd/mining"
"github.com/btcsuite/btcd/txscript"
"github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btcutil"
)
const (
// mempoolHeight is the height used for the "block" height field of the
// contextual transaction information provided in a transaction view.
mempoolHeight = 0x7fffffff
)
// mempoolTxDesc is a descriptor containing a transaction in the mempool along
// with additional metadata.
type mempoolTxDesc struct {
mining.TxDesc
// StartingPriority is the priority of the transaction when it was added
// to the pool.
StartingPriority float64
}
// mempoolConfig is a descriptor containing the memory pool configuration.
type mempoolConfig struct {
// Policy defines the various mempool configuration options related
// to policy.
Policy mempoolPolicy
// FetchUtxoView defines the function to use to fetch unspent
// transaction output information.
FetchUtxoView func(*btcutil.Tx) (*blockchain.UtxoViewpoint, error)
// Chain defines the concurrent safe block chain instance which houses
// the current best chain.
Chain *blockchain.BlockChain
// RelayNtfnChan defines the channel to send newly accepted transactions
// to. If unset or set to nil, notifications will not be sent.
RelayNtfnChan chan *btcutil.Tx
// SigCache defines a signature cache to use.
SigCache *txscript.SigCache
// TimeSource defines the timesource to use.
TimeSource blockchain.MedianTimeSource
// AddrIndex defines the optional address index instance to use for
// indexing the unconfirmed transactions in the memory pool.
// This can be nil if the address index is not enabled.
AddrIndex *indexers.AddrIndex
}
// mempoolPolicy houses the policy (configuration parameters) which is used to
// control the mempool.
type mempoolPolicy struct {
// DisableRelayPriority defines whether to relay free or low-fee
// transactions that do not have enough priority to be relayed.
DisableRelayPriority bool
// FreeTxRelayLimit defines the given amount in thousands of bytes
// per minute that transactions with no fee are rate limited to.
FreeTxRelayLimit float64
// MaxOrphanTxs is the maximum number of orphan transactions
// that can be queued.
MaxOrphanTxs int
// MaxOrphanTxSize is the maximum size allowed for orphan transactions.
// This helps prevent memory exhaustion attacks from sending a lot of
// of big orphans.
MaxOrphanTxSize int
// MaxSigOpsPerTx is the maximum number of signature operations
// in a single transaction we will relay or mine. It is a fraction
// of the max signature operations for a block.
MaxSigOpsPerTx int
// MinRelayTxFee defines the minimum transaction fee in BTC/kB to be
// considered a non-zero fee.
MinRelayTxFee btcutil.Amount
}
// txMemPool is used as a source of transactions that need to be mined into
// blocks and relayed to other peers. It is safe for concurrent access from
// multiple peers.
type txMemPool struct {
// The following variables must only be used atomically.
lastUpdated int64 // last time pool was updated
sync.RWMutex
cfg mempoolConfig
pool map[wire.ShaHash]*mempoolTxDesc
orphans map[wire.ShaHash]*btcutil.Tx
orphansByPrev map[wire.ShaHash]map[wire.ShaHash]*btcutil.Tx
outpoints map[wire.OutPoint]*btcutil.Tx
pennyTotal float64 // exponentially decaying total for penny spends.
lastPennyUnix int64 // unix time of last ``penny spend''
}
// Ensure the txMemPool type implements the mining.TxSource interface.
var _ mining.TxSource = (*txMemPool)(nil)
// removeOrphan is the internal function which implements the public
// RemoveOrphan. See the comment for RemoveOrphan for more details.
//
// This function MUST be called with the mempool lock held (for writes).
func (mp *txMemPool) removeOrphan(txHash *wire.ShaHash) {
// Nothing to do if passed tx is not an orphan.
tx, exists := mp.orphans[*txHash]
if !exists {
return
}
// Remove the reference from the previous orphan index.
for _, txIn := range tx.MsgTx().TxIn {
originTxHash := txIn.PreviousOutPoint.Hash
if orphans, exists := mp.orphansByPrev[originTxHash]; exists {
delete(orphans, *tx.Sha())
// Remove the map entry altogether if there are no
// longer any orphans which depend on it.
if len(orphans) == 0 {
delete(mp.orphansByPrev, originTxHash)
}
}
}
// Remove the transaction from the orphan pool.
delete(mp.orphans, *txHash)
}
// RemoveOrphan removes the passed orphan transaction from the orphan pool and
// previous orphan index.
//
// This function is safe for concurrent access.
func (mp *txMemPool) RemoveOrphan(txHash *wire.ShaHash) {
mp.Lock()
mp.removeOrphan(txHash)
mp.Unlock()
}
// limitNumOrphans limits the number of orphan transactions by evicting a random
// orphan if adding a new one would cause it to overflow the max allowed.
//
// This function MUST be called with the mempool lock held (for writes).
func (mp *txMemPool) limitNumOrphans() error {
if len(mp.orphans)+1 > mp.cfg.Policy.MaxOrphanTxs &&
mp.cfg.Policy.MaxOrphanTxs > 0 {
// Generate a cryptographically random hash.
randHashBytes := make([]byte, wire.HashSize)
_, err := rand.Read(randHashBytes)
if err != nil {
return err
}
randHashNum := new(big.Int).SetBytes(randHashBytes)
// Try to find the first entry that is greater than the random
// hash. Use the first entry (which is already pseudorandom due
// to Go's range statement over maps) as a fallback if none of
// the hashes in the orphan pool are larger than the random
// hash.
var foundHash *wire.ShaHash
for txHash := range mp.orphans {
if foundHash == nil {
foundHash = &txHash
}
txHashNum := blockchain.ShaHashToBig(&txHash)
if txHashNum.Cmp(randHashNum) > 0 {
foundHash = &txHash
break
}
}
mp.removeOrphan(foundHash)
}
return nil
}
// addOrphan adds an orphan transaction to the orphan pool.
//
// This function MUST be called with the mempool lock held (for writes).
func (mp *txMemPool) addOrphan(tx *btcutil.Tx) {
// Limit the number orphan transactions to prevent memory exhaustion. A
// random orphan is evicted to make room if needed.
mp.limitNumOrphans()
mp.orphans[*tx.Sha()] = tx
for _, txIn := range tx.MsgTx().TxIn {
originTxHash := txIn.PreviousOutPoint.Hash
if _, exists := mp.orphansByPrev[originTxHash]; !exists {
mp.orphansByPrev[originTxHash] =
make(map[wire.ShaHash]*btcutil.Tx)
}
mp.orphansByPrev[originTxHash][*tx.Sha()] = tx
}
txmpLog.Debugf("Stored orphan transaction %v (total: %d)", tx.Sha(),
len(mp.orphans))
}
// maybeAddOrphan potentially adds an orphan to the orphan pool.
//
// This function MUST be called with the mempool lock held (for writes).
func (mp *txMemPool) maybeAddOrphan(tx *btcutil.Tx) error {
// Ignore orphan transactions that are too large. This helps avoid
// a memory exhaustion attack based on sending a lot of really large
// orphans. In the case there is a valid transaction larger than this,
// it will ultimtely be rebroadcast after the parent transactions
// have been mined or otherwise received.
//
// Note that the number of orphan transactions in the orphan pool is
// also limited, so this equates to a maximum memory used of
// mp.cfg.Policy.MaxOrphanTxSize * mp.cfg.Policy.MaxOrphanTxs (which is ~5MB
// using the default values at the time this comment was written).
serializedLen := tx.MsgTx().SerializeSize()
if serializedLen > mp.cfg.Policy.MaxOrphanTxSize {
str := fmt.Sprintf("orphan transaction size of %d bytes is "+
"larger than max allowed size of %d bytes",
serializedLen, mp.cfg.Policy.MaxOrphanTxSize)
return txRuleError(wire.RejectNonstandard, str)
}
// Add the orphan if the none of the above disqualified it.
mp.addOrphan(tx)
return nil
}
// isTransactionInPool returns whether or not the passed transaction already
// exists in the main pool.
//
// This function MUST be called with the mempool lock held (for reads).
func (mp *txMemPool) isTransactionInPool(hash *wire.ShaHash) bool {
if _, exists := mp.pool[*hash]; exists {
return true
}
return false
}
// IsTransactionInPool returns whether or not the passed transaction already
// exists in the main pool.
//
// This function is safe for concurrent access.
func (mp *txMemPool) IsTransactionInPool(hash *wire.ShaHash) bool {
// Protect concurrent access.
mp.RLock()
defer mp.RUnlock()
return mp.isTransactionInPool(hash)
}
// isOrphanInPool returns whether or not the passed transaction already exists
// in the orphan pool.
//
// This function MUST be called with the mempool lock held (for reads).
func (mp *txMemPool) isOrphanInPool(hash *wire.ShaHash) bool {
if _, exists := mp.orphans[*hash]; exists {
return true
}
return false
}
// IsOrphanInPool returns whether or not the passed transaction already exists
// in the orphan pool.
//
// This function is safe for concurrent access.
func (mp *txMemPool) IsOrphanInPool(hash *wire.ShaHash) bool {
// Protect concurrent access.
mp.RLock()
defer mp.RUnlock()
return mp.isOrphanInPool(hash)
}
// haveTransaction returns whether or not the passed transaction already exists
// in the main pool or in the orphan pool.
//
// This function MUST be called with the mempool lock held (for reads).
func (mp *txMemPool) haveTransaction(hash *wire.ShaHash) bool {
return mp.isTransactionInPool(hash) || mp.isOrphanInPool(hash)
}
// HaveTransaction returns whether or not the passed transaction already exists
// in the main pool or in the orphan pool.
//
// This function is safe for concurrent access.
func (mp *txMemPool) HaveTransaction(hash *wire.ShaHash) bool {
// Protect concurrent access.
mp.RLock()
defer mp.RUnlock()
return mp.haveTransaction(hash)
}
// removeTransaction is the internal function which implements the public
// RemoveTransaction. See the comment for RemoveTransaction for more details.
//
// This function MUST be called with the mempool lock held (for writes).
func (mp *txMemPool) removeTransaction(tx *btcutil.Tx, removeRedeemers bool) {
txHash := tx.Sha()
if removeRedeemers {
// Remove any transactions which rely on this one.
for i := uint32(0); i < uint32(len(tx.MsgTx().TxOut)); i++ {
outpoint := wire.NewOutPoint(txHash, i)
if txRedeemer, exists := mp.outpoints[*outpoint]; exists {
mp.removeTransaction(txRedeemer, true)
}
}
}
// Remove the transaction if needed.
if txDesc, exists := mp.pool[*txHash]; exists {
// Remove unconfirmed address index entries associated with the
// transaction if enabled.
if mp.cfg.AddrIndex != nil {
mp.cfg.AddrIndex.RemoveUnconfirmedTx(txHash)
}
// Mark the referenced outpoints as unspent by the pool.
for _, txIn := range txDesc.Tx.MsgTx().TxIn {
delete(mp.outpoints, txIn.PreviousOutPoint)
}
delete(mp.pool, *txHash)
atomic.StoreInt64(&mp.lastUpdated, time.Now().Unix())
}
}
// RemoveTransaction removes the passed transaction from the mempool. When the
// removeRedeemers flag is set, any transactions that redeem outputs from the
// removed transaction will also be removed recursively from the mempool, as
// they would otherwise become orphans.
//
// This function is safe for concurrent access.
func (mp *txMemPool) RemoveTransaction(tx *btcutil.Tx, removeRedeemers bool) {
// Protect concurrent access.
mp.Lock()
defer mp.Unlock()
mp.removeTransaction(tx, removeRedeemers)
}
// RemoveDoubleSpends removes all transactions which spend outputs spent by the
// passed transaction from the memory pool. Removing those transactions then
// leads to removing all transactions which rely on them, recursively. This is
// necessary when a block is connected to the main chain because the block may
// contain transactions which were previously unknown to the memory pool.
//
// This function is safe for concurrent access.
func (mp *txMemPool) RemoveDoubleSpends(tx *btcutil.Tx) {
// Protect concurrent access.
mp.Lock()
defer mp.Unlock()
for _, txIn := range tx.MsgTx().TxIn {
if txRedeemer, ok := mp.outpoints[txIn.PreviousOutPoint]; ok {
if !txRedeemer.Sha().IsEqual(tx.Sha()) {
mp.removeTransaction(txRedeemer, true)
}
}
}
}
// addTransaction adds the passed transaction to the memory pool. It should
// not be called directly as it doesn't perform any validation. This is a
// helper for maybeAcceptTransaction.
//
// This function MUST be called with the mempool lock held (for writes).
func (mp *txMemPool) addTransaction(utxoView *blockchain.UtxoViewpoint, tx *btcutil.Tx, height int32, fee int64) {
// Add the transaction to the pool and mark the referenced outpoints
// as spent by the pool.
mp.pool[*tx.Sha()] = &mempoolTxDesc{
TxDesc: mining.TxDesc{
Tx: tx,
Added: time.Now(),
Height: height,
Fee: fee,
},
StartingPriority: calcPriority(tx.MsgTx(), utxoView, height),
}
for _, txIn := range tx.MsgTx().TxIn {
mp.outpoints[txIn.PreviousOutPoint] = tx
}
atomic.StoreInt64(&mp.lastUpdated, time.Now().Unix())
// Add unconfirmed address index entries associated with the transaction
// if enabled.
if mp.cfg.AddrIndex != nil {
mp.cfg.AddrIndex.AddUnconfirmedTx(tx, utxoView)
}
}
// checkPoolDoubleSpend checks whether or not the passed transaction is
// attempting to spend coins already spent by other transactions in the pool.
// Note it does not check for double spends against transactions already in the
// main chain.
//
// This function MUST be called with the mempool lock held (for reads).
func (mp *txMemPool) checkPoolDoubleSpend(tx *btcutil.Tx) error {
for _, txIn := range tx.MsgTx().TxIn {
if txR, exists := mp.outpoints[txIn.PreviousOutPoint]; exists {
str := fmt.Sprintf("output %v already spent by "+
"transaction %v in the memory pool",
txIn.PreviousOutPoint, txR.Sha())
return txRuleError(wire.RejectDuplicate, str)
}
}
return nil
}
// fetchInputUtxos loads utxo details about the input transactions referenced by
// the passed transaction. First, it loads the details form the viewpoint of
// the main chain, then it adjusts them based upon the contents of the
// transaction pool.
//
// This function MUST be called with the mempool lock held (for reads).
func (mp *txMemPool) fetchInputUtxos(tx *btcutil.Tx) (*blockchain.UtxoViewpoint, error) {
utxoView, err := mp.cfg.FetchUtxoView(tx)
if err != nil {
return nil, err
}
// Attempt to populate any missing inputs from the transaction pool.
for originHash, entry := range utxoView.Entries() {
if entry != nil && !entry.IsFullySpent() {
continue
}
if poolTxDesc, exists := mp.pool[originHash]; exists {
utxoView.AddTxOuts(poolTxDesc.Tx, mempoolHeight)
}
}
return utxoView, nil
}
// FetchTransaction returns the requested transaction from the transaction pool.
// This only fetches from the main transaction pool and does not include
// orphans.
//
// This function is safe for concurrent access.
func (mp *txMemPool) FetchTransaction(txHash *wire.ShaHash) (*btcutil.Tx, error) {
// Protect concurrent access.
mp.RLock()
defer mp.RUnlock()
if txDesc, exists := mp.pool[*txHash]; exists {
return txDesc.Tx, nil
}
return nil, fmt.Errorf("transaction is not in the pool")
}
// maybeAcceptTransaction is the internal function which implements the public
// MaybeAcceptTransaction. See the comment for MaybeAcceptTransaction for
// more details.
//
// This function MUST be called with the mempool lock held (for writes).
func (mp *txMemPool) maybeAcceptTransaction(tx *btcutil.Tx, isNew, rateLimit bool) ([]*wire.ShaHash, error) {
txHash := tx.Sha()
// Don't accept the transaction if it already exists in the pool. This
// applies to orphan transactions as well. This check is intended to
// be a quick check to weed out duplicates.
if mp.haveTransaction(txHash) {
str := fmt.Sprintf("already have transaction %v", txHash)
return nil, txRuleError(wire.RejectDuplicate, str)
}
// Perform preliminary sanity checks on the transaction. This makes
// use of btcchain which contains the invariant rules for what
// transactions are allowed into blocks.
err := blockchain.CheckTransactionSanity(tx)
if err != nil {
if cerr, ok := err.(blockchain.RuleError); ok {
return nil, chainRuleError(cerr)
}
return nil, err
}
// A standalone transaction must not be a coinbase transaction.
if blockchain.IsCoinBase(tx) {
str := fmt.Sprintf("transaction %v is an individual coinbase",
txHash)
return nil, txRuleError(wire.RejectInvalid, str)
}
// Don't accept transactions with a lock time after the maximum int32
// value for now. This is an artifact of older bitcoind clients which
// treated this field as an int32 and would treat anything larger
// incorrectly (as negative).
if tx.MsgTx().LockTime > math.MaxInt32 {
str := fmt.Sprintf("transaction %v has a lock time after "+
"2038 which is not accepted yet", txHash)
return nil, txRuleError(wire.RejectNonstandard, str)
}
// Get the current height of the main chain. A standalone transaction
// will be mined into the next block at best, so its height is at least
// one more than the current height.
best := mp.cfg.Chain.BestSnapshot()
nextBlockHeight := best.Height + 1
// Don't allow non-standard transactions if the network parameters
// forbid their relaying.
if !activeNetParams.RelayNonStdTxs {
err := checkTransactionStandard(tx, nextBlockHeight,
mp.cfg.TimeSource, mp.cfg.Policy.MinRelayTxFee)
if err != nil {
// Attempt to extract a reject code from the error so
// it can be retained. When not possible, fall back to
// a non standard error.
rejectCode, found := extractRejectCode(err)
if !found {
rejectCode = wire.RejectNonstandard
}
str := fmt.Sprintf("transaction %v is not standard: %v",
txHash, err)
return nil, txRuleError(rejectCode, str)
}
}
// The transaction may not use any of the same outputs as other
// transactions already in the pool as that would ultimately result in a
// double spend. This check is intended to be quick and therefore only
// detects double spends within the transaction pool itself. The
// transaction could still be double spending coins from the main chain
// at this point. There is a more in-depth check that happens later
// after fetching the referenced transaction inputs from the main chain
// which examines the actual spend data and prevents double spends.
err = mp.checkPoolDoubleSpend(tx)
if err != nil {
return nil, err
}
// Fetch all of the unspent transaction outputs referenced by the inputs
// to this transaction. This function also attempts to fetch the
// transaction itself to be used for detecting a duplicate transaction
// without needing to do a separate lookup.
utxoView, err := mp.fetchInputUtxos(tx)
if err != nil {
if cerr, ok := err.(blockchain.RuleError); ok {
return nil, chainRuleError(cerr)
}
return nil, err
}
// Don't allow the transaction if it exists in the main chain and is not
// not already fully spent.
txEntry := utxoView.LookupEntry(txHash)
if txEntry != nil && !txEntry.IsFullySpent() {
return nil, txRuleError(wire.RejectDuplicate,
"transaction already exists")
}
delete(utxoView.Entries(), *txHash)
// Transaction is an orphan if any of the referenced input transactions
// don't exist. Adding orphans to the orphan pool is not handled by
// this function, and the caller should use maybeAddOrphan if this
// behavior is desired.
var missingParents []*wire.ShaHash
for originHash, entry := range utxoView.Entries() {
if entry == nil || entry.IsFullySpent() {
// Must make a copy of the hash here since the iterator
// is replaced and taking its address directly would
// result in all of the entries pointing to the same
// memory location and thus all be the final hash.
hashCopy := originHash
missingParents = append(missingParents, &hashCopy)
}
}
if len(missingParents) > 0 {
return missingParents, nil
}
// Perform several checks on the transaction inputs using the invariant
// rules in btcchain for what transactions are allowed into blocks.
// Also returns the fees associated with the transaction which will be
// used later.
txFee, err := blockchain.CheckTransactionInputs(tx, nextBlockHeight,
utxoView)
if err != nil {
if cerr, ok := err.(blockchain.RuleError); ok {
return nil, chainRuleError(cerr)
}
return nil, err
}
// Don't allow transactions with non-standard inputs if the network
// parameters forbid their relaying.
if !activeNetParams.RelayNonStdTxs {
err := checkInputsStandard(tx, utxoView)
if err != nil {
// Attempt to extract a reject code from the error so
// it can be retained. When not possible, fall back to
// a non standard error.
rejectCode, found := extractRejectCode(err)
if !found {
rejectCode = wire.RejectNonstandard
}
str := fmt.Sprintf("transaction %v has a non-standard "+
"input: %v", txHash, err)
return nil, txRuleError(rejectCode, str)
}
}
// NOTE: if you modify this code to accept non-standard transactions,
// you should add code here to check that the transaction does a
// reasonable number of ECDSA signature verifications.
// Don't allow transactions with an excessive number of signature
// operations which would result in making it impossible to mine. Since
// the coinbase address itself can contain signature operations, the
// maximum allowed signature operations per transaction is less than
// the maximum allowed signature operations per block.
numSigOps, err := blockchain.CountP2SHSigOps(tx, false, utxoView)
if err != nil {
if cerr, ok := err.(blockchain.RuleError); ok {
return nil, chainRuleError(cerr)
}
return nil, err
}
numSigOps += blockchain.CountSigOps(tx)
if numSigOps > mp.cfg.Policy.MaxSigOpsPerTx {
str := fmt.Sprintf("transaction %v has too many sigops: %d > %d",
txHash, numSigOps, mp.cfg.Policy.MaxSigOpsPerTx)
return nil, txRuleError(wire.RejectNonstandard, str)
}
// Don't allow transactions with fees too low to get into a mined block.
//
// Most miners allow a free transaction area in blocks they mine to go
// alongside the area used for high-priority transactions as well as
// transactions with fees. A transaction size of up to 1000 bytes is
// considered safe to go into this section. Further, the minimum fee
// calculated below on its own would encourage several small
// transactions to avoid fees rather than one single larger transaction
// which is more desirable. Therefore, as long as the size of the
// transaction does not exceeed 1000 less than the reserved space for
// high-priority transactions, don't require a fee for it.
serializedSize := int64(tx.MsgTx().SerializeSize())
minFee := calcMinRequiredTxRelayFee(serializedSize,
mp.cfg.Policy.MinRelayTxFee)
if serializedSize >= (defaultBlockPrioritySize-1000) && txFee < minFee {
str := fmt.Sprintf("transaction %v has %d fees which is under "+
"the required amount of %d", txHash, txFee,
minFee)
return nil, txRuleError(wire.RejectInsufficientFee, str)
}
// Require that free transactions have sufficient priority to be mined
// in the next block. Transactions which are being added back to the
// memory pool from blocks that have been disconnected during a reorg
// are exempted.
if isNew && !mp.cfg.Policy.DisableRelayPriority && txFee < minFee {
currentPriority := calcPriority(tx.MsgTx(), utxoView,
nextBlockHeight)
if currentPriority <= minHighPriority {
str := fmt.Sprintf("transaction %v has insufficient "+
"priority (%g <= %g)", txHash,
currentPriority, minHighPriority)
return nil, txRuleError(wire.RejectInsufficientFee, str)
}
}
// Free-to-relay transactions are rate limited here to prevent
// penny-flooding with tiny transactions as a form of attack.
if rateLimit && txFee < minFee {
nowUnix := time.Now().Unix()
// we decay passed data with an exponentially decaying ~10
// minutes window - matches bitcoind handling.
mp.pennyTotal *= math.Pow(1.0-1.0/600.0,
float64(nowUnix-mp.lastPennyUnix))
mp.lastPennyUnix = nowUnix
// Are we still over the limit?
if mp.pennyTotal >= mp.cfg.Policy.FreeTxRelayLimit*10*1000 {
str := fmt.Sprintf("transaction %v has been rejected "+
"by the rate limiter due to low fees", txHash)
return nil, txRuleError(wire.RejectInsufficientFee, str)
}
oldTotal := mp.pennyTotal
mp.pennyTotal += float64(serializedSize)
txmpLog.Tracef("rate limit: curTotal %v, nextTotal: %v, "+
"limit %v", oldTotal, mp.pennyTotal,
mp.cfg.Policy.FreeTxRelayLimit*10*1000)
}
// Verify crypto signatures for each input and reject the transaction if
// any don't verify.
err = blockchain.ValidateTransactionScripts(tx, utxoView,
txscript.StandardVerifyFlags, mp.cfg.SigCache)
if err != nil {
if cerr, ok := err.(blockchain.RuleError); ok {
return nil, chainRuleError(cerr)
}
return nil, err
}
// Add to transaction pool.
mp.addTransaction(utxoView, tx, best.Height, txFee)
txmpLog.Debugf("Accepted transaction %v (pool size: %v)", txHash,
len(mp.pool))
return nil, nil
}
// MaybeAcceptTransaction is the main workhorse for handling insertion of new
// free-standing transactions into a memory pool. It includes functionality
// such as rejecting duplicate transactions, ensuring transactions follow all
// rules, detecting orphan transactions, and insertion into the memory pool.
//
// If the transaction is an orphan (missing parent transactions), the
// transaction is NOT added to the orphan pool, but each unknown referenced
// parent is returned. Use ProcessTransaction instead if new orphans should
// be added to the orphan pool.
//
// This function is safe for concurrent access.
func (mp *txMemPool) MaybeAcceptTransaction(tx *btcutil.Tx, isNew, rateLimit bool) ([]*wire.ShaHash, error) {
// Protect concurrent access.
mp.Lock()
defer mp.Unlock()
return mp.maybeAcceptTransaction(tx, isNew, rateLimit)
}
// processOrphans is the internal function which implements the public
// ProcessOrphans. See the comment for ProcessOrphans for more details.
//
// This function MUST be called with the mempool lock held (for writes).
func (mp *txMemPool) processOrphans(hash *wire.ShaHash) {
// Start with processing at least the passed hash.
processHashes := list.New()
processHashes.PushBack(hash)
for processHashes.Len() > 0 {
// Pop the first hash to process.
firstElement := processHashes.Remove(processHashes.Front())
processHash := firstElement.(*wire.ShaHash)
// Look up all orphans that are referenced by the transaction we
// just accepted. This will typically only be one, but it could
// be multiple if the referenced transaction contains multiple
// outputs. Skip to the next item on the list of hashes to
// process if there are none.
orphans, exists := mp.orphansByPrev[*processHash]
if !exists || orphans == nil {
continue
}
for _, tx := range orphans {
// Remove the orphan from the orphan pool. Current
// behavior requires that all saved orphans with
// a newly accepted parent are removed from the orphan
// pool and potentially added to the memory pool, but
// transactions which cannot be added to memory pool
// (including due to still being orphans) are expunged
// from the orphan pool.
//
// TODO(jrick): The above described behavior sounds
// like a bug, and I think we should investigate
// potentially moving orphans to the memory pool, but
// leaving them in the orphan pool if not all parent
// transactions are known yet.
orphanHash := tx.Sha()
mp.removeOrphan(orphanHash)
// Potentially accept the transaction into the
// transaction pool.
missingParents, err := mp.maybeAcceptTransaction(tx,
true, true)
if err != nil {
// TODO: Remove orphans that depend on this
// failed transaction.
txmpLog.Debugf("Unable to move "+
"orphan transaction %v to mempool: %v",
tx.Sha(), err)
continue
}
if len(missingParents) > 0 {
// Transaction is still an orphan, so add it
// back.
mp.addOrphan(tx)
continue
}
// Notify the caller of the new tx added to mempool.
if mp.cfg.RelayNtfnChan != nil {
mp.cfg.RelayNtfnChan <- tx
}
// Add this transaction to the list of transactions to
// process so any orphans that depend on this one are
// handled too.
//
// TODO(jrick): In the case that this is still an orphan,
// we know that any other transactions in the orphan
// pool with this orphan as their parent are still
// orphans as well, and should be removed. While
// recursively calling removeOrphan and
// maybeAcceptTransaction on these transactions is not
// wrong per se, it is overkill if all we care about is
// recursively removing child transactions of this
// orphan.
processHashes.PushBack(orphanHash)
}
}
}
// ProcessOrphans determines if there are any orphans which depend on the passed
// transaction hash (it is possible that they are no longer orphans) and
// potentially accepts them to the memory pool. It repeats the process for the
// newly accepted transactions (to detect further orphans which may no longer be
// orphans) until there are no more.
//
// This function is safe for concurrent access.
func (mp *txMemPool) ProcessOrphans(hash *wire.ShaHash) {
mp.Lock()
mp.processOrphans(hash)
mp.Unlock()
}
// ProcessTransaction is the main workhorse for handling insertion of new
// free-standing transactions into the memory pool. It includes functionality
// such as rejecting duplicate transactions, ensuring transactions follow all
// rules, orphan transaction handling, and insertion into the memory pool.
//
// This function is safe for concurrent access.
func (mp *txMemPool) ProcessTransaction(tx *btcutil.Tx, allowOrphan, rateLimit bool) error {
// Protect concurrent access.
mp.Lock()
defer mp.Unlock()
txmpLog.Tracef("Processing transaction %v", tx.Sha())
// Potentially accept the transaction to the memory pool.
missingParents, err := mp.maybeAcceptTransaction(tx, true, rateLimit)
if err != nil {
return err
}
if len(missingParents) == 0 {
// Notify the caller that the tx was added to the mempool.
if mp.cfg.RelayNtfnChan != nil {
mp.cfg.RelayNtfnChan <- tx
}
// Accept any orphan transactions that depend on this
// transaction (they may no longer be orphans if all inputs
// are now available) and repeat for those accepted
// transactions until there are no more.
mp.processOrphans(tx.Sha())
} else {
// The transaction is an orphan (has inputs missing). Reject
// it if the flag to allow orphans is not set.
if !allowOrphan {
// Only use the first missing parent transaction in
// the error message.
//
// NOTE: RejectDuplicate is really not an accurate
// reject code here, but it matches the reference
// implementation and there isn't a better choice due
// to the limited number of reject codes. Missing
// inputs is assumed to mean they are already spent
// which is not really always the case.
str := fmt.Sprintf("orphan transaction %v references "+
"outputs of unknown or fully-spent "+
"transaction %v", tx.Sha(), missingParents[0])
return txRuleError(wire.RejectDuplicate, str)
}
// Potentially add the orphan transaction to the orphan pool.
err := mp.maybeAddOrphan(tx)
if err != nil {
return err
}
}
return nil
}
// Count returns the number of transactions in the main pool. It does not
// include the orphan pool.
//
// This function is safe for concurrent access.
func (mp *txMemPool) Count() int {
mp.RLock()
defer mp.RUnlock()
return len(mp.pool)
}
// TxShas returns a slice of hashes for all of the transactions in the memory
// pool.
//
// This function is safe for concurrent access.
func (mp *txMemPool) TxShas() []*wire.ShaHash {
mp.RLock()
defer mp.RUnlock()
hashes := make([]*wire.ShaHash, len(mp.pool))
i := 0
for hash := range mp.pool {
hashCopy := hash
hashes[i] = &hashCopy
i++
}
return hashes
}
// TxDescs returns a slice of descriptors for all the transactions in the pool.
// The descriptors are to be treated as read only.
//
// This function is safe for concurrent access.
func (mp *txMemPool) TxDescs() []*mempoolTxDesc {
mp.RLock()
defer mp.RUnlock()
descs := make([]*mempoolTxDesc, len(mp.pool))
i := 0
for _, desc := range mp.pool {
descs[i] = desc
i++
}
return descs
}
// MiningDescs returns a slice of mining descriptors for all the transactions
// in the pool.
//
// This is part of the mining.TxSource interface implementation and is safe for
// concurrent access as required by the interface contract.
func (mp *txMemPool) MiningDescs() []*mining.TxDesc {
mp.RLock()
defer mp.RUnlock()
descs := make([]*mining.TxDesc, len(mp.pool))
i := 0
for _, desc := range mp.pool {
descs[i] = &desc.TxDesc
i++
}
return descs
}
// LastUpdated returns the last time a transaction was added to or removed from
// the main pool. It does not include the orphan pool.
//
// This function is safe for concurrent access.
func (mp *txMemPool) LastUpdated() time.Time {
return time.Unix(atomic.LoadInt64(&mp.lastUpdated), 0)
}
// newTxMemPool returns a new memory pool for validating and storing standalone
// transactions until they are mined into a block.
func newTxMemPool(cfg *mempoolConfig) *txMemPool {
memPool := &txMemPool{
cfg: *cfg,
pool: make(map[wire.ShaHash]*mempoolTxDesc),
orphans: make(map[wire.ShaHash]*btcutil.Tx),
orphansByPrev: make(map[wire.ShaHash]map[wire.ShaHash]*btcutil.Tx),
outpoints: make(map[wire.OutPoint]*btcutil.Tx),
}
return memPool
}