13157733617

Committed 05 Feb 2025 12:49PM UTC coverage: 57.712% (-1.1%) from 58.82%

Build # 13157733617

Build Type

Pull #9447

github

Committed by

yyforyongyu

Commit Message

sweep: rename methods for clarity

We now rename "third party" to "unknown" as the inputs can be spent via
an older sweeping tx, a third party (anchor), or a remote party (pin).
In fee bumper we don't have the info to distinguish the above cases, and
leave them to be further handled by the sweeper as it has more context.

Pull Request Pull Request #9447: sweep: start tracking input spending status in the fee bumper

Run Details

83 of 87 new or added lines in 2 files covered. (95.4%)

19472 existing lines in 252 files now uncovered.

103634 of 179570 relevant lines covered (57.71%)

24840.31 hits per line

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

89.59

/graph/db/graph_cache.go

package graphdb

import (
        "fmt"
        "sync"

        "github.com/btcsuite/btcd/btcutil"
        "github.com/lightningnetwork/lnd/graph/db/models"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/routing/route"
)

// GraphCacheNode is an interface for all the information the cache needs to know
// about a lightning node.
type GraphCacheNode interface {
        // PubKey is the node's public identity key.
        PubKey() route.Vertex

        // Features returns the node's p2p features.
        Features() *lnwire.FeatureVector

        // ForEachChannel iterates through all channels of a given node,
        // executing the passed callback with an edge info structure and the
        // policies of each end of the channel. The first edge policy is the
        // outgoing edge *to* the connecting node, while the second is the
        // incoming edge *from* the connecting node. If the callback returns an
        // error, then the iteration is halted with the error propagated back up
        // to the caller.
        ForEachChannel(kvdb.RTx,
                func(kvdb.RTx, *models.ChannelEdgeInfo,
                        *models.ChannelEdgePolicy,
                        *models.ChannelEdgePolicy) error) error
}

// DirectedChannel is a type that stores the channel information as seen from
// one side of the channel.
type DirectedChannel struct {
        // ChannelID is the unique identifier of this channel.
        ChannelID uint64

        // IsNode1 indicates if this is the node with the smaller public key.
        IsNode1 bool

        // OtherNode is the public key of the node on the other end of this
        // channel.
        OtherNode route.Vertex

        // Capacity is the announced capacity of this channel in satoshis.
        Capacity btcutil.Amount

        // OutPolicySet is a boolean that indicates whether the node has an
        // outgoing policy set. For pathfinding only the existence of the policy
        // is important to know, not the actual content.
        OutPolicySet bool

        // InPolicy is the incoming policy *from* the other node to this node.
        // In path finding, we're walking backward from the destination to the
        // source, so we're always interested in the edge that arrives to us
        // from the other node.
        InPolicy *models.CachedEdgePolicy

        // Inbound fees of this node.
        InboundFee lnwire.Fee
}

// DeepCopy creates a deep copy of the channel, including the incoming policy.
func (c *DirectedChannel) DeepCopy() *DirectedChannel {
        channelCopy := *c

        if channelCopy.InPolicy != nil {
                inPolicyCopy := *channelCopy.InPolicy
                channelCopy.InPolicy = &inPolicyCopy

                // The fields for the ToNode can be overwritten by the path
                // finding algorithm, which is why we need a deep copy in the
                // first place. So we always start out with nil values, just to
                // be sure they don't contain any old data.
                channelCopy.InPolicy.ToNodePubKey = nil
                channelCopy.InPolicy.ToNodeFeatures = nil
        }

        return &channelCopy
}

// GraphCache is a type that holds a minimal set of information of the public
// channel graph that can be used for pathfinding.
type GraphCache struct {
        nodeChannels map[route.Vertex]map[uint64]*DirectedChannel
        nodeFeatures map[route.Vertex]*lnwire.FeatureVector

        mtx sync.RWMutex
}

// NewGraphCache creates a new graphCache.
func NewGraphCache(preAllocNumNodes int) *GraphCache {
        return &GraphCache{
                nodeChannels: make(
                        map[route.Vertex]map[uint64]*DirectedChannel,
                        // A channel connects two nodes, so we can look it up
                        // from both sides, meaning we get double the number of
                        // entries.
                        preAllocNumNodes*2,
                ),
                nodeFeatures: make(
                        map[route.Vertex]*lnwire.FeatureVector,
                        preAllocNumNodes,
                ),
        }
}

