multi: let ForEachChannel be a method on ChannelGraph

Having a `ForEachChannel` method on the `LightningNode` struct itself
results in a `kvdb.Backend` object needing to be stored within the
LightningNode struct. In this commit, this method is replaced with a
`ForEachNodeChannel` method on the `ChannelGraph` struct will perform
the same function without needing the db pointer to be stored within the
LightningNode. This change, the LightningNode struct more closely
represents the schema on disk.

The existing `ForEachNodeChannel` method on `ChannelGraph` is renamed to
`ForEachNodeDirectedChannel`. It performs a slightly different function
since it's call-back operates on Cached policies.
This commit is contained in:
Elle Mouton committed 2023-11-08 10:51:50 +02:00
1 parent dacb86fdbc
commit fa7c1e250b
7 files changed
+112 -76

No files matched your search

+19 -9
View File
@@ -53,6 +53,8 @@ func ChannelGraphFromDatabase(db *channeldb.ChannelGraph) ChannelGraph {
// channeldb.LightningNode. The wrapper method implement the autopilot.Node // channeldb.LightningNode. The wrapper method implement the autopilot.Node
// interface. // interface.
type dbNode struct { type dbNode struct {
db *channeldb.ChannelGraph
tx kvdb.RTx tx kvdb.RTx
node *channeldb.LightningNode node *channeldb.LightningNode
@@ -86,25 +88,30 @@ func (d dbNode) Addrs() []net.Addr {
// //
// NOTE: Part of the autopilot.Node interface. // NOTE: Part of the autopilot.Node interface.
func (d dbNode) ForEachChannel(cb func(ChannelEdge) error) error { func (d dbNode) ForEachChannel(cb func(ChannelEdge) error) error {
return d.node.ForEachChannel(d.tx, func(tx kvdb.RTx, return d.db.ForEachNodeChannel(d.tx, d.node.PubKeyBytes,
ei *channeldb.ChannelEdgeInfo, ep, _ *channeldb.ChannelEdgePolicy) error { func(tx kvdb.RTx, ei *channeldb.ChannelEdgeInfo, ep,
_ *channeldb.ChannelEdgePolicy) error {
// Skip channels for which no outgoing edge policy is available. // Skip channels for which no outgoing edge policy is
// available.
// //
// TODO(joostjager): Ideally the case where channels have a nil // TODO(joostjager): Ideally the case where channels
// policy should be supported, as autopilot is not looking at // have a nil policy should be supported, as autopilot
// the policies. For now, it is not easily possible to get a // is not looking at the policies. For now, it is not
// reference to the other end LightningNode object without // easily possible to get a reference to the other end
// retrieving the policy. // LightningNode object without retrieving the policy.
if ep == nil { if ep == nil {
return nil return nil
} }
edge := ChannelEdge{ edge := ChannelEdge{
ChanID: lnwire.NewShortChanIDFromInt(ep.ChannelID), ChanID: lnwire.NewShortChanIDFromInt(
ep.ChannelID,
),
Capacity: ei.Capacity, Capacity: ei.Capacity,
Peer: dbNode{ Peer: dbNode{
tx: tx, tx: tx,
db: d.db,
node: ep.Node, node: ep.Node,
}, },
} }
@@ -128,6 +135,7 @@ func (d *databaseChannelGraph) ForEachNode(cb func(Node) error) error {
} }
node := dbNode{ node := dbNode{
db: d.db,
tx: tx, tx: tx,
node: n, node: n,
} }
@@ -266,6 +274,7 @@ func (d *databaseChannelGraph) addRandChannel(node1, node2 *btcec.PublicKey,
ChanID: chanID, ChanID: chanID,
Capacity: capacity, Capacity: capacity,
Peer: dbNode{ Peer: dbNode{
db: d.db,
node: vertex1, node: vertex1,
}, },
}, },
@@ -273,6 +282,7 @@ func (d *databaseChannelGraph) addRandChannel(node1, node2 *btcec.PublicKey,
ChanID: chanID, ChanID: chanID,
Capacity: capacity, Capacity: capacity,
Peer: dbNode{ Peer: dbNode{
db: d.db,
node: vertex2, node: vertex2,
}, },
}, },
+30 -27
View File
@@ -459,16 +459,14 @@ func (c *ChannelGraph) ForEachChannel(cb func(*ChannelEdgeInfo,
}, func() {}) }, func() {})
} }
// ForEachNodeChannel iterates through all channels of a given node, executing the // ForEachNodeDirectedChannel iterates through all channels of a given node,
// passed callback with an edge info structure and the policies of each end // executing the passed callback on the directed edge representing the channel
// of the channel. The first edge policy is the outgoing edge *to* the // and its incoming policy. If the callback returns an error, then the iteration
// connecting node, while the second is the incoming edge *from* the // is halted with the error propagated back up to the caller.
// connecting node. If the callback returns an error, then the iteration is
// halted with the error propagated back up to the caller.
// //
// Unknown policies are passed into the callback as nil values. // Unknown policies are passed into the callback as nil values.
func (c *ChannelGraph) ForEachNodeChannel(tx kvdb.RTx, node route.Vertex, func (c *ChannelGraph) ForEachNodeDirectedChannel(tx kvdb.RTx,
cb func(channel *DirectedChannel) error) error { node route.Vertex, cb func(channel *DirectedChannel) error) error {
if c.graphCache != nil { if c.graphCache != nil {
return c.graphCache.ForEachChannel(node, cb) return c.graphCache.ForEachChannel(node, cb)
@@ -557,8 +555,9 @@ func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex,
return c.ForEachNode(func(tx kvdb.RTx, node *LightningNode) error { return c.ForEachNode(func(tx kvdb.RTx, node *LightningNode) error {
channels := make(map[uint64]*DirectedChannel) channels := make(map[uint64]*DirectedChannel)
err := node.ForEachChannel(tx, func(tx kvdb.RTx, err := c.ForEachNodeChannel(tx, node.PubKeyBytes,
e *ChannelEdgeInfo, p1 *ChannelEdgePolicy, func(tx kvdb.RTx, e *ChannelEdgeInfo,
p1 *ChannelEdgePolicy,
p2 *ChannelEdgePolicy) error { p2 *ChannelEdgePolicy) error {
toNodeCallback := func() route.Vertex { toNodeCallback := func() route.Vertex {
@@ -574,13 +573,16 @@ func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex,
var cachedInPolicy *CachedEdgePolicy var cachedInPolicy *CachedEdgePolicy
if p2 != nil { if p2 != nil {
cachedInPolicy := NewCachedPolicy(p2) cachedInPolicy := NewCachedPolicy(p2)
cachedInPolicy.ToNodePubKey = toNodeCallback cachedInPolicy.ToNodePubKey =
cachedInPolicy.ToNodeFeatures = toNodeFeatures toNodeCallback
cachedInPolicy.ToNodeFeatures =
toNodeFeatures
} }
directedChannel := &DirectedChannel{ directedChannel := &DirectedChannel{
ChannelID: e.ChannelID, ChannelID: e.ChannelID,
IsNode1: node.PubKeyBytes == e.NodeKey1Bytes, IsNode1: node.PubKeyBytes ==
e.NodeKey1Bytes,
OtherNode: e.NodeKey2Bytes, OtherNode: e.NodeKey2Bytes,
Capacity: e.Capacity, Capacity: e.Capacity,
OutPolicySet: p1 != nil, OutPolicySet: p1 != nil,
@@ -588,7 +590,8 @@ func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex,
} }
if node.PubKeyBytes == e.NodeKey2Bytes { if node.PubKeyBytes == e.NodeKey2Bytes {
directedChannel.OtherNode = e.NodeKey1Bytes directedChannel.OtherNode =
e.NodeKey1Bytes
} }
channels[e.ChannelID] = directedChannel channels[e.ChannelID] = directedChannel
@@ -2740,15 +2743,18 @@ func (l *LightningNode) NodeAnnouncement(signed bool) (*lnwire.NodeAnnouncement,
// isPublic determines whether the node is seen as public within the graph from // isPublic determines whether the node is seen as public within the graph from
// the source node's point of view. An existing database transaction can also be // the source node's point of view. An existing database transaction can also be
// specified. // specified.
func (l *LightningNode) isPublic(tx kvdb.RTx, sourcePubKey []byte) (bool, error) { func (c *ChannelGraph) isPublic(tx kvdb.RTx, nodePub route.Vertex,
sourcePubKey []byte) (bool, error) {
// In order to determine whether this node is publicly advertised within // In order to determine whether this node is publicly advertised within
// the graph, we'll need to look at all of its edges and check whether // the graph, we'll need to look at all of its edges and check whether
// they extend to any other node than the source node. errDone will be // they extend to any other node than the source node. errDone will be
// used to terminate the check early. // used to terminate the check early.
nodeIsPublic := false nodeIsPublic := false
errDone := errors.New("done") errDone := errors.New("done")
err := l.ForEachChannel(tx, func(_ kvdb.RTx, info *ChannelEdgeInfo, err := c.ForEachNodeChannel(tx, nodePub, func(tx kvdb.RTx,
_, _ *ChannelEdgePolicy) error { info *ChannelEdgeInfo, _ *ChannelEdgePolicy,
_ *ChannelEdgePolicy) error {
// If this edge doesn't extend to the source node, we'll // If this edge doesn't extend to the source node, we'll
// terminate our search as we can now conclude that the node is // terminate our search as we can now conclude that the node is
@@ -3003,10 +3009,10 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend,
return traversal(tx) return traversal(tx)
} }
// ForEachChannel iterates through all channels of this node, executing the // ForEachNodeChannel iterates through all channels of the given node,
// passed callback with an edge info structure and the policies of each end // executing the passed callback with an edge info structure and the policies
// of the channel. The first edge policy is the outgoing edge *to* the // of each end of the channel. The first edge policy is the outgoing edge *to*
// connecting node, while the second is the incoming edge *from* the // the connecting node, while the second is the incoming edge *from* the
// connecting node. If the callback returns an error, then the iteration is // connecting node. If the callback returns an error, then the iteration is
// halted with the error propagated back up to the caller. // halted with the error propagated back up to the caller.
// //
@@ -3016,14 +3022,11 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend,
// should be passed as the first argument. Otherwise the first argument should // should be passed as the first argument. Otherwise the first argument should
// be nil and a fresh transaction will be created to execute the graph // be nil and a fresh transaction will be created to execute the graph
// traversal. // traversal.
func (l *LightningNode) ForEachChannel(tx kvdb.RTx, func (c *ChannelGraph) ForEachNodeChannel(tx kvdb.RTx, nodePub route.Vertex,
cb func(kvdb.RTx, *ChannelEdgeInfo, *ChannelEdgePolicy, cb func(kvdb.RTx, *ChannelEdgeInfo, *ChannelEdgePolicy,
*ChannelEdgePolicy) error) error { *ChannelEdgePolicy) error) error {
nodePub := l.PubKeyBytes[:] return nodeTraversal(tx, nodePub[:], c.db, cb)
db := l.db
return nodeTraversal(tx, nodePub, db, cb)
} }
// ChannelEdgeInfo represents a fully authenticated channel along with all its // ChannelEdgeInfo represents a fully authenticated channel along with all its
@@ -3718,7 +3721,7 @@ func (c *ChannelGraph) IsPublicNode(pubKey [33]byte) (bool, error) {
return err return err
} }
nodeIsPublic, err = node.isPublic(tx, ourPubKey) nodeIsPublic, err = c.isPublic(tx, node.PubKeyBytes, ourPubKey)
return err return err
}, func() { }, func() {
nodeIsPublic = false nodeIsPublic = false
+27 -15
View File
@@ -1056,24 +1056,33 @@ func TestGraphTraversal(t *testing.T) {
// outgoing channels for a particular node. // outgoing channels for a particular node.
numNodeChans := 0 numNodeChans := 0
firstNode, secondNode := nodeList[0], nodeList[1] firstNode, secondNode := nodeList[0], nodeList[1]
err = firstNode.ForEachChannel(nil, func(_ kvdb.RTx, _ *ChannelEdgeInfo, err = graph.ForEachNodeChannel(nil, firstNode.PubKeyBytes,
outEdge, inEdge *ChannelEdgePolicy) error { func(_ kvdb.RTx, _ *ChannelEdgeInfo, outEdge,
inEdge *ChannelEdgePolicy) error {
// All channels between first and second node should have fully // All channels between first and second node should
// (both sides) specified policies. // have fully (both sides) specified policies.
if inEdge == nil || outEdge == nil { if inEdge == nil || outEdge == nil {
return fmt.Errorf("channel policy not present") return fmt.Errorf("channel policy not present")
} }
// Each should indicate that it's outgoing (pointed // Each should indicate that it's outgoing (pointed
// towards the second node). // towards the second node).
if !bytes.Equal(outEdge.Node.PubKeyBytes[:], secondNode.PubKeyBytes[:]) { if !bytes.Equal(
outEdge.Node.PubKeyBytes[:],
secondNode.PubKeyBytes[:],
) {
return fmt.Errorf("wrong outgoing edge") return fmt.Errorf("wrong outgoing edge")
} }
// The incoming edge should also indicate that it's pointing to // The incoming edge should also indicate that it's
// the origin node. // pointing to the origin node.
if !bytes.Equal(inEdge.Node.PubKeyBytes[:], firstNode.PubKeyBytes[:]) { if !bytes.Equal(
inEdge.Node.PubKeyBytes[:],
firstNode.PubKeyBytes[:],
) {
return fmt.Errorf("wrong outgoing edge") return fmt.Errorf("wrong outgoing edge")
} }
@@ -2280,8 +2289,9 @@ func TestIncompleteChannelPolicies(t *testing.T) {
// Ensure that channel is reported with unknown policies. // Ensure that channel is reported with unknown policies.
checkPolicies := func(node *LightningNode, expectedIn, expectedOut bool) { checkPolicies := func(node *LightningNode, expectedIn, expectedOut bool) {
calls := 0 calls := 0
err := node.ForEachChannel(nil, func(_ kvdb.RTx, _ *ChannelEdgeInfo, err := graph.ForEachNodeChannel(nil, node.PubKeyBytes,
outEdge, inEdge *ChannelEdgePolicy) error { func(_ kvdb.RTx, _ *ChannelEdgeInfo, outEdge,
inEdge *ChannelEdgePolicy) error {
if !expectedOut && outEdge != nil { if !expectedOut && outEdge != nil {
t.Fatalf("Expected no outgoing policy") t.Fatalf("Expected no outgoing policy")
@@ -3470,8 +3480,8 @@ func BenchmarkForEachChannel(b *testing.B) {
} }
} }
// TestGraphCacheForEachNodeChannel tests that the ForEachNodeChannel method // TestGraphCacheForEachNodeChannel tests that the ForEachNodeDirectedChannel
// works as expected, and is able to handle nil self edges. // method works as expected, and is able to handle nil self edges.
func TestGraphCacheForEachNodeChannel(t *testing.T) { func TestGraphCacheForEachNodeChannel(t *testing.T) {
graph, err := MakeTestGraph(t) graph, err := MakeTestGraph(t)
require.NoError(t, err) require.NoError(t, err)
@@ -3498,11 +3508,13 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) {
// We should be able to accumulate the single channel added, even // We should be able to accumulate the single channel added, even
// though we have a nil edge policy here. // though we have a nil edge policy here.
var numChans int var numChans int
err = graph.ForEachNodeChannel(nil, node1.PubKeyBytes, err = graph.ForEachNodeDirectedChannel(nil, node1.PubKeyBytes,
func(channel *DirectedChannel) error { func(_ *DirectedChannel) error {
numChans++ numChans++
return nil return nil
}) },
)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, numChans, 1) require.Equal(t, numChans, 1)
+1 -1
View File
@@ -75,7 +75,7 @@ func (g *CachedGraph) Close() error {
func (g *CachedGraph) forEachNodeChannel(nodePub route.Vertex, func (g *CachedGraph) forEachNodeChannel(nodePub route.Vertex,
cb func(channel *channeldb.DirectedChannel) error) error { cb func(channel *channeldb.DirectedChannel) error) error {
return g.graph.ForEachNodeChannel(g.tx, nodePub, cb) return g.graph.ForEachNodeDirectedChannel(g.tx, nodePub, cb)
} }
// sourceNode returns the source node of the graph. // sourceNode returns the source node of the graph.
+8 -5
View File
@@ -2824,16 +2824,19 @@ func (r *ChannelRouter) ForEachNode(
func (r *ChannelRouter) ForAllOutgoingChannels(cb func(kvdb.RTx, func (r *ChannelRouter) ForAllOutgoingChannels(cb func(kvdb.RTx,
*channeldb.ChannelEdgeInfo, *channeldb.ChannelEdgePolicy) error) error { *channeldb.ChannelEdgeInfo, *channeldb.ChannelEdgePolicy) error) error {
return r.selfNode.ForEachChannel(nil, func(tx kvdb.RTx, return r.cfg.Graph.ForEachNodeChannel(nil, r.selfNode.PubKeyBytes,
c *channeldb.ChannelEdgeInfo, func(tx kvdb.RTx, c *channeldb.ChannelEdgeInfo,
e, _ *channeldb.ChannelEdgePolicy) error { e *channeldb.ChannelEdgePolicy,
_ *channeldb.ChannelEdgePolicy) error {
if e == nil { if e == nil {
return fmt.Errorf("channel from self node has no policy") return fmt.Errorf("channel from self node " +
"has no policy")
} }
return cb(tx, c, e) return cb(tx, c, e)
}) },
)
} }
// AddProof updates the channel edge info with proof which is needed to // AddProof updates the channel edge info with proof which is needed to
+26 -18
View File
@@ -6151,18 +6151,19 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context,
channels []*lnrpc.ChannelEdge channels []*lnrpc.ChannelEdge
) )
if err := node.ForEachChannel(nil, func(_ kvdb.RTx, err = graph.ForEachNodeChannel(nil, node.PubKeyBytes,
edge *channeldb.ChannelEdgeInfo, func(_ kvdb.RTx, edge *channeldb.ChannelEdgeInfo,
c1, c2 *channeldb.ChannelEdgePolicy) error { c1, c2 *channeldb.ChannelEdgePolicy) error {
numChannels++ numChannels++
totalCapacity += edge.Capacity totalCapacity += edge.Capacity
// Only populate the node's channels if the user requested them. // Only populate the node's channels if the user
// requested them.
if in.IncludeChannels { if in.IncludeChannels {
// Do not include unannounced channels - private // Do not include unannounced channels - private
// channels or public channels whose authentication // channels or public channels whose
// proof were not confirmed yet. // authentication proof were not confirmed yet.
if edge.AuthProof == nil { if edge.AuthProof == nil {
return nil return nil
} }
@@ -6174,7 +6175,9 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context,
} }
return nil return nil
}); err != nil { },
)
if err != nil {
return nil, err return nil, err
} }
@@ -6763,24 +6766,28 @@ func (r *rpcServer) FeeReport(ctx context.Context,
} }
var feeReports []*lnrpc.ChannelFeeReport var feeReports []*lnrpc.ChannelFeeReport
err = selfNode.ForEachChannel(nil, func(_ kvdb.RTx, chanInfo *channeldb.ChannelEdgeInfo, err = channelGraph.ForEachNodeChannel(nil, selfNode.PubKeyBytes,
func(_ kvdb.RTx, chanInfo *channeldb.ChannelEdgeInfo,
edgePolicy, _ *channeldb.ChannelEdgePolicy) error { edgePolicy, _ *channeldb.ChannelEdgePolicy) error {
// Self node should always have policies for its channels. // Self node should always have policies for its
// channels.
if edgePolicy == nil { if edgePolicy == nil {
return fmt.Errorf("no policy for outgoing channel %v ", return fmt.Errorf("no policy for outgoing "+
chanInfo.ChannelID) "channel %v ", chanInfo.ChannelID)
} }
// We'll compute the effective fee rate by converting from a // We'll compute the effective fee rate by converting
// fixed point fee rate to a floating point fee rate. The fee // from a fixed point fee rate to a floating point fee
// rate field in the database the amount of mSAT charged per // rate. The fee rate field in the database the amount
// 1mil mSAT sent, so will divide by this to get the proper fee // of mSAT charged per 1mil mSAT sent, so will divide by
// rate. // this to get the proper fee rate.
feeRateFixedPoint := edgePolicy.FeeProportionalMillionths feeRateFixedPoint :=
edgePolicy.FeeProportionalMillionths
feeRate := float64(feeRateFixedPoint) / feeBase feeRate := float64(feeRateFixedPoint) / feeBase
// TODO(roasbeef): also add stats for revenue for each channel // TODO(roasbeef): also add stats for revenue for each
// channel
feeReports = append(feeReports, &lnrpc.ChannelFeeReport{ feeReports = append(feeReports, &lnrpc.ChannelFeeReport{
ChanId: chanInfo.ChannelID, ChanId: chanInfo.ChannelID,
ChannelPoint: chanInfo.ChannelPoint.String(), ChannelPoint: chanInfo.ChannelPoint.String(),
@@ -6790,7 +6797,8 @@ func (r *rpcServer) FeeReport(ctx context.Context,
}) })
return nil return nil
}) },
)
if err != nil { if err != nil {
return nil, err return nil, err
} }
+1 -1
View File
@@ -3092,7 +3092,7 @@ func (s *server) establishPersistentConnections() error {
// TODO(roasbeef): instead iterate over link nodes and query graph for // TODO(roasbeef): instead iterate over link nodes and query graph for
// each of the nodes. // each of the nodes.
selfPub := s.identityECDH.PubKey().SerializeCompressed() selfPub := s.identityECDH.PubKey().SerializeCompressed()
err = sourceNode.ForEachChannel(nil, func( err = s.graphDB.ForEachNodeChannel(nil, sourceNode.PubKeyBytes, func(
tx kvdb.RTx, tx kvdb.RTx,
chanInfo *channeldb.ChannelEdgeInfo, chanInfo *channeldb.ChannelEdgeInfo,
policy, _ *channeldb.ChannelEdgePolicy) error { policy, _ *channeldb.ChannelEdgePolicy) error {