multi: use minimal policy in cache

This commit is contained in:
Oliver Gugger committed 2021-09-29 17:00:04 +02:00
1 parent 15d3f62d5e
commit 1d1c42f9ba
19 files changed
+629 -227

No files matched your search

-1
View File
@@ -2315,7 +2315,6 @@ func updateEdgePolicy(tx kvdb.RwTx, edge *ChannelEdgePolicy,
)
copy(fromNodePubKey[:], fromNode)
copy(toNodePubKey[:], toNode)
// TODO(guggero): Fetch lightning nodes before updating the cache!
graphCache.UpdatePolicy(edge, fromNodePubKey, toNodePubKey, isUpdate1)
return isUpdate1, nil
+139 -16
View File
@@ -32,6 +32,92 @@ type GraphCacheNode interface {
*ChannelEdgePolicy) error) error
}
// CachedEdgePolicy is a struct that only caches the information of a
// ChannelEdgePolicy that we actually use for pathfinding and therefore need to
// store in the cache.
type CachedEdgePolicy struct {
// ChannelID is the unique channel ID for the channel. The first 3
// bytes are the block height, the next 3 the index within the block,
// and the last 2 bytes are the output index for the channel.
ChannelID uint64
// MessageFlags is a bitfield which indicates the presence of optional
// fields (like max_htlc) in the policy.
MessageFlags lnwire.ChanUpdateMsgFlags
// ChannelFlags is a bitfield which signals the capabilities of the
// channel as well as the directed edge this update applies to.
ChannelFlags lnwire.ChanUpdateChanFlags
// TimeLockDelta is the number of blocks this node will subtract from
// the expiry of an incoming HTLC. This value expresses the time buffer
// the node would like to HTLC exchanges.
TimeLockDelta uint16
// MinHTLC is the smallest value HTLC this node will forward, expressed
// in millisatoshi.
MinHTLC lnwire.MilliSatoshi
// MaxHTLC is the largest value HTLC this node will forward, expressed
// in millisatoshi.
MaxHTLC lnwire.MilliSatoshi
// FeeBaseMSat is the base HTLC fee that will be charged for forwarding
// ANY HTLC, expressed in mSAT's.
FeeBaseMSat lnwire.MilliSatoshi
// FeeProportionalMillionths is the rate that the node will charge for
// HTLCs for each millionth of a satoshi forwarded.
FeeProportionalMillionths lnwire.MilliSatoshi
// ToNodePubKey is a function that returns the to node of a policy.
// Since we only ever store the inbound policy, this is always the node
// that we query the channels for in ForEachChannel(). Therefore, we can
// save a lot of space by not storing this information in the memory and
// instead just set this function when we copy the policy from cache in
// ForEachChannel().
ToNodePubKey func() route.Vertex
// ToNodeFeatures are the to node's features. They are never set while
// the edge is in the cache, only on the copy that is returned in
// ForEachChannel().
ToNodeFeatures *lnwire.FeatureVector
}
// ComputeFee computes the fee to forward an HTLC of `amt` milli-satoshis over
// the passed active payment channel. This value is currently computed as
// specified in BOLT07, but will likely change in the near future.
func (c *CachedEdgePolicy) ComputeFee(
amt lnwire.MilliSatoshi) lnwire.MilliSatoshi {
return c.FeeBaseMSat + (amt*c.FeeProportionalMillionths)/feeRateParts
}
// ComputeFeeFromIncoming computes the fee to forward an HTLC given the incoming
// amount.
func (c *CachedEdgePolicy) ComputeFeeFromIncoming(
incomingAmt lnwire.MilliSatoshi) lnwire.MilliSatoshi {
return incomingAmt - divideCeil(
feeRateParts*(incomingAmt-c.FeeBaseMSat),
feeRateParts+c.FeeProportionalMillionths,
)
}
// NewCachedPolicy turns a full policy into a minimal one that can be cached.
func NewCachedPolicy(policy *ChannelEdgePolicy) *CachedEdgePolicy {
return &CachedEdgePolicy{
ChannelID: policy.ChannelID,
MessageFlags: policy.MessageFlags,
ChannelFlags: policy.ChannelFlags,
TimeLockDelta: policy.TimeLockDelta,
MinHTLC: policy.MinHTLC,
MaxHTLC: policy.MaxHTLC,
FeeBaseMSat: policy.FeeBaseMSat,
FeeProportionalMillionths: policy.FeeProportionalMillionths,
}
}
// DirectedChannel is a type that stores the channel information as seen from
// one side of the channel.
type DirectedChannel struct {
@@ -48,11 +134,35 @@ type DirectedChannel struct {
// Capacity is the announced capacity of this channel in satoshis.
Capacity btcutil.Amount
// OutPolicy is the outgoing policy from this node *to* the other node.
OutPolicy *ChannelEdgePolicy
// 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.
InPolicy *ChannelEdgePolicy
// 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 *CachedEdgePolicy
}
// 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
@@ -181,15 +291,6 @@ func (c *GraphCache) updateOrAddEdge(node route.Vertex, edge *DirectedChannel) {
func (c *GraphCache) UpdatePolicy(policy *ChannelEdgePolicy, fromNode,
toNode route.Vertex, edge1 bool) {
// If a policy's node is nil, we can't cache it yet as that would lead
// to problems in pathfinding.
if policy.Node == nil {
// TODO(guggero): Fix this problem!
log.Warnf("Cannot cache policy because of missing node (from "+
"%x to %x)", fromNode[:], toNode[:])
return
}
c.mtx.Lock()
defer c.mtx.Unlock()
@@ -209,17 +310,17 @@ func (c *GraphCache) UpdatePolicy(policy *ChannelEdgePolicy, fromNode,
// This is node 1, and it is edge 1, so this is the outgoing
// policy for node 1.
case channel.IsNode1 && edge1:
channel.OutPolicy = policy
channel.OutPolicySet = true
// This is node 2, and it is edge 2, so this is the outgoing
// policy for node 2.
case !channel.IsNode1 && !edge1:
channel.OutPolicy = policy
channel.OutPolicySet = true
// The other two cases left mean it's the inbound policy for the
// node.
default:
channel.InPolicy = policy
channel.InPolicy = NewCachedPolicy(policy)
}
}
@@ -303,8 +404,30 @@ func (c *GraphCache) ForEachChannel(node route.Vertex,
return nil
}
features, ok := c.nodeFeatures[node]
if !ok {
log.Warnf("Node %v has no features defined, falling back to "+
"default feature vector for path finding", node)
features = lnwire.EmptyFeatureVector()
}
toNodeCallback := func() route.Vertex {
return node
}
for _, channel := range channels {
if err := cb(channel); err != nil {
// 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
}
if err := cb(channelCopy); err != nil {
return err
}
}
+45 -8
View File
@@ -63,18 +63,25 @@ func TestGraphCacheAddNode(t *testing.T) {
runTest := func(nodeA, nodeB route.Vertex) {
t.Helper()
channelFlagA, channelFlagB := 0, 1
if nodeA == pubKey2 {
channelFlagA, channelFlagB = 1, 0
}
outPolicy1 := &ChannelEdgePolicy{
ChannelID: 1000,
ChannelFlags: 0,
ChannelFlags: lnwire.ChanUpdateChanFlags(channelFlagA),
Node: &LightningNode{
PubKeyBytes: nodeB,
Features: lnwire.EmptyFeatureVector(),
},
}
inPolicy1 := &ChannelEdgePolicy{
ChannelID: 1000,
ChannelFlags: 1,
ChannelFlags: lnwire.ChanUpdateChanFlags(channelFlagB),
Node: &LightningNode{
PubKeyBytes: nodeA,
Features: lnwire.EmptyFeatureVector(),
},
}
node := &node{
@@ -93,18 +100,48 @@ func TestGraphCacheAddNode(t *testing.T) {
cache := NewGraphCache()
require.NoError(t, cache.AddNode(nil, node))
fromChannels := cache.nodeChannels[nodeA]
toChannels := cache.nodeChannels[nodeB]
var fromChannels, toChannels []*DirectedChannel
_ = cache.ForEachChannel(nodeA, func(c *DirectedChannel) error {
fromChannels = append(fromChannels, c)
return nil
})
_ = cache.ForEachChannel(nodeB, func(c *DirectedChannel) error {
toChannels = append(toChannels, c)
return nil
})
require.Len(t, fromChannels, 1)
require.Len(t, toChannels, 1)
require.Equal(t, outPolicy1, fromChannels[0].OutPolicy)
require.Equal(t, inPolicy1, fromChannels[0].InPolicy)
require.Equal(t, outPolicy1 != nil, fromChannels[0].OutPolicySet)
assertCachedPolicyEqual(t, inPolicy1, fromChannels[0].InPolicy)
require.Equal(t, inPolicy1, toChannels[0].OutPolicy)
require.Equal(t, outPolicy1, toChannels[0].InPolicy)
require.Equal(t, inPolicy1 != nil, toChannels[0].OutPolicySet)
assertCachedPolicyEqual(t, outPolicy1, toChannels[0].InPolicy)
}
runTest(pubKey1, pubKey2)
runTest(pubKey2, pubKey1)
}
func assertCachedPolicyEqual(t *testing.T, original *ChannelEdgePolicy,
cached *CachedEdgePolicy) {
require.Equal(t, original.ChannelID, cached.ChannelID)
require.Equal(t, original.MessageFlags, cached.MessageFlags)
require.Equal(t, original.ChannelFlags, cached.ChannelFlags)
require.Equal(t, original.TimeLockDelta, cached.TimeLockDelta)
require.Equal(t, original.MinHTLC, cached.MinHTLC)
require.Equal(t, original.MaxHTLC, cached.MaxHTLC)
require.Equal(t, original.FeeBaseMSat, cached.FeeBaseMSat)
require.Equal(
t, original.FeeProportionalMillionths,
cached.FeeProportionalMillionths,
)
require.Equal(
t,
route.Vertex(original.Node.PubKeyBytes),
cached.ToNodePubKey(),
)
require.Equal(t, original.Node.Features, cached.ToNodeFeatures)
}
+352 -110
View File
@@ -42,7 +42,10 @@ var (
_, _ = testSig.R.SetString("63724406601629180062774974542967536251589935445068131219452686511677818569431", 10)
_, _ = testSig.S.SetString("18801056069249825825291287104931333862866033135609736119018462340006816851118", 10)
testFeatures = lnwire.NewFeatureVector(nil, lnwire.Features)
testFeatures = lnwire.NewFeatureVector(
lnwire.NewRawFeatureVector(lnwire.GossipQueriesRequired),
lnwire.Features,
)
testPub = route.Vertex{2, 202, 4}
)
@@ -146,6 +149,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
if err := graph.AddLightningNode(node); err != nil {
t.Fatalf("unable to add node: %v", err)
}
assertNodeInCache(t, graph, node, testFeatures)
// Next, fetch the node from the database to ensure everything was
// serialized properly.
@@ -170,6 +174,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
if err := graph.DeleteLightningNode(testPub); err != nil {
t.Fatalf("unable to delete node; %v", err)
}
assertNodeNotInCache(t, graph, testPub)
// Finally, attempt to fetch the node again. This should fail as the
// node should have been deleted from the database.
@@ -200,6 +205,7 @@ func TestPartialNode(t *testing.T) {
if err := graph.AddLightningNode(node); err != nil {
t.Fatalf("unable to add node: %v", err)
}
assertNodeInCache(t, graph, node, nil)
// Next, fetch the node from the database to ensure everything was
// serialized properly.
@@ -232,6 +238,7 @@ func TestPartialNode(t *testing.T) {
if err := graph.DeleteLightningNode(testPub); err != nil {
t.Fatalf("unable to delete node: %v", err)
}
assertNodeNotInCache(t, graph, testPub)
// Finally, attempt to fetch the node again. This should fail as the
// node should have been deleted from the database.
@@ -390,6 +397,7 @@ func TestEdgeInsertionDeletion(t *testing.T) {
if err := graph.AddChannelEdge(&edgeInfo); err != nil {
t.Fatalf("unable to create channel edge: %v", err)
}
assertEdgeWithNoPoliciesInCache(t, graph, &edgeInfo)
// Ensure that both policies are returned as unknown (nil).
_, e1, e2, err := graph.FetchChannelEdgesByID(chanID)
@@ -405,6 +413,7 @@ func TestEdgeInsertionDeletion(t *testing.T) {
if err := graph.DeleteChannelEdges(false, chanID); err != nil {
t.Fatalf("unable to delete edge: %v", err)
}
assertNoEdge(t, graph, chanID)
// Ensure that any query attempts to lookup the delete channel edge are
// properly deleted.
@@ -544,6 +553,9 @@ func TestDisconnectBlockAtHeight(t *testing.T) {
if err := graph.AddChannelEdge(&edgeInfo3); err != nil {
t.Fatalf("unable to create channel edge: %v", err)
}
assertEdgeWithNoPoliciesInCache(t, graph, &edgeInfo)
assertEdgeWithNoPoliciesInCache(t, graph, &edgeInfo2)
assertEdgeWithNoPoliciesInCache(t, graph, &edgeInfo3)
// Call DisconnectBlockAtHeight, which should prune every channel
// that has a funding height of 'height' or greater.
@@ -551,6 +563,9 @@ func TestDisconnectBlockAtHeight(t *testing.T) {
if err != nil {
t.Fatalf("unable to prune %v", err)
}
assertNoEdge(t, graph, edgeInfo.ChannelID)
assertNoEdge(t, graph, edgeInfo2.ChannelID)
assertEdgeWithNoPoliciesInCache(t, graph, &edgeInfo3)
// The two edges should have been removed.
if len(removed) != 2 {
@@ -769,6 +784,7 @@ func TestEdgeInfoUpdates(t *testing.T) {
if err := graph.AddLightningNode(node1); err != nil {
t.Fatalf("unable to add node: %v", err)
}
assertNodeInCache(t, graph, node1, testFeatures)
node2, err := createTestVertex(graph.db)
if err != nil {
t.Fatalf("unable to create test node: %v", err)
@@ -776,6 +792,7 @@ func TestEdgeInfoUpdates(t *testing.T) {
if err := graph.AddLightningNode(node2); err != nil {
t.Fatalf("unable to add node: %v", err)
}
assertNodeInCache(t, graph, node2, testFeatures)
// Create an edge and add it to the db.
edgeInfo, edge1, edge2 := createChannelEdge(graph.db, node1, node2)
@@ -785,11 +802,13 @@ func TestEdgeInfoUpdates(t *testing.T) {
if err := graph.UpdateEdgePolicy(edge1); err != ErrEdgeNotFound {
t.Fatalf("expected ErrEdgeNotFound, got: %v", err)
}
require.Len(t, graph.graphCache.nodeChannels, 0)
// Add the edge info.
if err := graph.AddChannelEdge(edgeInfo); err != nil {
t.Fatalf("unable to create channel edge: %v", err)
}
assertEdgeWithNoPoliciesInCache(t, graph, edgeInfo)
chanID := edgeInfo.ChannelID
outpoint := edgeInfo.ChannelPoint
@@ -799,9 +818,11 @@ func TestEdgeInfoUpdates(t *testing.T) {
if err := graph.UpdateEdgePolicy(edge1); err != nil {
t.Fatalf("unable to update edge: %v", err)
}
assertEdgeWithPolicyInCache(t, graph, edgeInfo, edge1, true)
if err := graph.UpdateEdgePolicy(edge2); err != nil {
t.Fatalf("unable to update edge: %v", err)
}
assertEdgeWithPolicyInCache(t, graph, edgeInfo, edge2, false)
// Check for existence of the edge within the database, it should be
// found.
@@ -856,6 +877,191 @@ func TestEdgeInfoUpdates(t *testing.T) {
assertEdgeInfoEqual(t, dbEdgeInfo, edgeInfo)
}
func assertNodeInCache(t *testing.T, g *ChannelGraph, n *LightningNode,
expectedFeatures *lnwire.FeatureVector) {
// Let's check the internal view first.
require.Equal(
t, expectedFeatures, g.graphCache.nodeFeatures[n.PubKeyBytes],
)
// The external view should reflect this as well. Except when we expect
// the features to be nil internally, we return an empty feature vector
// on the public interface instead.
if expectedFeatures == nil {
expectedFeatures = lnwire.EmptyFeatureVector()
}
features := g.graphCache.GetFeatures(n.PubKeyBytes)
require.Equal(t, expectedFeatures, features)
}
func assertNodeNotInCache(t *testing.T, g *ChannelGraph, n route.Vertex) {
_, ok := g.graphCache.nodeFeatures[n]
require.False(t, ok)
_, ok = g.graphCache.nodeChannels[n]
require.False(t, ok)
// We should get the default features for this node.
features := g.graphCache.GetFeatures(n)
require.Equal(t, lnwire.EmptyFeatureVector(), features)
}
func assertEdgeWithNoPoliciesInCache(t *testing.T, g *ChannelGraph,
e *ChannelEdgeInfo) {
// Let's check the internal view first.
require.NotEmpty(t, g.graphCache.nodeChannels[e.NodeKey1Bytes])
require.NotEmpty(t, g.graphCache.nodeChannels[e.NodeKey2Bytes])
expectedNode1Channel := &DirectedChannel{
ChannelID: e.ChannelID,
IsNode1: true,
OtherNode: e.NodeKey2Bytes,
Capacity: e.Capacity,
OutPolicySet: false,
InPolicy: nil,
}
require.Contains(
t, g.graphCache.nodeChannels[e.NodeKey1Bytes], e.ChannelID,
)
require.Equal(
t, expectedNode1Channel,
g.graphCache.nodeChannels[e.NodeKey1Bytes][e.ChannelID],
)
expectedNode2Channel := &DirectedChannel{
ChannelID: e.ChannelID,
IsNode1: false,
OtherNode: e.NodeKey1Bytes,
Capacity: e.Capacity,
OutPolicySet: false,
InPolicy: nil,
}
require.Contains(
t, g.graphCache.nodeChannels[e.NodeKey2Bytes], e.ChannelID,
)
require.Equal(
t, expectedNode2Channel,
g.graphCache.nodeChannels[e.NodeKey2Bytes][e.ChannelID],
)
// The external view should reflect this as well.
var foundChannel *DirectedChannel
err := g.graphCache.ForEachChannel(
e.NodeKey1Bytes, func(c *DirectedChannel) error {
if c.ChannelID == e.ChannelID {
foundChannel = c
}
return nil
},
)
require.NoError(t, err)
require.NotNil(t, foundChannel)
require.Equal(t, expectedNode1Channel, foundChannel)
err = g.graphCache.ForEachChannel(
e.NodeKey2Bytes, func(c *DirectedChannel) error {
if c.ChannelID == e.ChannelID {
foundChannel = c
}
return nil
},
)
require.NoError(t, err)
require.NotNil(t, foundChannel)
require.Equal(t, expectedNode2Channel, foundChannel)
}
func assertNoEdge(t *testing.T, g *ChannelGraph, chanID uint64) {
// Make sure no channel in the cache has the given channel ID. If there
// are no channels at all, that is fine as well.
for _, channels := range g.graphCache.nodeChannels {
for _, channel := range channels {
require.NotEqual(t, channel.ChannelID, chanID)
}
}
}
func assertEdgeWithPolicyInCache(t *testing.T, g *ChannelGraph,
e *ChannelEdgeInfo, p *ChannelEdgePolicy, policy1 bool) {
// Check the internal state first.
c1, ok := g.graphCache.nodeChannels[e.NodeKey1Bytes][e.ChannelID]
require.True(t, ok)
if policy1 {
require.True(t, c1.OutPolicySet)
} else {
require.NotNil(t, c1.InPolicy)
require.Equal(
t, p.FeeProportionalMillionths,
c1.InPolicy.FeeProportionalMillionths,
)
}
c2, ok := g.graphCache.nodeChannels[e.NodeKey2Bytes][e.ChannelID]
require.True(t, ok)
if policy1 {
require.NotNil(t, c2.InPolicy)
require.Equal(
t, p.FeeProportionalMillionths,
c2.InPolicy.FeeProportionalMillionths,
)
} else {
require.True(t, c2.OutPolicySet)
}
// Now for both nodes make sure that the external view is also correct.
var (
c1Ext *DirectedChannel
c2Ext *DirectedChannel
)
require.NoError(t, g.graphCache.ForEachChannel(
e.NodeKey1Bytes, func(c *DirectedChannel) error {
c1Ext = c
return nil
},
))
require.NoError(t, g.graphCache.ForEachChannel(
e.NodeKey2Bytes, func(c *DirectedChannel) error {
c2Ext = c
return nil
},
))
// Only compare the fields that are actually copied, then compare the
// values of the functions separately.
require.Equal(t, c1, c1Ext.DeepCopy())
require.Equal(t, c2, c2Ext.DeepCopy())
if policy1 {
require.Equal(
t, p.FeeProportionalMillionths,
c2Ext.InPolicy.FeeProportionalMillionths,
)
require.Equal(
t, route.Vertex(e.NodeKey2Bytes),
c2Ext.InPolicy.ToNodePubKey(),
)
require.Equal(t, testFeatures, c2Ext.InPolicy.ToNodeFeatures)
} else {
require.Equal(
t, p.FeeProportionalMillionths,
c1Ext.InPolicy.FeeProportionalMillionths,
)
require.Equal(
t, route.Vertex(e.NodeKey1Bytes),
c1Ext.InPolicy.ToNodePubKey(),
)
require.Equal(t, testFeatures, c1Ext.InPolicy.ToNodeFeatures)
}
}
func randEdgePolicy(chanID uint64, db kvdb.Backend) *ChannelEdgePolicy {
update := prand.Int63()
@@ -890,106 +1096,10 @@ func TestGraphTraversal(t *testing.T) {
// We'd like to test some of the graph traversal capabilities within
// the DB, so we'll create a series of fake nodes to insert into the
// graph.
// graph. And we'll create 5 channels between each node pair.
const numNodes = 20
nodes := make([]*LightningNode, numNodes)
nodeIndex := map[string]struct{}{}
for i := 0; i < numNodes; i++ {
node, err := createTestVertex(graph.db)
if err != nil {
t.Fatalf("unable to create node: %v", err)
}
nodes[i] = node
nodeIndex[node.Alias] = struct{}{}
}
// Add each of the nodes into the graph, they should be inserted
// without error.
for _, node := range nodes {
if err := graph.AddLightningNode(node); err != nil {
t.Fatalf("unable to add node: %v", err)
}
}
// Iterate over each node as returned by the graph, if all nodes are
// reached, then the map created above should be empty.
err = graph.ForEachNode(func(_ kvdb.RTx, node *LightningNode) error {
delete(nodeIndex, node.Alias)
return nil
})
if err != nil {
t.Fatalf("for each failure: %v", err)
}
if len(nodeIndex) != 0 {
t.Fatalf("all nodes not reached within ForEach")
}
// Determine which node is "smaller", we'll need this in order to
// properly create the edges for the graph.
var firstNode, secondNode *LightningNode
if bytes.Compare(nodes[0].PubKeyBytes[:], nodes[1].PubKeyBytes[:]) == -1 {
firstNode = nodes[0]
secondNode = nodes[1]
} else {
firstNode = nodes[0]
secondNode = nodes[1]
}
// Create 5 channels between the first two nodes we generated above.
const numChannels = 5
chanIndex := map[uint64]struct{}{}
for i := 0; i < numChannels; i++ {
txHash := sha256.Sum256([]byte{byte(i)})
chanID := uint64(i + 1)
op := wire.OutPoint{
Hash: txHash,
Index: 0,
}
edgeInfo := ChannelEdgeInfo{
ChannelID: chanID,
ChainHash: key,
AuthProof: &ChannelAuthProof{
NodeSig1Bytes: testSig.Serialize(),
NodeSig2Bytes: testSig.Serialize(),
BitcoinSig1Bytes: testSig.Serialize(),
BitcoinSig2Bytes: testSig.Serialize(),
},
ChannelPoint: op,
Capacity: 1000,
}
copy(edgeInfo.NodeKey1Bytes[:], nodes[0].PubKeyBytes[:])
copy(edgeInfo.NodeKey2Bytes[:], nodes[1].PubKeyBytes[:])
copy(edgeInfo.BitcoinKey1Bytes[:], nodes[0].PubKeyBytes[:])
copy(edgeInfo.BitcoinKey2Bytes[:], nodes[1].PubKeyBytes[:])
err := graph.AddChannelEdge(&edgeInfo)
if err != nil {
t.Fatalf("unable to add node: %v", err)
}
// Create and add an edge with random data that points from
// node1 -> node2.
edge := randEdgePolicy(chanID, graph.db)
edge.ChannelFlags = 0
edge.Node = secondNode
edge.SigBytes = testSig.Serialize()
if err := graph.UpdateEdgePolicy(edge); err != nil {
t.Fatalf("unable to update edge: %v", err)
}
// Create another random edge that points from node2 -> node1
// this time.
edge = randEdgePolicy(chanID, graph.db)
edge.ChannelFlags = 1
edge.Node = firstNode
edge.SigBytes = testSig.Serialize()
if err := graph.UpdateEdgePolicy(edge); err != nil {
t.Fatalf("unable to update edge: %v", err)
}
chanIndex[chanID] = struct{}{}
}
chanIndex, nodeList := fillTestGraph(t, graph, numNodes, numChannels)
// Iterate through all the known channels within the graph DB, once
// again if the map is empty that indicates that all edges have
@@ -1000,16 +1110,13 @@ func TestGraphTraversal(t *testing.T) {
delete(chanIndex, ei.ChannelID)
return nil
})
if err != nil {
t.Fatalf("for each failure: %v", err)
}
if len(chanIndex) != 0 {
t.Fatalf("all edges not reached within ForEach")
}
require.NoError(t, err)
require.Len(t, chanIndex, 0)
// Finally, we want to test the ability to iterate over all the
// outgoing channels for a particular node.
numNodeChans := 0
firstNode, secondNode := nodeList[0], nodeList[1]
err = firstNode.ForEachChannel(nil, func(_ kvdb.RTx, _ *ChannelEdgeInfo,
outEdge, inEdge *ChannelEdgePolicy) error {
@@ -1034,13 +1141,148 @@ func TestGraphTraversal(t *testing.T) {
numNodeChans++
return nil
})
if err != nil {
t.Fatalf("for each failure: %v", err)
require.NoError(t, err)
require.Equal(t, numChannels, numNodeChans)
}
func TestGraphCacheTraversal(t *testing.T) {
t.Parallel()
graph, cleanUp, err := MakeTestGraph()
defer cleanUp()
require.NoError(t, err)
// We'd like to test some of the graph traversal capabilities within
// the DB, so we'll create a series of fake nodes to insert into the
// graph. And we'll create 5 channels between each node pair.
const numNodes = 20
const numChannels = 5
chanIndex, nodeList := fillTestGraph(t, graph, numNodes, numChannels)
// Iterate through all the known channels within the graph DB, once
// again if the map is empty that indicates that all edges have
// properly been reached.
numNodeChans := 0
for _, node := range nodeList {
err = graph.graphCache.ForEachChannel(
node.PubKeyBytes, func(d *DirectedChannel) error {
delete(chanIndex, d.ChannelID)
if !d.OutPolicySet || d.InPolicy == nil {
return fmt.Errorf("channel policy not " +
"present")
}
// The incoming edge should also indicate that
// it's pointing to the origin node.
inPolicyNodeKey := d.InPolicy.ToNodePubKey()
if !bytes.Equal(
inPolicyNodeKey[:], node.PubKeyBytes[:],
) {
return fmt.Errorf("wrong outgoing edge")
}
numNodeChans++
return nil
},
)
require.NoError(t, err)
}
if numNodeChans != numChannels {
t.Fatalf("all edges for node not reached within ForEach: "+
"expected %v, got %v", numChannels, numNodeChans)
require.Len(t, chanIndex, 0)
// We count the channels for both nodes, so there should be double the
// amount now. Except for the very last node, that doesn't have any
// channels to make the loop easier in fillTestGraph().
require.Equal(t, numChannels*2*(numNodes-1), numNodeChans)
}
func fillTestGraph(t *testing.T, graph *ChannelGraph, numNodes,
numChannels int) (map[uint64]struct{}, []*LightningNode) {
nodes := make([]*LightningNode, numNodes)
nodeIndex := map[string]struct{}{}
for i := 0; i < numNodes; i++ {
node, err := createTestVertex(graph.db)
require.NoError(t, err)
nodes[i] = node
nodeIndex[node.Alias] = struct{}{}
}
// Add each of the nodes into the graph, they should be inserted
// without error.
for _, node := range nodes {
require.NoError(t, graph.AddLightningNode(node))
}
// Iterate over each node as returned by the graph, if all nodes are
// reached, then the map created above should be empty.
err := graph.ForEachNode(func(_ kvdb.RTx, node *LightningNode) error {
delete(nodeIndex, node.Alias)
return nil
})
require.NoError(t, err)
require.Len(t, nodeIndex, 0)
// Create a number of channels between each of the node pairs generated
// above. This will result in numChannels*(numNodes-1) channels.
chanIndex := map[uint64]struct{}{}
for n := 0; n < numNodes-1; n++ {
node1 := nodes[n]
node2 := nodes[n+1]
if bytes.Compare(node1.PubKeyBytes[:], node2.PubKeyBytes[:]) == -1 {
node1, node2 = node2, node1
}
for i := 0; i < numChannels; i++ {
txHash := sha256.Sum256([]byte{byte(i)})
chanID := uint64((n << 4) + i + 1)
op := wire.OutPoint{
Hash: txHash,
Index: 0,
}
edgeInfo := ChannelEdgeInfo{
ChannelID: chanID,
ChainHash: key,
AuthProof: &ChannelAuthProof{
NodeSig1Bytes: testSig.Serialize(),
NodeSig2Bytes: testSig.Serialize(),
BitcoinSig1Bytes: testSig.Serialize(),
BitcoinSig2Bytes: testSig.Serialize(),
},
ChannelPoint: op,
Capacity: 1000,
}
copy(edgeInfo.NodeKey1Bytes[:], node1.PubKeyBytes[:])
copy(edgeInfo.NodeKey2Bytes[:], node2.PubKeyBytes[:])
copy(edgeInfo.BitcoinKey1Bytes[:], node1.PubKeyBytes[:])
copy(edgeInfo.BitcoinKey2Bytes[:], node2.PubKeyBytes[:])
err := graph.AddChannelEdge(&edgeInfo)
require.NoError(t, err)
// Create and add an edge with random data that points
// from node1 -> node2.
edge := randEdgePolicy(chanID, graph.db)
edge.ChannelFlags = 0
edge.Node = node2
edge.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(edge))
// Create another random edge that points from
// node2 -> node1 this time.
edge = randEdgePolicy(chanID, graph.db)
edge.ChannelFlags = 1
edge.Node = node1
edge.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(edge))
chanIndex[chanID] = struct{}{}
}
}
return chanIndex, nodes
}
func assertPruneTip(t *testing.T, graph *ChannelGraph, blockHash *chainhash.Hash,