// Stats returns statistics about the current cache size.
func (c *GraphCache) Stats() string {
        c.mtx.RLock()
        defer c.mtx.RUnlock()

        numChannels := 0
        for node := range c.nodeChannels {
                numChannels += len(c.nodeChannels[node])
        }
        return fmt.Sprintf("num_node_features=%d, num_nodes=%d, "+
                "num_channels=%d", len(c.nodeFeatures), len(c.nodeChannels),
                numChannels)
}

// AddNodeFeatures adds a graph node and its features to the cache.
func (c *GraphCache) AddNodeFeatures(node GraphCacheNode) {
        nodePubKey := node.PubKey()

        // Only hold the lock for a short time. The `ForEachChannel()` below is
        // possibly slow as it has to go to the backend, so we can unlock
        // between the calls. And the AddChannel() method will acquire its own
        // lock anyway.
        c.mtx.Lock()
        c.nodeFeatures[nodePubKey] = node.Features()
        c.mtx.Unlock()
}

// AddNode adds a graph node, including all the (directed) channels of that
// node.
func (c *GraphCache) AddNode(tx kvdb.RTx, node GraphCacheNode) error {
        c.AddNodeFeatures(node)

        return node.ForEachChannel(
                tx, func(tx kvdb.RTx, info *models.ChannelEdgeInfo,
                        outPolicy *models.ChannelEdgePolicy,
                        inPolicy *models.ChannelEdgePolicy) error {

                        c.AddChannel(info, outPolicy, inPolicy)

                        return nil
                },
        )
}

// AddChannel adds a non-directed channel, meaning that the order of policy 1
// and policy 2 does not matter, the directionality is extracted from the info
// and policy flags automatically. The policy will be set as the outgoing policy
// on one node and the incoming policy on the peer's side.
func (c *GraphCache) AddChannel(info *models.ChannelEdgeInfo,
        policy1 *models.ChannelEdgePolicy, policy2 *models.ChannelEdgePolicy) {

        if info == nil {
                return
        }

        if policy1 != nil && policy1.IsDisabled() &&
                policy2 != nil && policy2.IsDisabled() {

                return
        }

        // Create the edge entry for both nodes.
        c.mtx.Lock()
        c.updateOrAddEdge(info.NodeKey1Bytes, &DirectedChannel{
                ChannelID: info.ChannelID,
                IsNode1:   true,
                OtherNode: info.NodeKey2Bytes,
                Capacity:  info.Capacity,
        })
        c.updateOrAddEdge(info.NodeKey2Bytes, &DirectedChannel{
                ChannelID: info.ChannelID,
                IsNode1:   false,
                OtherNode: info.NodeKey1Bytes,
                Capacity:  info.Capacity,
        })
        c.mtx.Unlock()

        // The policy's node is always the to_node. So if policy 1 has to_node
        // of node 2 then we have the policy 1 as seen from node 1.
        if policy1 != nil {
                fromNode, toNode := info.NodeKey1Bytes, info.NodeKey2Bytes
                if policy1.ToNode != info.NodeKey2Bytes {
                        fromNode, toNode = toNode, fromNode
                }
                isEdge1 := policy1.ChannelFlags&lnwire.ChanUpdateDirection == 0
                c.UpdatePolicy(policy1, fromNode, toNode, isEdge1)
        }
        if policy2 != nil {
                fromNode, toNode := info.NodeKey2Bytes, info.NodeKey1Bytes
                if policy2.ToNode != info.NodeKey1Bytes {
                        fromNode, toNode = toNode, fromNode
                }
                isEdge1 := policy2.ChannelFlags&lnwire.ChanUpdateDirection == 0
                c.UpdatePolicy(policy2, fromNode, toNode, isEdge1)
        }
}

// updateOrAddEdge makes sure the edge information for a node is either updated
// if it already exists or is added to that node's list of channels.
func (c *GraphCache) updateOrAddEdge(node route.Vertex, edge *DirectedChannel) {
        if len(c.nodeChannels[node]) == 0 {
                c.nodeChannels[node] = make(map[uint64]*DirectedChannel)
        }

        c.nodeChannels[node][edge.ChannelID] = edge
}

