mempool/mempool.go

@@ -100,9 +100,15 @@ type Config struct {
 	// This can be nil if the address index is not enabled.
 	AddrIndex *indexers.AddrIndex
 
-	// FeeEstimatator provides a feeEstimator. If it is not nil, the mempool
+	// AddTxToFeeEstimation defines an optional function to be called whenever a
-	// records all new transactions it observes into the feeEstimator.
+	// new transaction is added to the mempool, which can be used to track fees
-	FeeEstimator *FeeEstimator
+	// for the purposes of smart fee estimation.
+	AddTxToFeeEstimation func(txHash *chainhash.Hash, fee, size int64)
+
+	// RemoveTxFromFeeEstimation defines an optional function to be called
+	// whenever a transaction is removed from the mempool in order to track fee
+	// estimation.
+	RemoveTxFromFeeEstimation func(txHash *chainhash.Hash)
 }
 
 // Policy houses the policy (configuration parameters) which is used to
@@ -491,6 +497,13 @@ func (mp *TxPool) removeTransaction(tx *btcutil.Tx, removeRedeemers bool) {
 			delete(mp.outpoints, txIn.PreviousOutPoint)
 		}
 		delete(mp.pool, *txHash)
+
+		// Inform associated fee estimator that the transaction has been removed
+		// from the mempool
+		if mp.cfg.RemoveTxFromFeeEstimation != nil {
+			mp.cfg.RemoveTxFromFeeEstimation(txHash)
+		}
+
 		atomic.StoreInt64(&mp.lastUpdated, time.Now().Unix())
 	}
 }
@@ -559,9 +572,11 @@ func (mp *TxPool) addTransaction(utxoView *blockchain.UtxoViewpoint, tx *btcutil
 		mp.cfg.AddrIndex.AddUnconfirmedTx(tx, utxoView)
 	}
 
-	// Record this tx for fee estimation if enabled.
+	// Inform the associated fee estimator that a new transaction has been added
-	if mp.cfg.FeeEstimator != nil {
+	// to the mempool.
-		mp.cfg.FeeEstimator.ObserveTransaction(txD)
+	size := GetTxVirtualSize(txD.Tx)
+	if mp.cfg.AddTxToFeeEstimation != nil {
+		mp.cfg.AddTxToFeeEstimation(txD.Tx.Hash(), txD.Fee, size)
 	}
 
 	return txD

netsync/interface.go

@@ -8,6 +8,7 @@ import (
 	"github.com/lbryio/lbcd/blockchain"
 	"github.com/lbryio/lbcd/chaincfg"
 	"github.com/lbryio/lbcd/chaincfg/chainhash"
+	"github.com/lbryio/lbcd/fees"
 	"github.com/lbryio/lbcd/mempool"
 	"github.com/lbryio/lbcd/peer"
 	"github.com/lbryio/lbcd/wire"
@@ -37,5 +38,5 @@ type Config struct {
 	DisableCheckpoints bool
 	MaxPeers           int
 
-	FeeEstimator *mempool.FeeEstimator
+	FeeEstimator *fees.Estimator
 }

netsync/manager.go

@@ -16,6 +16,7 @@ import (
 	"github.com/lbryio/lbcd/chaincfg"
 	"github.com/lbryio/lbcd/chaincfg/chainhash"
 	"github.com/lbryio/lbcd/database"
+	"github.com/lbryio/lbcd/fees"
 	"github.com/lbryio/lbcd/mempool"
 	peerpkg "github.com/lbryio/lbcd/peer"
 	"github.com/lbryio/lbcd/wire"
@@ -205,7 +206,7 @@ type SyncManager struct {
 	nextCheckpoint   *chaincfg.Checkpoint
 
 	// An optional fee estimator.
-	feeEstimator *mempool.FeeEstimator
+	feeEstimator *fees.Estimator
 }
 
 // resetHeaderState sets the headers-first mode state to values appropriate for
@@ -1414,6 +1415,13 @@ func (sm *SyncManager) handleBlockchainNotification(notification *blockchain.Not
 		iv := wire.NewInvVect(wire.InvTypeBlock, block.Hash())
 		sm.peerNotifier.RelayInventory(iv, block.MsgBlock().Header)
 
+		if !sm.feeEstimator.IsEnabled() {
+			// fee estimation can only start after we have performed an initial
+			// sync, otherwise we'll start adding mempool transactions at the
+			// wrong height.
+			sm.feeEstimator.Enable(block.Height())
+		}
+
 	// A block has been connected to the main block chain.
 	case blockchain.NTBlockConnected:
 		block, ok := notification.Data.(*btcutil.Block)
@@ -1422,6 +1430,12 @@ func (sm *SyncManager) handleBlockchainNotification(notification *blockchain.Not
 			break
 		}
 
+		// Account for transactions mined in the newly connected block for fee
+		// estimation. This must be done before attempting to remove
+		// transactions from the mempool because the mempool will alert the
+		// estimator of the txs that are leaving
+		sm.feeEstimator.ProcessBlock(block)
+
 		// Remove all of the transactions (except the coinbase) in the
 		// connected block from the transaction pool.  Secondly, remove any
 		// transactions which are now double spends as a result of these
@@ -1438,20 +1452,6 @@ func (sm *SyncManager) handleBlockchainNotification(notification *blockchain.Not
 			sm.peerNotifier.AnnounceNewTransactions(acceptedTxs)
 		}
 
-		// Register block with the fee estimator, if it exists.
-		if sm.feeEstimator != nil {
-			err := sm.feeEstimator.RegisterBlock(block)
-
-			// If an error is somehow generated then the fee estimator
-			// has entered an invalid state. Since it doesn't know how
-			// to recover, create a new one.
-			if err != nil {
-				sm.feeEstimator = mempool.NewFeeEstimator(
-					mempool.DefaultEstimateFeeMaxRollback,
-					mempool.DefaultEstimateFeeMinRegisteredBlocks)
-			}
-		}
-
 	// A block has been disconnected from the main block chain.
 	case blockchain.NTBlockDisconnected:
 		block, ok := notification.Data.(*btcutil.Block)
@@ -1473,10 +1473,6 @@ func (sm *SyncManager) handleBlockchainNotification(notification *blockchain.Not
 			}
 		}
 
-		// Rollback previous block recorded by the fee estimator.
-		if sm.feeEstimator != nil {
-			sm.feeEstimator.Rollback(block.Hash())
-		}
 	}
 }
 

rpcserver.go

@@ -36,6 +36,7 @@ import (
 	"github.com/lbryio/lbcd/chaincfg"
 	"github.com/lbryio/lbcd/chaincfg/chainhash"
 	"github.com/lbryio/lbcd/database"
+	"github.com/lbryio/lbcd/fees"
 	"github.com/lbryio/lbcd/mempool"
 	"github.com/lbryio/lbcd/mining"
 	"github.com/lbryio/lbcd/mining/cpuminer"
@@ -893,7 +894,7 @@ func handleEstimateFee(s *rpcServer, cmd interface{}, closeChan <-chan struct{})
 		}
 	}
 
