chainfee: introduce filterManager and use it for fee floor

This commit introduces a filterManager that uses our outbound peers'
feefilter values to determine an acceptable minimum feerate that
ensures successful transaction propagation. Under the hood, a moving
median is used as it is more resistant to shocks than a moving average.

lnwallet/chainfee/estimator.go

@@ -43,6 +43,16 @@ const (
 	// WebAPIResponseTimeout specifies the timeout value for receiving a
 	// fee response from the api source.
 	WebAPIResponseTimeout = 10 * time.Second
+
+	// economicalFeeMode is a mode that bitcoind uses to serve
+	// non-conservative fee estimates. These fee estimates are less
+	// resistant to shocks.
+	economicalFeeMode = "ECONOMICAL"
+
+	// filterCapConfTarget is the conf target that will be used to cap our
+	// minimum feerate if we used the median of our peers' feefilter
+	// values.
+	filterCapConfTarget = uint32(1)
 )
 
 var (
@@ -150,6 +160,11 @@ type BtcdEstimator struct {
 	minFeeManager *minFeeManager
 
 	btcdConn *rpcclient.Client
+
+	// filterManager uses our peer's feefilter values to determine a
+	// suitable feerate to use that will allow successful transaction
+	// propagation.
+	filterManager *filterManager
 }
 
 // NewBtcdEstimator creates a new BtcdEstimator given a fully populated
@@ -167,9 +182,14 @@ func NewBtcdEstimator(rpcConfig rpcclient.ConnConfig,
 		return nil, err
 	}
 
+	fetchCb := func() ([]SatPerKWeight, error) {
+		return fetchBtcdFilters(chainConn)
+	}
+
 	return &BtcdEstimator{
 		fallbackFeePerKW: fallBackFeeRate,
 		btcdConn:         chainConn,
+		filterManager:    newFilterManager(fetchCb),
 	}, nil
 }
 
@@ -193,6 +213,8 @@ func (b *BtcdEstimator) Start() error {
 	}
 	b.minFeeManager = minRelayFeeManager
 
+	b.filterManager.Start()
+
 	return nil
 }
 