// UpdatePolicy updates a single policy on both the from and to node. The order
// of the from and to node is not strictly important. But we assume that a
// channel edge was added beforehand so that the directed channel struct already
// exists in the cache.
func (c *GraphCache) UpdatePolicy(policy *models.ChannelEdgePolicy, fromNode,
        toNode route.Vertex, edge1 bool) {

        // Extract inbound fee if possible and available. If there is a decoding
        // error, ignore this policy.
        var inboundFee lnwire.Fee
        _, err := policy.ExtraOpaqueData.ExtractRecords(&inboundFee)
        if err != nil {
                log.Errorf("Failed to extract records from edge policy %v: %v",
                        policy.ChannelID, err)

                return
        }

        c.mtx.Lock()
        defer c.mtx.Unlock()

        updatePolicy := func(nodeKey route.Vertex) {
                if len(c.nodeChannels[nodeKey]) == 0 {
                        log.Warnf("Node=%v not found in graph cache", nodeKey)

                        return
                }

                channel, ok := c.nodeChannels[nodeKey][policy.ChannelID]
                if !ok {
                        log.Warnf("Channel=%v not found in graph cache",
                                policy.ChannelID)

                        return
                }

                // Edge 1 is defined as the policy for the direction of node1 to
                // node2.
                switch {
                // This is node 1, and it is edge 1, so this is the outgoing
                // policy for node 1.
                case channel.IsNode1 && edge1:
                        channel.OutPolicySet = true
                        channel.InboundFee = inboundFee

                // This is node 2, and it is edge 2, so this is the outgoing
                // policy for node 2.
                case !channel.IsNode1 && !edge1:
                        channel.OutPolicySet = true
                        channel.InboundFee = inboundFee

                // The other two cases left mean it's the inbound policy for the
                // node.
                default:
                        channel.InPolicy = models.NewCachedPolicy(policy)
                }
        }

        updatePolicy(fromNode)
        updatePolicy(toNode)
}

// RemoveNode completely removes a node and all its channels (including the
// peer's side).
func (c *GraphCache) RemoveNode(node route.Vertex) {
        c.mtx.Lock()
        defer c.mtx.Unlock()

        delete(c.nodeFeatures, node)

        // First remove all channels from the other nodes' lists.
        for _, channel := range c.nodeChannels[node] {
                c.removeChannelIfFound(channel.OtherNode, channel.ChannelID)
        }

        // Then remove our whole node completely.
        delete(c.nodeChannels, node)
}

// RemoveChannel removes a single channel between two nodes.
func (c *GraphCache) RemoveChannel(node1, node2 route.Vertex, chanID uint64) {
        c.mtx.Lock()
        defer c.mtx.Unlock()

        // Remove that one channel from both sides.
        c.removeChannelIfFound(node1, chanID)
        c.removeChannelIfFound(node2, chanID)
}

// removeChannelIfFound removes a single channel from one side.
func (c *GraphCache) removeChannelIfFound(node route.Vertex, chanID uint64) {
        if len(c.nodeChannels[node]) == 0 {
                return
        }

        delete(c.nodeChannels[node], chanID)
}

// UpdateChannel updates the channel edge information for a specific edge. We
// expect the edge to already exist and be known. If it does not yet exist, this
// call is a no-op.
func (c *GraphCache) UpdateChannel(info *models.ChannelEdgeInfo) {
        c.mtx.Lock()
        defer c.mtx.Unlock()

        if len(c.nodeChannels[info.NodeKey1Bytes]) == 0 ||
                len(c.nodeChannels[info.NodeKey2Bytes]) == 0 {

                return
        }

        channel, ok := c.nodeChannels[info.NodeKey1Bytes][info.ChannelID]
        if ok {
                // We only expect to be called when the channel is already
                // known.
                channel.Capacity = info.Capacity
                channel.OtherNode = info.NodeKey2Bytes
        }

        channel, ok = c.nodeChannels[info.NodeKey2Bytes][info.ChannelID]
        if ok {
                channel.Capacity = info.Capacity
                channel.OtherNode = info.NodeKey1Bytes
        }
}

// getChannels returns a copy of the passed node's channels or nil if there
// isn't any.
func (c *GraphCache) getChannels(node route.Vertex) []*DirectedChannel {
        c.mtx.RLock()
        defer c.mtx.RUnlock()

        channels, ok := c.nodeChannels[node]
        if !ok {
                return nil
        }

        features, ok := c.nodeFeatures[node]
        if !ok {
                // If the features were set to nil explicitly, that's fine here.
                // The router will overwrite the features of the destination
                // node with those found in the invoice if necessary. But if we
                // didn't yet get a node announcement we want to mimic the
                // behavior of the old DB based code that would always set an
                // empty feature vector instead of leaving it nil.
                features = lnwire.EmptyFeatureVector()
        }

        toNodeCallback := func() route.Vertex {
                return node
        }

        i := 0
        channelsCopy := make([]*DirectedChannel, len(channels))
        for _, channel := range channels {
                // We need to copy the channel and policy to avoid it being
                // updated in the cache if the path finding algorithm sets
                // fields on it (currently only the ToNodeFeatures of the
                // policy).
                channelCopy := channel.DeepCopy()
                if channelCopy.InPolicy != nil {
                        channelCopy.InPolicy.ToNodePubKey = toNodeCallback
                        channelCopy.InPolicy.ToNodeFeatures = features
                }

                channelsCopy[i] = channelCopy
                i++
        }

        return channelsCopy
}

