13566028875

Committed 27 Feb 2025 12:09PM UTC coverage: 49.396% (-9.4%) from 58.748%

Build # 13566028875

Build Type

Pull #9555

github

Committed by

ellemouton

Commit Message

graph/db: populate the graph cache in Start instead of during construction

In this commit, we move the graph cache population logic out of the
ChannelGraph constructor and into its Start method instead.

Pull Request Pull Request #9555: graph: extract cache from CRUD [6]

Run Details

34 of 54 new or added lines in 4 files covered. (62.96%)

27464 existing lines in 436 files now uncovered.

101095 of 204664 relevant lines covered (49.4%)

1.54 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/lnwallet/chainfee/filtermanager.go

package chainfee

import (
        "encoding/json"
        "fmt"
        "sort"
        "sync"
        "time"

        "github.com/btcsuite/btcd/btcutil"
        "github.com/btcsuite/btcd/rpcclient"
        "github.com/lightningnetwork/lnd/fn/v2"
)

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
}

1	package chainfee
2
3	import (
4	"encoding/json"
5	"fmt"
6	"sort"
7	"sync"
8	"time"
9
10	"github.com/btcsuite/btcd/btcutil"
11	"github.com/btcsuite/btcd/rpcclient"
12	"github.com/lightningnetwork/lnd/fn/v2"
13	)
14
15	const (
16	// fetchFilterInterval is the interval between successive fetches of
17	// our peers' feefilters.
18	fetchFilterInterval = time.Minute * 5
19
20	// minNumFilters is the minimum number of feefilters we need during a
21	// polling interval. If we have fewer than this, we won't consider the
22	// data.
23	minNumFilters = 6
24	)
25
26	var (
27	// errNoData is an error that's returned if fetchMedianFilter is
28	// called and there is no data available.
29	errNoData = fmt.Errorf("no data available")
30	)
31
32	// filterManager is responsible for determining an acceptable minimum fee to
33	// use based on our peers' feefilter values.
34	type filterManager struct {
35	// median stores the median of our outbound peer's feefilter values.
36	median fn.Option[SatPerKWeight]
37	medianMtx sync.RWMutex
38
39	fetchFunc func() ([]SatPerKWeight, error)
40
41	wg sync.WaitGroup
42	quit chan struct{}
43	}
44
45	// newFilterManager constructs a filterManager with a callback that fetches the
46	// set of peers' feefilters.
UNCOV 47	func newFilterManager(cb func() ([]SatPerKWeight, error)) *filterManager {	×
UNCOV 48	f := &filterManager{	×
UNCOV 49	quit: make(chan struct{}),	×
UNCOV 50	}	×
UNCOV 51		×
UNCOV 52	f.fetchFunc = cb	×
UNCOV 53		×
UNCOV 54	return f	×
UNCOV 55	}	×
56
57	// Start starts the filterManager.
58	func (f *filterManager) Start() {	×
59	f.wg.Add(1)	×
60	go f.fetchPeerFilters()	×
61	}	×
62
63	// Stop stops the filterManager.
64	func (f *filterManager) Stop() {	×
65	close(f.quit)	×
66	f.wg.Wait()	×
67	}	×
68
69	// fetchPeerFilters fetches our peers' feefilter values and calculates the
70	// median.
71	func (f *filterManager) fetchPeerFilters() {	×
72	defer f.wg.Done()	×
73		×
74	filterTicker := time.NewTicker(fetchFilterInterval)	×
75	defer filterTicker.Stop()	×
76		×
77	for {	×
78	select {	×
79	case <-filterTicker.C:	×
80	filters, err := f.fetchFunc()	×
81	if err != nil {	×
82	log.Errorf("Encountered err while fetching "+	×
83	"fee filters: %v", err)	×
84	return	×
85	}	×
86
87	f.updateMedian(filters)	×
88
89	case <-f.quit:	×
90	return	×
91	}
92	}
93	}
94
95	// fetchMedianFilter fetches the median feefilter value.
UNCOV 96	func (f *filterManager) FetchMedianFilter() (SatPerKWeight, error) {	×
UNCOV 97	f.medianMtx.RLock()	×
UNCOV 98	defer f.medianMtx.RUnlock()	×
UNCOV 99		×
UNCOV 100	// If there is no median, return errNoData so the caller knows to	×
UNCOV 101	// ignore the output and continue.	×
UNCOV 102	return f.median.UnwrapOrErr(errNoData)	×
UNCOV 103	}	×
104
105	type bitcoindPeerInfoResp struct {
106	Inbound bool `json:"inbound"`
107	MinFeeFilter float64 `json:"minfeefilter"`
108	}
109
110	func fetchBitcoindFilters(client *rpcclient.Client) ([]SatPerKWeight, error) {	×
111	resp, err := client.RawRequest("getpeerinfo", nil)	×
112	if err != nil {	×
113	return nil, err	×
114	}	×
115
116	var peerResps []bitcoindPeerInfoResp	×
117	err = json.Unmarshal(resp, &peerResps)	×
118	if err != nil {	×
119	return nil, err	×
120	}	×
121
122	// We filter for outbound peers since it is harder for an attacker to
123	// be our outbound peer and therefore harder for them to manipulate us
124	// into broadcasting high-fee or low-fee transactions.
125	var outboundPeerFilters []SatPerKWeight	×
126	for _, peerResp := range peerResps {	×
127	if peerResp.Inbound {	×
128	continue	×
129	}
130
131	// The value that bitcoind returns for the "minfeefilter" field
132	// is in fractions of a bitcoin that represents the satoshis
133	// per KvB. We need to convert this fraction to whole satoshis
134	// by multiplying with COIN. Then we need to convert the
135	// sats/KvB to sats/KW.
136	//
137	// Convert the sats/KvB from fractions of a bitcoin to whole
138	// satoshis.
139	filterKVByte := SatPerKVByte(	×
140	peerResp.MinFeeFilter * btcutil.SatoshiPerBitcoin,	×
141	)	×
142		×
143	if !isWithinBounds(filterKVByte) {	×
144	continue	×
145	}
146
147	// Convert KvB to KW and add it to outboundPeerFilters.
148	outboundPeerFilters = append(	×
149	outboundPeerFilters, filterKVByte.FeePerKWeight(),	×
150	)	×
151	}
152
153	// Check that we have enough data to use. We don't return an error as
154	// that would stop the querying goroutine.
155	if len(outboundPeerFilters) < minNumFilters {	×
156	return nil, nil	×
157	}	×
158
159	return outboundPeerFilters, nil	×
160	}
161
162	func fetchBtcdFilters(client *rpcclient.Client) ([]SatPerKWeight, error) {	×
163	resp, err := client.GetPeerInfo()	×
164	if err != nil {	×
165	return nil, err	×
166	}	×
167
168	var outboundPeerFilters []SatPerKWeight	×
169	for _, peerResp := range resp {	×
170	// We filter for outbound peers since it is harder for an	×
171	// attacker to be our outbound peer and therefore harder for	×
172	// them to manipulate us into broadcasting high-fee or low-fee	×
173	// transactions.	×
174	if peerResp.Inbound {	×
175	continue	×
176	}
177
178	// The feefilter is already in units of sat/KvB.
179	filter := SatPerKVByte(peerResp.FeeFilter)	×
180		×
181	if !isWithinBounds(filter) {	×
182	continue	×
183	}
184
185	outboundPeerFilters = append(	×
186	outboundPeerFilters, filter.FeePerKWeight(),	×
187	)	×
188	}
189
190	// Check that we have enough data to use. We don't return an error as
191	// that would stop the querying goroutine.
192	if len(outboundPeerFilters) < minNumFilters {	×
193	return nil, nil	×
194	}	×
195
196	return outboundPeerFilters, nil	×
197	}
198
199	// updateMedian takes a slice of feefilter values, computes the median, and
200	// updates our stored median value.
UNCOV 201	func (f *filterManager) updateMedian(feeFilters []SatPerKWeight) {	×
UNCOV 202	// If there are no elements, don't update.	×
UNCOV 203	numElements := len(feeFilters)	×
UNCOV 204	if numElements == 0 {	×
UNCOV 205	return	×
UNCOV 206	}	×
207
UNCOV 208	f.medianMtx.Lock()	×
UNCOV 209	defer f.medianMtx.Unlock()	×
UNCOV 210		×
UNCOV 211	// Log the new median.	×
UNCOV 212	median := med(feeFilters)	×
UNCOV 213	f.median = fn.Some(median)	×
UNCOV 214	log.Debugf("filterManager updated moving median to: %v",	×
UNCOV 215	median.FeePerKVByte())	×
216	}
217
218	// isWithinBounds returns false if the filter is unusable and true if it is.
219	func isWithinBounds(filter SatPerKVByte) bool {	×
220	// Ignore values of 0 and MaxSatoshi. A value of 0 likely means that	×
221	// the peer hasn't sent over a feefilter and a value of MaxSatoshi	×
222	// means the peer is using bitcoind and is in IBD.	×
223	switch filter {	×
224	case 0:	×
225	return false	×
226
227	case btcutil.MaxSatoshi:	×
228	return false	×
229	}
230
231	return true	×
232	}
233
234	// med calculates the median of a slice.
235	// NOTE: Passing in an empty slice will panic!
UNCOV 236	func med(f []SatPerKWeight) SatPerKWeight {	×
UNCOV 237	// Copy the original slice so that sorting doesn't modify the original.	×
UNCOV 238	fCopy := make([]SatPerKWeight, len(f))	×
UNCOV 239	copy(fCopy, f)	×
UNCOV 240		×
UNCOV 241	sort.Slice(fCopy, func(i, j int) bool {	×
UNCOV 242	return fCopy[i] < fCopy[j]	×
UNCOV 243	})	×
244
UNCOV 245	var median SatPerKWeight	×
UNCOV 246		×
UNCOV 247	numElements := len(fCopy)	×
UNCOV 248	switch numElements % 2 {	×
UNCOV 249	case 0:	×
UNCOV 250	// There's an even number of elements, so we need to average.	×
UNCOV 251	middle := numElements / 2	×
UNCOV 252	upper := fCopy[middle]	×
UNCOV 253	lower := fCopy[middle-1]	×
UNCOV 254	median = (upper + lower) / 2	×
255
UNCOV 256	case 1:	×
UNCOV 257	median = fCopy[numElements/2]	×
258	}
259
UNCOV 260	return median	×
261	}

lightningnetwork / lnd / 13566028875

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous