diff --git a/graph/builder.go b/graph/builder.go index 2cf4d1f34..9d3baebfe 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -1276,7 +1276,7 @@ func (b *Builder) ForAllOutgoingChannels(ctx context.Context, reset func()) error { return b.cfg.Graph.ForEachNodeChannel( - ctx, b.cfg.SelfNode, + ctx, lnwire.GossipVersion1, b.cfg.SelfNode, func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy, _ *models.ChannelEdgePolicy) error { diff --git a/graph/db/graph.go b/graph/db/graph.go index d7398e045..baee3eccb 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -597,11 +597,12 @@ func (c *ChannelGraph) ForEachSourceNodeChannel(ctx context.Context, // ForEachNodeChannel iterates through all channels of the given node. func (c *ChannelGraph) ForEachNodeChannel(ctx context.Context, - nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, + v lnwire.GossipVersion, nodePub route.Vertex, + cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, reset func()) error { - return c.db.ForEachNodeChannel(ctx, nodePub, cb, reset) + return c.db.ForEachNodeChannel(ctx, v, nodePub, cb, reset) } // ForEachNode iterates through all stored vertices/nodes in the graph. @@ -905,6 +906,15 @@ func (c *VersionedGraph) HasChannelEdge(chanID uint64) (bool, bool, error) { return c.db.HasChannelEdge(c.v, chanID) } +// ForEachNodeChannel iterates through all channels of the given node. +func (c *VersionedGraph) ForEachNodeChannel(ctx context.Context, + nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, + *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy) error, reset func()) error { + + return c.db.ForEachNodeChannel(ctx, c.v, nodePub, cb, reset) +} + // ForEachChannel iterates through all channel edges stored within the graph. func (c *VersionedGraph) ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 3e71f5e6e..2edfc064a 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -150,6 +150,10 @@ var versionedTests = []versionedTest{ name: "edge policy crud", test: testEdgePolicyCRUD, }, + { + name: "incomplete channel policies", + test: testIncompleteChannelPolicies, + }, { name: "partial node", test: testPartialNode, @@ -1753,7 +1757,7 @@ func TestGraphTraversal(t *testing.T) { numNodeChans := 0 firstNode, secondNode := nodeList[0], nodeList[1] err = graph.ForEachNodeChannel( - ctx, firstNode.PubKeyBytes, + ctx, lnwire.GossipVersion1, firstNode.PubKeyBytes, func(_ *models.ChannelEdgeInfo, outEdge, inEdge *models.ChannelEdgePolicy) error { @@ -3810,32 +3814,26 @@ func TestFetchChanInfos(t *testing.T) { } } -// TestIncompleteChannelPolicies tests that a channel that only has a policy +// testIncompleteChannelPolicies tests that a channel that only has a policy // specified on one end is properly returned in ForEachChannel calls from // both sides. -func TestIncompleteChannelPolicies(t *testing.T) { +func testIncompleteChannelPolicies(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) // Create two nodes. - node1 := createTestVertex(t, lnwire.GossipVersion1) - if err := graph.AddNode(ctx, node1); err != nil { - t.Fatalf("unable to add node: %v", err) - } - node2 := createTestVertex(t, lnwire.GossipVersion1) - if err := graph.AddNode(ctx, node2); err != nil { - t.Fatalf("unable to add node: %v", err) - } + node1 := createTestVertex(t, v) + require.NoError(t, graph.AddNode(ctx, node1)) + node2 := createTestVertex(t, v) + require.NoError(t, graph.AddNode(ctx, node2)) channel, chanID := createEdge( - lnwire.GossipVersion1, uint32(0), 0, 0, 0, node1, node2, + v, uint32(0), 0, 0, 0, node1, node2, ) - if err := graph.AddChannelEdge(ctx, channel); err != nil { - t.Fatalf("unable to create channel edge: %v", err) - } + require.NoError(t, graph.AddChannelEdge(ctx, channel)) // Ensure that channel is reported with unknown policies. checkPolicies := func(node *models.Node, expectedIn, @@ -3847,21 +3845,8 @@ func TestIncompleteChannelPolicies(t *testing.T) { func(_ *models.ChannelEdgeInfo, outEdge, inEdge *models.ChannelEdgePolicy) error { - if !expectedOut && outEdge != nil { - t.Fatalf("Expected no outgoing policy") - } - - if expectedOut && outEdge == nil { - t.Fatalf("Expected an outgoing policy") - } - - if !expectedIn && inEdge != nil { - t.Fatalf("Expected no incoming policy") - } - - if expectedIn && inEdge == nil { - t.Fatalf("Expected an incoming policy") - } + require.Equal(t, expectedOut, outEdge != nil) + require.Equal(t, expectedIn, inEdge != nil) calls++ @@ -3874,34 +3859,30 @@ func TestIncompleteChannelPolicies(t *testing.T) { checkPolicies(node2, false, false) + newTestEdgePolicy := func(isNode1 bool, + toNode route.Vertex) *models.ChannelEdgePolicy { + + policy := newEdgePolicy( + v, chanID.ToUint64(), nextUpdateTime().Unix(), isNode1, + ) + policy.ToNode = toNode + policy.SigBytes = testSig.Serialize() + + return policy + } + // Only create an edge policy for node1 and leave the policy for node2 // unknown. - updateTime := time.Unix(1234, 0) - - edgePolicy := newEdgePolicy( - lnwire.GossipVersion1, chanID.ToUint64(), updateTime.Unix(), - true, - ) - edgePolicy.ToNode = node2.PubKeyBytes - edgePolicy.SigBytes = testSig.Serialize() - if err := graph.UpdateEdgePolicy(ctx, edgePolicy); err != nil { - t.Fatalf("unable to update edge: %v", err) - } + edgePolicy := newTestEdgePolicy(true, node2.PubKeyBytes) + require.NoError(t, graph.UpdateEdgePolicy(ctx, edgePolicy)) checkPolicies(node1, false, true) checkPolicies(node2, true, false) // Create second policy and assert that both policies are reported // as present. - edgePolicy = newEdgePolicy( - lnwire.GossipVersion1, chanID.ToUint64(), updateTime.Unix(), - false, - ) - edgePolicy.ToNode = node1.PubKeyBytes - edgePolicy.SigBytes = testSig.Serialize() - if err := graph.UpdateEdgePolicy(ctx, edgePolicy); err != nil { - t.Fatalf("unable to update edge: %v", err) - } + edgePolicy = newTestEdgePolicy(false, node1.PubKeyBytes) + require.NoError(t, graph.UpdateEdgePolicy(ctx, edgePolicy)) checkPolicies(node1, true, true) checkPolicies(node2, true, true) @@ -5024,7 +5005,9 @@ func BenchmarkForEachChannel(b *testing.B) { return nil } - err := graph.ForEachNodeChannel(ctx, n, cb, func() {}) + err := graph.ForEachNodeChannel( + ctx, lnwire.GossipVersion1, n, cb, func() {}, + ) require.NoError(b, err) } } diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index c650bd23b..0936386be 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -73,8 +73,9 @@ type Store interface { //nolint:interfacebloat // to the caller. // // Unknown policies are passed into the callback as nil values. - ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, - cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + ForEachNodeChannel(ctx context.Context, v lnwire.GossipVersion, + nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, + *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, reset func()) error // ForEachNodeCached is similar to forEachNode, but it returns diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 03eaf02c7..22c87f0ee 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -3690,10 +3690,15 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend, // halted with the error propagated back up to the caller. // // Unknown policies are passed into the callback as nil values. -func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex, +func (c *KVStore) ForEachNodeChannel(_ context.Context, + v lnwire.GossipVersion, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, reset func()) error { + if v != lnwire.GossipVersion1 { + return ErrVersionNotSupportedForKVDB + } + return nodeTraversal( nil, nodePub[:], c.db, func(_ kvdb.RTx, info *models.ChannelEdgeInfo, policy, diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index b66e2fe49..44e2fd75b 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -913,7 +913,7 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, } return forEachNodeChannel( - ctx, db, s.cfg, nodeID, + ctx, db, s.cfg, lnwire.GossipVersion1, nodeID, func(info *models.ChannelEdgeInfo, outPolicy *models.ChannelEdgePolicy, _ *models.ChannelEdgePolicy) error { @@ -1027,14 +1027,15 @@ func (s *SQLStore) ForEachNodeCacheable(ctx context.Context, // Unknown policies are passed into the callback as nil values. // // NOTE: part of the Store interface. -func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, +func (s *SQLStore) ForEachNodeChannel(ctx context.Context, + v lnwire.GossipVersion, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, reset func()) error { return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { dbNode, err := db.GetNodeByPubKey( ctx, sqlc.GetNodeByPubKeyParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), PubKey: nodePub[:], }, ) @@ -1044,7 +1045,7 @@ func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, return fmt.Errorf("unable to fetch node: %w", err) } - return forEachNodeChannel(ctx, db, s.cfg, dbNode.ID, cb) + return forEachNodeChannel(ctx, db, s.cfg, v, dbNode.ID, cb) }, reset) } @@ -3524,14 +3525,15 @@ func forEachNodeCacheable(ctx context.Context, cfg *sqldb.QueryConfig, // edge information, the outgoing policy and the incoming policy for the // channel and node combo. func forEachNodeChannel(ctx context.Context, db SQLQueries, - cfg *SQLStoreConfig, id int64, cb func(*models.ChannelEdgeInfo, + cfg *SQLStoreConfig, v lnwire.GossipVersion, id int64, + cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { - // Get all the V1 channels for this node. + // Get all the channels for this node. rows, err := db.ListChannelsByNodeID( ctx, sqlc.ListChannelsByNodeIDParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), NodeID1: id, }, ) diff --git a/server.go b/server.go index 7ba74176a..ad97261ed 100644 --- a/server.go +++ b/server.go @@ -1131,7 +1131,8 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, *models.ChannelEdgePolicy) error, reset func()) error { - return s.graphDB.ForEachNodeChannel(ctx, selfVertex, + return s.v1Graph.ForEachNodeChannel( + ctx, selfVertex, func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy, _ *models.ChannelEdgePolicy) error {