// ForEachChannel invokes the given callback for each channel of the given node.
func (c *GraphCache) ForEachChannel(node route.Vertex,
        cb func(channel *DirectedChannel) error) error {

        // Obtain a copy of the node's channels. We need do this in order to
        // avoid deadlocks caused by interaction with the graph cache, channel
        // state and the graph database from multiple goroutines. This snapshot
        // is only used for path finding where being stale is acceptable since
        // the real world graph and our representation may always become
        // slightly out of sync for a short time and the actual channel state
        // is stored separately.
        channels := c.getChannels(node)
        for _, channel := range channels {
                if err := cb(channel); err != nil {
                        return err
                }
        }

        return nil
}

// ForEachNode iterates over the adjacency list of the graph, executing the
// call back for each node and the set of channels that emanate from the given
// node.
//
// NOTE: This method should be considered _read only_, the channels or nodes
// passed in MUST NOT be modified.
func (c *GraphCache) ForEachNode(cb func(node route.Vertex,
        channels map[uint64]*DirectedChannel) error) error {

        c.mtx.RLock()
        defer c.mtx.RUnlock()

        for node, channels := range c.nodeChannels {
                // We don't make a copy here since this is a read-only RPC
                // call. We also don't need the node features either for this
                // call.
                if err := cb(node, channels); err != nil {
                        return err
                }
        }

        return nil
}

// GetFeatures returns the features of the node with the given ID. If no
// features are known for the node, an empty feature vector is returned.
func (c *GraphCache) GetFeatures(node route.Vertex) *lnwire.FeatureVector {
        c.mtx.RLock()
        defer c.mtx.RUnlock()

        features, ok := c.nodeFeatures[node]
        if !ok || features == nil {
                // The router expects the features to never be nil, so we return
                // an empty feature set instead.
                return lnwire.EmptyFeatureVector()
        }

        return features
}