@@ -219,6 +241,8 @@ func (b *BtcdEstimator) fetchMinRelayFee() (SatPerKWeight, error) {
 //
 // NOTE: This method is part of the Estimator interface.
 func (b *BtcdEstimator) Stop() error {
+	b.filterManager.Stop()
+
 	b.btcdConn.Shutdown()
 
 	return nil
@@ -250,12 +274,47 @@ func (b *BtcdEstimator) EstimateFeePerKW(numBlocks uint32) (SatPerKWeight, error
 //
 // NOTE: This method is part of the Estimator interface.
 func (b *BtcdEstimator) RelayFeePerKW() SatPerKWeight {
-	return b.minFeeManager.fetchMinFee()
+	// Get a suitable minimum feerate to use. This may optionally use the
+	// median of our peers' feefilter values.
+	feeCapClosure := func() (SatPerKWeight, error) {
+		return b.fetchEstimateInner(filterCapConfTarget)
+	}
+
+	return chooseMinFee(
+		b.minFeeManager.fetchMinFee, b.filterManager.FetchMedianFilter,
+		feeCapClosure,
+	)
 }
 
 // fetchEstimate returns a fee estimate for a transaction to be confirmed in
 // confTarget blocks. The estimate is returned in sat/kw.
 func (b *BtcdEstimator) fetchEstimate(confTarget uint32) (SatPerKWeight, error) {
+	satPerKw, err := b.fetchEstimateInner(confTarget)
+	if err != nil {
+		return 0, err
+	}
+
+	// Finally, we'll enforce our fee floor by choosing the higher of the
+	// minimum relay fee and the feerate returned by the filterManager.
+	absoluteMinFee := b.RelayFeePerKW()
+
+	if satPerKw < absoluteMinFee {
+		log.Debugf("Estimated fee rate of %v sat/kw is too low, "+
+			"using fee floor of %v sat/kw instead", satPerKw,
+			absoluteMinFee)
+
+		satPerKw = absoluteMinFee
+	}
+
+	log.Debugf("Returning %v sat/kw for conf target of %v",
+		int64(satPerKw), confTarget)
+
+	return satPerKw, nil
+}
+
+func (b *BtcdEstimator) fetchEstimateInner(confTarget uint32) (SatPerKWeight,
+	error) {
+
 	// First, we'll fetch the estimate for our confirmation target.
 	btcPerKB, err := b.btcdConn.EstimateFee(int64(confTarget))
 	if err != nil {
@@ -271,20 +330,7 @@ func (b *BtcdEstimator) fetchEstimate(confTarget uint32) (SatPerKWeight, error)
 
 	// Since we use fee rates in sat/kw internally, we'll convert the
 	// estimated fee rate from its sat/kb representation to sat/kw.
-	satPerKw := SatPerKVByte(satPerKB).FeePerKWeight()
-
-	// Finally, we'll enforce our fee floor.
-	if satPerKw < b.minFeeManager.fetchMinFee() {
-		log.Debugf("Estimated fee rate of %v sat/kw is too low, "+
-			"using fee floor of %v sat/kw instead", satPerKw,
-			b.minFeeManager)
-		satPerKw = b.minFeeManager.fetchMinFee()
-	}
-
-	log.Debugf("Returning %v sat/kw for conf target of %v",
-		int64(satPerKw), confTarget)
-
-	return satPerKw, nil
+	return SatPerKVByte(satPerKB).FeePerKWeight(), nil
 }
 
 // A compile-time assertion to ensure that BtcdEstimator implements the
@@ -314,6 +360,11 @@ type BitcoindEstimator struct {
 	// TODO(ziggie): introduce an interface for the client to enhance
 	// testability of the estimator.
 	bitcoindConn *rpcclient.Client
+
+	// filterManager uses our peer's feefilter values to determine a
+	// suitable feerate to use that will allow successful transaction
+	// propagation.
+	filterManager *filterManager
 }
 
 // NewBitcoindEstimator creates a new BitcoindEstimator given a fully populated
@@ -333,10 +384,15 @@ func NewBitcoindEstimator(rpcConfig rpcclient.ConnConfig, feeMode string,
 		return nil, err
 	}
 
+	fetchCb := func() ([]SatPerKWeight, error) {
+		return fetchBitcoindFilters(chainConn)
+	}
+
 	return &BitcoindEstimator{
 		fallbackFeePerKW: fallBackFeeRate,
 		bitcoindConn:     chainConn,
 		feeMode:          feeMode,
+		filterManager:    newFilterManager(fetchCb),
 	}, nil
 }
 
@@ -357,6 +413,8 @@ func (b *BitcoindEstimator) Start() error {
 	}
 	b.minFeeManager = relayFeeManager
 
+	b.filterManager.Start()
+
 	return nil
 }
 
@@ -394,6 +452,7 @@ func (b *BitcoindEstimator) fetchMinMempoolFee() (SatPerKWeight, error) {
 //
 // NOTE: This method is part of the Estimator interface.
 func (b *BitcoindEstimator) Stop() error {
+	b.filterManager.Stop()
 	return nil
 }
 
@@ -411,7 +470,7 @@ func (b *BitcoindEstimator) EstimateFeePerKW(
 		numBlocks = maxBlockTarget
 	}
 
-	feeEstimate, err := b.fetchEstimate(numBlocks)
+	feeEstimate, err := b.fetchEstimate(numBlocks, b.feeMode)
 	switch {
 	// If the estimator doesn't have enough data, or returns an error, then
 	// to return a proper value, then we'll return the default fall back
@@ -432,12 +491,51 @@ func (b *BitcoindEstimator) EstimateFeePerKW(
 //
 // NOTE: This method is part of the Estimator interface.
 func (b *BitcoindEstimator) RelayFeePerKW() SatPerKWeight {
-	return b.minFeeManager.fetchMinFee()
+	// Get a suitable minimum feerate to use. This may optionally use the
+	// median of our peers' feefilter values.
+	feeCapClosure := func() (SatPerKWeight, error) {
+		return b.fetchEstimateInner(
+			filterCapConfTarget, economicalFeeMode,
+		)
+	}
+
+	return chooseMinFee(
+		b.minFeeManager.fetchMinFee, b.filterManager.FetchMedianFilter,
+		feeCapClosure,
+	)
 }
 
 // fetchEstimate returns a fee estimate for a transaction to be confirmed in
 // confTarget blocks. The estimate is returned in sat/kw.
-func (b *BitcoindEstimator) fetchEstimate(confTarget uint32) (SatPerKWeight, error) {
+func (b *BitcoindEstimator) fetchEstimate(confTarget uint32, feeMode string) (
+	SatPerKWeight, error) {
+
+	satPerKw, err := b.fetchEstimateInner(confTarget, feeMode)
+	if err != nil {
+		return 0, err
+	}
+
+	// Finally, we'll enforce our fee floor by choosing the higher of the
+	// minimum relay fee and the feerate returned by the filterManager.
+	absoluteMinFee := b.RelayFeePerKW()
+
+	if satPerKw < absoluteMinFee {
+		log.Debugf("Estimated fee rate of %v sat/kw is too low, "+
+			"using fee floor of %v sat/kw instead", satPerKw,
+			absoluteMinFee)
+
+		satPerKw = absoluteMinFee
+	}
+
+	log.Debugf("Returning %v sat/kw for conf target of %v",
+		int64(satPerKw), confTarget)
+
+	return satPerKw, nil
+}
+
+func (b *BitcoindEstimator) fetchEstimateInner(confTarget uint32,
+	feeMode string) (SatPerKWeight, error) {
+
 	// First, we'll send an "estimatesmartfee" command as a raw request,
 	// since it isn't supported by btcd but is available in bitcoind.
 	target, err := json.Marshal(uint64(confTarget))
@@ -446,7 +544,7 @@ func (b *BitcoindEstimator) fetchEstimate(confTarget uint32) (SatPerKWeight, err
 	}
 
 	// The mode must be either ECONOMICAL or CONSERVATIVE.
-	mode, err := json.Marshal(b.feeMode)
+	mode, err := json.Marshal(feeMode)
 	if err != nil {
 		return 0, err
 	}
@@ -483,22 +581,50 @@ func (b *BitcoindEstimator) fetchEstimate(confTarget uint32) (SatPerKWeight, err
 
 	// Since we use fee rates in sat/kw internally, we'll convert the
 	// estimated fee rate from its sat/kb representation to sat/kw.
-	satPerKw := SatPerKVByte(satPerKB).FeePerKWeight()
+	return SatPerKVByte(satPerKB).FeePerKWeight(), nil
+}
 
-	// Finally, we'll enforce our fee floor.
-	minRelayFee := b.minFeeManager.fetchMinFee()
-	if satPerKw < minRelayFee {
-		log.Debugf("Estimated fee rate of %v is too low, "+
-			"using fee floor of %v instead", satPerKw,
-			minRelayFee)
+// chooseMinFee takes the minimum relay fee and the median of our peers'
+// feefilter values and takes the higher of the two. It then compares the value
+// against a maximum fee and caps it if the value is higher than the maximum
+// fee. This function is only called if we have data for our peers' feefilter.
+// The returned value will be used as the fee floor for calls to
+// RelayFeePerKW.
+func chooseMinFee(minRelayFeeFunc func() SatPerKWeight,
+	medianFilterFunc func() (SatPerKWeight, error),
+	feeCapFunc func() (SatPerKWeight, error)) SatPerKWeight {
 
-		satPerKw = minRelayFee
+	minRelayFee := minRelayFeeFunc()
+
+	medianFilter, err := medianFilterFunc()
+	if err != nil {
+		// If we don't have feefilter data, we fallback to using our
+		// minimum relay fee.
+		return minRelayFee
 	}
 
-	log.Debugf("Returning %v for conf target of %v",
-		int64(satPerKw), confTarget)
+	feeCap, err := feeCapFunc()
+	if err != nil {
+		// If we encountered an error, don't use the medianFilter and
+		// instead fallback to using our minimum relay fee.
+		return minRelayFee
+	}
 
-	return satPerKw, nil
+	// If the median feefilter is higher than our minimum relay fee, use it
+	// instead.
+	if medianFilter > minRelayFee {
+		// Only apply the cap if the median filter was used. This is
+		// to prevent an adversary from taking up the majority of our
+		// outbound peer slots and forcing us to use a high median
+		// filter value.
+		if medianFilter > feeCap {
+			return feeCap
+		}
+
+		return medianFilter
+	}
+
+	return minRelayFee
 }
 
 // A compile-time assertion to ensure that BitcoindEstimator implements the

lnwallet/chainfee/filtermanager.go

@@ -0,0 +1,261 @@
+package chainfee
+
+import (
+	"encoding/json"
+	"fmt"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/btcsuite/btcd/btcutil"
+	"github.com/btcsuite/btcd/rpcclient"
+	"github.com/lightningnetwork/lnd/fn"
+)
+
+const (
+	// fetchFilterInterval is the interval between successive fetches of
+	// our peers' feefilters.
+	fetchFilterInterval = time.Minute * 5
+
+	// minNumFilters is the minimum number of feefilters we need during a
+	// polling interval. If we have fewer than this, we won't consider the
+	// data.
+	minNumFilters = 6
+)
+
+var (
+	// errNoData is an error that's returned if fetchMedianFilter is
+	// called and there is no data available.
+	errNoData = fmt.Errorf("no data available")
+)
+
+// filterManager is responsible for determining an acceptable minimum fee to
+// use based on our peers' feefilter values.
+type filterManager struct {
+	// median stores the median of our outbound peer's feefilter values.
+	median    fn.Option[SatPerKWeight]
+	medianMtx sync.RWMutex
+
+	fetchFunc func() ([]SatPerKWeight, error)
+
+	wg   sync.WaitGroup
+	quit chan struct{}
+}
+
+// newFilterManager constructs a filterManager with a callback that fetches the
+// set of peers' feefilters.
+func newFilterManager(cb func() ([]SatPerKWeight, error)) *filterManager {
+	f := &filterManager{
+		quit: make(chan struct{}),
+	}
+
+	f.fetchFunc = cb
+
+	return f
+}
+
+// Start starts the filterManager.
+func (f *filterManager) Start() {
+	f.wg.Add(1)
+	go f.fetchPeerFilters()
+}
+
+// Stop stops the filterManager.
+func (f *filterManager) Stop() {
+	close(f.quit)
+	f.wg.Wait()
+}
+
+// fetchPeerFilters fetches our peers' feefilter values and calculates the
+// median.
+func (f *filterManager) fetchPeerFilters() {
+	defer f.wg.Done()
+
+	filterTicker := time.NewTicker(fetchFilterInterval)
+	defer filterTicker.Stop()
+
+	for {
+		select {
+		case <-filterTicker.C:
+			filters, err := f.fetchFunc()
+			if err != nil {
+				log.Errorf("Encountered err while fetching "+
+					"fee filters: %v", err)
+				return
+			}
+
+			f.updateMedian(filters)
+
+		case <-f.quit:
+			return
+		}
+	}
+}
+
+// fetchMedianFilter fetches the median feefilter value.
+func (f *filterManager) FetchMedianFilter() (SatPerKWeight, error) {
+	f.medianMtx.RLock()
+	defer f.medianMtx.RUnlock()
+
+	// If there is no median, return errNoData so the caller knows to
+	// ignore the output and continue.
+	return f.median.UnwrapOrErr(errNoData)
+}
+
+type bitcoindPeerInfoResp struct {
+	Inbound      bool    `json:"inbound"`
+	MinFeeFilter float64 `json:"minfeefilter"`
+}
+
+func fetchBitcoindFilters(client *rpcclient.Client) ([]SatPerKWeight, error) {
+	resp, err := client.RawRequest("getpeerinfo", nil)
+	if err != nil {
+		return nil, err
+	}
+
+	var peerResps []bitcoindPeerInfoResp
+	err = json.Unmarshal(resp, &peerResps)
+	if err != nil {
+		return nil, err
+	}
+
+	// We filter for outbound peers since it is harder for an attacker to
+	// be our outbound peer and therefore harder for them to manipulate us
+	// into broadcasting high-fee or low-fee transactions.
+	var outboundPeerFilters []SatPerKWeight
+	for _, peerResp := range peerResps {
+		if peerResp.Inbound {
+			continue
+		}
+
+		// The value that bitcoind returns for the "minfeefilter" field
+		// is in fractions of a bitcoin that represents the satoshis
+		// per KvB. We need to convert this fraction to whole satoshis
+		// by multiplying with COIN. Then we need to convert the
+		// sats/KvB to sats/KW.
+		//
+		// Convert the sats/KvB from fractions of a bitcoin to whole
+		// satoshis.
+		filterKVByte := SatPerKVByte(
+			peerResp.MinFeeFilter * btcutil.SatoshiPerBitcoin,
+		)
+
+		if !isWithinBounds(filterKVByte) {
+			continue
+		}
+
+		// Convert KvB to KW and add it to outboundPeerFilters.
+		outboundPeerFilters = append(
+			outboundPeerFilters, filterKVByte.FeePerKWeight(),
+		)
+	}
+
+	// Check that we have enough data to use. We don't return an error as
+	// that would stop the querying goroutine.
+	if len(outboundPeerFilters) < minNumFilters {
+		return nil, nil
+	}
+
+	return outboundPeerFilters, nil
+}
+
+func fetchBtcdFilters(client *rpcclient.Client) ([]SatPerKWeight, error) {
+	resp, err := client.GetPeerInfo()
+	if err != nil {
+		return nil, err
+	}
+
+	var outboundPeerFilters []SatPerKWeight
+	for _, peerResp := range resp {
+		// We filter for outbound peers since it is harder for an
+		// attacker to be our outbound peer and therefore harder for
+		// them to manipulate us into broadcasting high-fee or low-fee
+		// transactions.
+		if peerResp.Inbound {
+			continue
+		}
+
+		// The feefilter is already in units of sat/KvB.
+		filter := SatPerKVByte(peerResp.FeeFilter)
+
+		if !isWithinBounds(filter) {
+			continue
+		}
+
+		outboundPeerFilters = append(
+			outboundPeerFilters, filter.FeePerKWeight(),
+		)
+	}
+
+	// Check that we have enough data to use. We don't return an error as
+	// that would stop the querying goroutine.
+	if len(outboundPeerFilters) < minNumFilters {
+		return nil, nil
+	}
+
+	return outboundPeerFilters, nil
+}
+
+// updateMedian takes a slice of feefilter values, computes the median, and
+// updates our stored median value.
+func (f *filterManager) updateMedian(feeFilters []SatPerKWeight) {
+	// If there are no elements, don't update.
+	numElements := len(feeFilters)
+	if numElements == 0 {
+		return
+	}
+
+	f.medianMtx.Lock()
+	defer f.medianMtx.Unlock()
+
+	// Log the new median.
+	median := med(feeFilters)
+	f.median = fn.Some(median)
+	log.Debugf("filterManager updated moving median to: %v",
+		median.FeePerKVByte())
+}
+
+// isWithinBounds returns false if the filter is unusable and true if it is.
+func isWithinBounds(filter SatPerKVByte) bool {
+	// Ignore values of 0 and MaxSatoshi. A value of 0 likely means that
+	// the peer hasn't sent over a feefilter and a value of MaxSatoshi
+	// means the peer is using bitcoind and is in IBD.
+	switch filter {
+	case 0:
+		return false
+
+	case btcutil.MaxSatoshi:
+		return false
+	}
+
+	return true
+}
+
+// med calculates the median of a slice.
+// NOTE: Passing in an empty slice will panic!
+func med(f []SatPerKWeight) SatPerKWeight {
+	// Copy the original slice so that sorting doesn't modify the original.
+	fCopy := make([]SatPerKWeight, len(f))
+	copy(fCopy, f)
+
+	sort.Slice(fCopy, func(i, j int) bool {
+		return fCopy[i] < fCopy[j]
+	})
+
+	var median SatPerKWeight
+
+	numElements := len(fCopy)
+	switch numElements % 2 {
+	case 0:
+		// There's an even number of elements, so we need to average.
+		middle := numElements / 2
+		upper := fCopy[middle]
+		lower := fCopy[middle-1]
+		median = (upper + lower) / 2
+
+	case 1:
+		median = fCopy[numElements/2]
+	}
+
+	return median
+}

lnwallet/chainfee/filtermanager_test.go

@@ -0,0 +1,52 @@
+package chainfee
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestFeeFilterMedian(t *testing.T) {
+	tests := []struct {
+		name           string
+		fetchedFilters []SatPerKWeight
+		expectedMedian SatPerKWeight
+		expectedErr    error
+	}{
+		{
+			name:           "no filter data",
+			fetchedFilters: nil,
+			expectedErr:    errNoData,
+		},
+		{
+			name: "even number of filters",
+			fetchedFilters: []SatPerKWeight{
+				15000, 18000, 19000, 25000,
+			},
+			expectedMedian: SatPerKWeight(18500),
+		},
+		{
+			name: "odd number of filters",
+			fetchedFilters: []SatPerKWeight{
+				15000, 18000, 25000,
+			},
+			expectedMedian: SatPerKWeight(18000),
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			cb := func() ([]SatPerKWeight, error) {
+				return nil, nil
+			}
+			fm := newFilterManager(cb)
+
+			fm.updateMedian(test.fetchedFilters)
+
+			median, err := fm.FetchMedianFilter()
+			require.Equal(t, test.expectedErr, err)
+			require.Equal(t, test.expectedMedian, median)
+		})
+	}
+}