-	feeRate, err := s.cfg.FeeEstimator.EstimateFee(uint32(c.NumBlocks))
+	feeRate, err := s.cfg.FeeEstimator.EstimateFee(int32(c.NumBlocks))
 
 	if err != nil {
 		return nil, &btcjson.RPCError{
@@ -906,12 +907,36 @@ func handleEstimateFee(s *rpcServer, cmd interface{}, closeChan <-chan struct{})
 	return float64(feeRate), nil
 }
 
+// handleEstimateSmartFee implements the estimatesmartfee command.
+//
+// The default estimation mode when unset is assumed as "conservative". As of
+// 2018-12, the only supported mode is "conservative".
 func handleEstimateSmartFee(s *rpcServer, cmd interface{}, closeChan <-chan struct{}) (interface{}, error) {
 	c := cmd.(*btcjson.EstimateSmartFeeCmd)
 
-	rpcsLog.Debugf("EstimateSmartFee is not implemented; falling back to EstimateFee. Requested mode: %s", c.EstimateMode)
+	mode := btcjson.EstimateModeConservative
+	if c.EstimateMode != nil {
+		mode = *c.EstimateMode
+	}
 
-	return handleEstimateFee(s, &btcjson.EstimateFeeCmd{NumBlocks: c.ConfTarget}, closeChan)
+	if mode != btcjson.EstimateModeConservative {
+		return nil, &btcjson.RPCError{
+			Code: btcjson.ErrRPCInvalidParameter,
+			Message: "Only the default and conservative modes " +
+				"are supported for smart fee estimation at the moment",
+		}
+	}
+
+	fee, err := s.cfg.FeeEstimator.EstimateFee(int32(c.ConfTarget))
+	if err != nil {
+		return nil, internalRPCError(err.Error(), "Could not estimate fee")
+	}
+
+	feeRate := float64(fee) / btcutil.SatoshiPerBitcoin
+	return &btcjson.EstimateSmartFeeResult{
+		FeeRate: &feeRate,
+		Blocks:  c.ConfTarget,
+	}, nil
 }
 
 func handleGenerate(s *rpcServer, cmd interface{}, closeChan <-chan struct{}) (interface{}, error) {
@@ -4012,6 +4037,7 @@ type rpcServer struct {
 	gbtWorkState           *gbtWorkState
 	helpCacher             *helpCacher
 	requestProcessShutdown chan struct{}
+	feeEstimator           *fees.Estimator
 	quit                   chan int
 }
 
@@ -4845,7 +4871,7 @@ type rpcserverConfig struct {
 
 	// The fee estimator keeps track of how long transactions are left in
 	// the mempool before they are mined into blocks.
-	FeeEstimator *mempool.FeeEstimator
+	FeeEstimator *fees.Estimator
 
 	// Services represents the services supported by this node.
 	Services wire.ServiceFlag
@@ -4859,6 +4885,7 @@ func newRPCServer(config *rpcserverConfig) (*rpcServer, error) {
 		gbtWorkState:           newGbtWorkState(config.TimeSource),
 		helpCacher:             newHelpCacher(),
 		requestProcessShutdown: make(chan struct{}),
+		feeEstimator:           config.FeeEstimator,
 		quit:                   make(chan int),
 	}
 	if cfg.RPCUser != "" && cfg.RPCPass != "" {

server.go

@@ -14,6 +14,7 @@ import (
 	"fmt"
 	"math"
 	"net"
+	"path"
 	"runtime"
 	"sort"
 	"strconv"
@@ -31,6 +32,7 @@ import (
 	claimtrieconfig "github.com/lbryio/lbcd/claimtrie/config"
 	"github.com/lbryio/lbcd/connmgr"
 	"github.com/lbryio/lbcd/database"
+	"github.com/lbryio/lbcd/fees"
 	"github.com/lbryio/lbcd/mempool"
 	"github.com/lbryio/lbcd/mining"
 	"github.com/lbryio/lbcd/mining/cpuminer"
@@ -38,6 +40,7 @@ import (
 	"github.com/lbryio/lbcd/peer"
 	"github.com/lbryio/lbcd/txscript"
 	"github.com/lbryio/lbcd/wire"
+	"github.com/lbryio/lbcutil"
 	btcutil "github.com/lbryio/lbcutil"
 	"github.com/lbryio/lbcutil/bloom"
 )
@@ -241,7 +244,7 @@ type server struct {
 
 	// The fee estimator keeps track of how long transactions are left in
 	// the mempool before they are mined into blocks.
-	feeEstimator *mempool.FeeEstimator
+	feeEstimator *fees.Estimator
 
 	// cfCheckptCaches stores a cached slice of filter headers for cfcheckpt
 	// messages for each filter type.
@@ -2417,13 +2420,7 @@ func (s *server) Stop() error {
 		s.rpcServer.Stop()
 	}
 
-	// Save fee estimator state in the database.
+	s.feeEstimator.Close()
-	s.db.Update(func(tx database.Tx) error {
-		metadata := tx.Metadata()
-		metadata.Put(mempool.EstimateFeeDatabaseKey, s.feeEstimator.Save())
-
-		return nil
-	})
 
 	// Signal the remaining goroutines to quit.
 	close(s.quit)
@@ -2755,35 +2752,25 @@ func newServer(listenAddrs, agentBlacklist, agentWhitelist []string,
 		return nil, err
 	}
 
-	// Search for a FeeEstimator state in the database. If none can be found
+	feC := fees.EstimatorConfig{
-	// or if it cannot be loaded, create a new one.
+		MinBucketFee: cfg.minRelayTxFee,
-	db.Update(func(tx database.Tx) error {
+		MaxBucketFee: lbcutil.Amount(fees.DefaultMaxBucketFeeMultiplier) * cfg.minRelayTxFee,
-		metadata := tx.Metadata()
+		MaxConfirms:  fees.DefaultMaxConfirmations,
-		feeEstimationData := metadata.Get(mempool.EstimateFeeDatabaseKey)
+		FeeRateStep:  fees.DefaultFeeRateStep,
-		if feeEstimationData != nil {
+		DatabaseFile: path.Join(cfg.DataDir, "feesdb"),
-			// delete it from the database so that we don't try to restore the
-			// same thing again somehow.
-			metadata.Delete(mempool.EstimateFeeDatabaseKey)
-
-			// If there is an error, log it and make a new fee estimator.
-			var err error
-			s.feeEstimator, err = mempool.RestoreFeeEstimator(feeEstimationData)
 
+		// 1e5 is the previous (up to 1.1.0) mempool.DefaultMinRelayTxFee that
+		// un-upgraded wallets will be using, so track this particular rate
+		// explicitly. Note that bumping this value will cause the existing fees
+		// database to become invalid and will force nodes to explicitly delete
+		// it.
+		ExtraBucketFee: 1e5,
+	}
+	fe, err := fees.NewEstimator(&feC)
 	if err != nil {
-				peerLog.Errorf("Failed to restore fee estimator %v", err)
+		return nil, err
-			}
-		}
-
-		return nil
-	})
-
-	// If no feeEstimator has been found, or if the one that has been found
-	// is behind somehow, create a new one and start over.
-	if s.feeEstimator == nil || s.feeEstimator.LastKnownHeight() != s.chain.BestSnapshot().Height {
-		s.feeEstimator = mempool.NewFeeEstimator(
-			mempool.DefaultEstimateFeeMaxRollback,
-			mempool.DefaultEstimateFeeMinRegisteredBlocks)
 	}
+	s.feeEstimator = fe
 
 	txC := mempool.Config{
 		Policy: mempool.Policy{
@@ -2808,7 +2795,8 @@ func newServer(listenAddrs, agentBlacklist, agentWhitelist []string,
 		SigCache:                  s.sigCache,
 		HashCache:                 s.hashCache,
 		AddrIndex:                 s.addrIndex,
-		FeeEstimator:       s.feeEstimator,
+		AddTxToFeeEstimation:      s.feeEstimator.AddMemPoolTransaction,
+		RemoveTxFromFeeEstimation: s.feeEstimator.RemoveMemPoolTransaction,
 	}
 	s.txMemPool = mempool.New(&txC)