1	package graphdb
2
3	import (
4	"fmt"
5	"sync"
6
7	"github.com/btcsuite/btcd/btcutil"
8	"github.com/lightningnetwork/lnd/graph/db/models"
9	"github.com/lightningnetwork/lnd/kvdb"
10	"github.com/lightningnetwork/lnd/lnwire"
11	"github.com/lightningnetwork/lnd/routing/route"
12	)
13
14	// GraphCacheNode is an interface for all the information the cache needs to know
15	// about a lightning node.
16	type GraphCacheNode interface {
17	// PubKey is the node's public identity key.
18	PubKey() route.Vertex
19
20	// Features returns the node's p2p features.
21	Features() *lnwire.FeatureVector
22
23	// ForEachChannel iterates through all channels of a given node,
24	// executing the passed callback with an edge info structure and the
25	// policies of each end of the channel. The first edge policy is the
26	// outgoing edge to the connecting node, while the second is the
27	// incoming edge from the connecting node. If the callback returns an
28	// error, then the iteration is halted with the error propagated back up
29	// to the caller.
30	ForEachChannel(kvdb.RTx,
31	func(kvdb.RTx, *models.ChannelEdgeInfo,
32	*models.ChannelEdgePolicy,
33	*models.ChannelEdgePolicy) error) error
34	}
35
36	// DirectedChannel is a type that stores the channel information as seen from
37	// one side of the channel.
38	type DirectedChannel struct {
39	// ChannelID is the unique identifier of this channel.
40	ChannelID uint64
41
42	// IsNode1 indicates if this is the node with the smaller public key.
43	IsNode1 bool
44
45	// OtherNode is the public key of the node on the other end of this
46	// channel.
47	OtherNode route.Vertex
48
49	// Capacity is the announced capacity of this channel in satoshis.
50	Capacity btcutil.Amount
51
52	// OutPolicySet is a boolean that indicates whether the node has an
53	// outgoing policy set. For pathfinding only the existence of the policy
54	// is important to know, not the actual content.
55	OutPolicySet bool
56
57	// InPolicy is the incoming policy from the other node to this node.
58	// In path finding, we're walking backward from the destination to the
59	// source, so we're always interested in the edge that arrives to us
60	// from the other node.
61	InPolicy *models.CachedEdgePolicy
62
63	// Inbound fees of this node.
64	InboundFee lnwire.Fee
65	}
66
67	// DeepCopy creates a deep copy of the channel, including the incoming policy.
68	func (c DirectedChannel) DeepCopy() DirectedChannel {	1,336✔
69	channelCopy := *c	1,336✔
70		1,336✔
71	if channelCopy.InPolicy != nil {	2,646✔
72	inPolicyCopy := *channelCopy.InPolicy	1,310✔
73	channelCopy.InPolicy = &inPolicyCopy	1,310✔
74		1,310✔
75	// The fields for the ToNode can be overwritten by the path	1,310✔
76	// finding algorithm, which is why we need a deep copy in the	1,310✔
77	// first place. So we always start out with nil values, just to	1,310✔
78	// be sure they don't contain any old data.	1,310✔
79	channelCopy.InPolicy.ToNodePubKey = nil	1,310✔
80	channelCopy.InPolicy.ToNodeFeatures = nil	1,310✔
81	}	1,310✔
82
83	return &channelCopy	1,336✔
84	}
85
86	// GraphCache is a type that holds a minimal set of information of the public
87	// channel graph that can be used for pathfinding.
88	type GraphCache struct {
89	nodeChannels map[route.Vertex]map[uint64]*DirectedChannel
90	nodeFeatures map[route.Vertex]*lnwire.FeatureVector
91
92	mtx sync.RWMutex
93	}
94
95	// NewGraphCache creates a new graphCache.
96	func NewGraphCache(preAllocNumNodes int) *GraphCache {	142✔
97	return &GraphCache{	142✔
98	nodeChannels: make(	142✔
99	map[route.Vertex]map[uint64]*DirectedChannel,	142✔
100	// A channel connects two nodes, so we can look it up	142✔
101	// from both sides, meaning we get double the number of	142✔
102	// entries.	142✔
103	preAllocNumNodes*2,	142✔
104	),	142✔
105	nodeFeatures: make(	142✔
106	map[route.Vertex]*lnwire.FeatureVector,	142✔
107	preAllocNumNodes,	142✔
108	),	142✔
109	}	142✔
110	}	142✔
111
112	// Stats returns statistics about the current cache size.
113	func (c *GraphCache) Stats() string {	377✔
114	c.mtx.RLock()	377✔
115	defer c.mtx.RUnlock()	377✔
116		377✔
117	numChannels := 0	377✔
118	for node := range c.nodeChannels {	773✔
119	numChannels += len(c.nodeChannels[node])	396✔
120	}	396✔
121	return fmt.Sprintf("num_node_features=%d, num_nodes=%d, "+	377✔
122	"num_channels=%d", len(c.nodeFeatures), len(c.nodeChannels),	377✔
123	numChannels)	377✔
124	}
125
126	// AddNodeFeatures adds a graph node and its features to the cache.
127	func (c *GraphCache) AddNodeFeatures(node GraphCacheNode) {	714✔
128	nodePubKey := node.PubKey()	714✔
129		714✔
130	// Only hold the lock for a short time. The `ForEachChannel()` below is	714✔
131	// possibly slow as it has to go to the backend, so we can unlock	714✔
132	// between the calls. And the AddChannel() method will acquire its own	714✔
133	// lock anyway.	714✔
134	c.mtx.Lock()	714✔
135	c.nodeFeatures[nodePubKey] = node.Features()	714✔
136	c.mtx.Unlock()	714✔
137	}	714✔
138
139	// AddNode adds a graph node, including all the (directed) channels of that
140	// node.
141	func (c *GraphCache) AddNode(tx kvdb.RTx, node GraphCacheNode) error {	614✔
142	c.AddNodeFeatures(node)	614✔
143		614✔
144	return node.ForEachChannel(	614✔
145	tx, func(tx kvdb.RTx, info *models.ChannelEdgeInfo,	614✔
146	outPolicy *models.ChannelEdgePolicy,	614✔
147	inPolicy *models.ChannelEdgePolicy) error {	625✔
148		11✔
149	c.AddChannel(info, outPolicy, inPolicy)	11✔
150		11✔
151	return nil	11✔
152	},	11✔
153	)
154	}
155
156	// AddChannel adds a non-directed channel, meaning that the order of policy 1
157	// and policy 2 does not matter, the directionality is extracted from the info
158	// and policy flags automatically. The policy will be set as the outgoing policy
159	// on one node and the incoming policy on the peer's side.
160	func (c GraphCache) AddChannel(info models.ChannelEdgeInfo,
161	policy1 models.ChannelEdgePolicy, policy2 models.ChannelEdgePolicy) {	1,693✔
162		1,693✔
163	if info == nil {	1,693✔
164	return	×
165	}	×
166
167	if policy1 != nil && policy1.IsDisabled() &&	1,693✔
168	policy2 != nil && policy2.IsDisabled() {	1,693✔
UNCOV 169		×
UNCOV 170	return	×
UNCOV 171	}	×
172
173	// Create the edge entry for both nodes.
174	c.mtx.Lock()	1,693✔
175	c.updateOrAddEdge(info.NodeKey1Bytes, &DirectedChannel{	1,693✔
176	ChannelID: info.ChannelID,	1,693✔
177	IsNode1: true,	1,693✔
178	OtherNode: info.NodeKey2Bytes,	1,693✔
179	Capacity: info.Capacity,	1,693✔
180	})	1,693✔
181	c.updateOrAddEdge(info.NodeKey2Bytes, &DirectedChannel{	1,693✔
182	ChannelID: info.ChannelID,	1,693✔
183	IsNode1: false,	1,693✔
184	OtherNode: info.NodeKey1Bytes,	1,693✔
185	Capacity: info.Capacity,	1,693✔
186	})	1,693✔
187	c.mtx.Unlock()	1,693✔
188		1,693✔
189	// The policy's node is always the to_node. So if policy 1 has to_node	1,693✔
190	// of node 2 then we have the policy 1 as seen from node 1.	1,693✔
191	if policy1 != nil {	2,094✔
192	fromNode, toNode := info.NodeKey1Bytes, info.NodeKey2Bytes	401✔
193	if policy1.ToNode != info.NodeKey2Bytes {	403✔
194	fromNode, toNode = toNode, fromNode	2✔
195	}	2✔
196	isEdge1 := policy1.ChannelFlags&lnwire.ChanUpdateDirection == 0	401✔
197	c.UpdatePolicy(policy1, fromNode, toNode, isEdge1)	401✔
198	}
199	if policy2 != nil {	2,094✔
200	fromNode, toNode := info.NodeKey2Bytes, info.NodeKey1Bytes	401✔
201	if policy2.ToNode != info.NodeKey1Bytes {	403✔
202	fromNode, toNode = toNode, fromNode	2✔
203	}	2✔
204	isEdge1 := policy2.ChannelFlags&lnwire.ChanUpdateDirection == 0	401✔
205	c.UpdatePolicy(policy2, fromNode, toNode, isEdge1)	401✔
206	}
207	}
208
209	// updateOrAddEdge makes sure the edge information for a node is either updated
210	// if it already exists or is added to that node's list of channels.
211	func (c GraphCache) updateOrAddEdge(node route.Vertex, edge DirectedChannel) {	3,386✔
212	if len(c.nodeChannels[node]) == 0 {	4,241✔
213	c.nodeChannels[node] = make(map[uint64]*DirectedChannel)	855✔
214	}	855✔
215
216	c.nodeChannels[node][edge.ChannelID] = edge	3,386✔
217	}
218
219	// UpdatePolicy updates a single policy on both the from and to node. The order
220	// of the from and to node is not strictly important. But we assume that a
221	// channel edge was added beforehand so that the directed channel struct already
222	// exists in the cache.
223	func (c GraphCache) UpdatePolicy(policy models.ChannelEdgePolicy, fromNode,
224	toNode route.Vertex, edge1 bool) {	3,076✔
225		3,076✔
226	// Extract inbound fee if possible and available. If there is a decoding	3,076✔
227	// error, ignore this policy.	3,076✔
228	var inboundFee lnwire.Fee	3,076✔
229	_, err := policy.ExtraOpaqueData.ExtractRecords(&inboundFee)	3,076✔
230	if err != nil {	3,076✔
231	log.Errorf("Failed to extract records from edge policy %v: %v",	×
232	policy.ChannelID, err)	×
233		×
234	return	×
235	}	×
236
237	c.mtx.Lock()	3,076✔
238	defer c.mtx.Unlock()	3,076✔
239		3,076✔
240	updatePolicy := func(nodeKey route.Vertex) {	9,228✔
241	if len(c.nodeChannels[nodeKey]) == 0 {	6,152✔
242	log.Warnf("Node=%v not found in graph cache", nodeKey)	×
243		×
244	return	×
245	}	×
246
247	channel, ok := c.nodeChannels[nodeKey][policy.ChannelID]	6,152✔
248	if !ok {	6,152✔
249	log.Warnf("Channel=%v not found in graph cache",	×
250	policy.ChannelID)	×
251		×
252	return	×
253	}	×
254
255	// Edge 1 is defined as the policy for the direction of node1 to
256	// node2.
257	switch {	6,152✔
258	// This is node 1, and it is edge 1, so this is the outgoing
259	// policy for node 1.
260	case channel.IsNode1 && edge1:	1,541✔
261	channel.OutPolicySet = true	1,541✔
262	channel.InboundFee = inboundFee	1,541✔
263
264	// This is node 2, and it is edge 2, so this is the outgoing
265	// policy for node 2.
266	case !channel.IsNode1 && !edge1:	1,535✔
267	channel.OutPolicySet = true	1,535✔
268	channel.InboundFee = inboundFee	1,535✔
269
270	// The other two cases left mean it's the inbound policy for the
271	// node.
272	default:	3,076✔
273	channel.InPolicy = models.NewCachedPolicy(policy)	3,076✔
274	}
275	}
276
277	updatePolicy(fromNode)	3,076✔
278	updatePolicy(toNode)	3,076✔
279	}
280
281	// RemoveNode completely removes a node and all its channels (including the
282	// peer's side).
283	func (c *GraphCache) RemoveNode(node route.Vertex) {	60✔
284	c.mtx.Lock()	60✔
285	defer c.mtx.Unlock()	60✔
286		60✔
287	delete(c.nodeFeatures, node)	60✔
288		60✔
289	// First remove all channels from the other nodes' lists.	60✔
290	for _, channel := range c.nodeChannels[node] {	60✔
291	c.removeChannelIfFound(channel.OtherNode, channel.ChannelID)	×
292	}	×
293
294	// Then remove our whole node completely.
295	delete(c.nodeChannels, node)	60✔
296	}
297
298	// RemoveChannel removes a single channel between two nodes.
299	func (c *GraphCache) RemoveChannel(node1, node2 route.Vertex, chanID uint64) {	267✔
300	c.mtx.Lock()	267✔
301	defer c.mtx.Unlock()	267✔
302		267✔
303	// Remove that one channel from both sides.	267✔
304	c.removeChannelIfFound(node1, chanID)	267✔
305	c.removeChannelIfFound(node2, chanID)	267✔
306	}	267✔
307
308	// removeChannelIfFound removes a single channel from one side.
309	func (c *GraphCache) removeChannelIfFound(node route.Vertex, chanID uint64) {	534✔
310	if len(c.nodeChannels[node]) == 0 {	714✔
311	return	180✔
312	}	180✔
313
314	delete(c.nodeChannels[node], chanID)	354✔
315	}
316
317	// UpdateChannel updates the channel edge information for a specific edge. We
318	// expect the edge to already exist and be known. If it does not yet exist, this
319	// call is a no-op.
320	func (c GraphCache) UpdateChannel(info models.ChannelEdgeInfo) {	1✔
321	c.mtx.Lock()	1✔
322	defer c.mtx.Unlock()	1✔
323		1✔
324	if len(c.nodeChannels[info.NodeKey1Bytes]) == 0 \|\|	1✔
325	len(c.nodeChannels[info.NodeKey2Bytes]) == 0 {	1✔
326		×
327	return	×
328	}	×
329
330	channel, ok := c.nodeChannels[info.NodeKey1Bytes][info.ChannelID]	1✔
331	if ok {	2✔
332	// We only expect to be called when the channel is already	1✔
333	// known.	1✔
334	channel.Capacity = info.Capacity	1✔
335	channel.OtherNode = info.NodeKey2Bytes	1✔
336	}	1✔
337
338	channel, ok = c.nodeChannels[info.NodeKey2Bytes][info.ChannelID]	1✔
339	if ok {	2✔
340	channel.Capacity = info.Capacity	1✔
341	channel.OtherNode = info.NodeKey1Bytes	1✔
342	}	1✔
343	}
344
345	// getChannels returns a copy of the passed node's channels or nil if there
346	// isn't any.
347	func (c GraphCache) getChannels(node route.Vertex) []DirectedChannel {	501✔
348	c.mtx.RLock()	501✔
349	defer c.mtx.RUnlock()	501✔
350		501✔
351	channels, ok := c.nodeChannels[node]	501✔
352	if !ok {	508✔
353	return nil	7✔
354	}	7✔
355
356	features, ok := c.nodeFeatures[node]	494✔
357	if !ok {	508✔
358	// If the features were set to nil explicitly, that's fine here.	14✔
359	// The router will overwrite the features of the destination	14✔
360	// node with those found in the invoice if necessary. But if we	14✔
361	// didn't yet get a node announcement we want to mimic the	14✔
362	// behavior of the old DB based code that would always set an	14✔
363	// empty feature vector instead of leaving it nil.	14✔
364	features = lnwire.EmptyFeatureVector()	14✔
365	}	14✔
366
367	toNodeCallback := func() route.Vertex {	942✔
368	return node	448✔
369	}	448✔
370
371	i := 0	494✔
372	channelsCopy := make([]*DirectedChannel, len(channels))	494✔
373	for _, channel := range channels {	1,826✔
374	// We need to copy the channel and policy to avoid it being	1,332✔
375	// updated in the cache if the path finding algorithm sets	1,332✔
376	// fields on it (currently only the ToNodeFeatures of the	1,332✔
377	// policy).	1,332✔
378	channelCopy := channel.DeepCopy()	1,332✔
379	if channelCopy.InPolicy != nil {	2,639✔
380	channelCopy.InPolicy.ToNodePubKey = toNodeCallback	1,307✔
381	channelCopy.InPolicy.ToNodeFeatures = features	1,307✔
382	}	1,307✔
383
384	channelsCopy[i] = channelCopy	1,332✔
385	i++	1,332✔
386	}
387
388	return channelsCopy	494✔
389	}
390
391	// ForEachChannel invokes the given callback for each channel of the given node.
392	func (c *GraphCache) ForEachChannel(node route.Vertex,
393	cb func(channel *DirectedChannel) error) error {	501✔
394		501✔
395	// Obtain a copy of the node's channels. We need do this in order to	501✔
396	// avoid deadlocks caused by interaction with the graph cache, channel	501✔
397	// state and the graph database from multiple goroutines. This snapshot	501✔
398	// is only used for path finding where being stale is acceptable since	501✔
399	// the real world graph and our representation may always become	501✔
400	// slightly out of sync for a short time and the actual channel state	501✔
401	// is stored separately.	501✔
402	channels := c.getChannels(node)	501✔
403	for _, channel := range channels {	1,833✔
404	if err := cb(channel); err != nil {	1,332✔
405	return err	×
406	}	×
407	}
408
409	return nil	501✔
410	}
411
412	// ForEachNode iterates over the adjacency list of the graph, executing the
413	// call back for each node and the set of channels that emanate from the given
414	// node.
415	//
416	// NOTE: This method should be considered _read only_, the channels or nodes
417	// passed in MUST NOT be modified.
418	func (c *GraphCache) ForEachNode(cb func(node route.Vertex,
419	channels map[uint64]*DirectedChannel) error) error {	2✔
420		2✔
421	c.mtx.RLock()	2✔
422	defer c.mtx.RUnlock()	2✔
423		2✔
424	for node, channels := range c.nodeChannels {	6✔
425	// We don't make a copy here since this is a read-only RPC	4✔
426	// call. We also don't need the node features either for this	4✔
427	// call.	4✔
428	if err := cb(node, channels); err != nil {	4✔
429	return err	×
430	}	×
431	}
432
433	return nil	2✔
434	}
435
436	// GetFeatures returns the features of the node with the given ID. If no
437	// features are known for the node, an empty feature vector is returned.
438	func (c GraphCache) GetFeatures(node route.Vertex) lnwire.FeatureVector {	459✔
439	c.mtx.RLock()	459✔
440	defer c.mtx.RUnlock()	459✔
441		459✔
442	features, ok := c.nodeFeatures[node]	459✔
443	if !ok \|\| features == nil {	469✔
444	// The router expects the features to never be nil, so we return	10✔
445	// an empty feature set instead.	10✔
446	return lnwire.EmptyFeatureVector()	10✔
447	}	10✔
448
449	return features	449✔
450	}

lightningnetwork / lnd / 13157733617

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