graph/db: thread context to AddLightningNode

This commit is contained in:
Elle Mouton committed 2025-06-17 19:20:27 +02:00
1 parent c1740c14ba
commit d1cfb47428
15 files changed
+126 -81

No files matched your search

+8 -3
View File
@@ -422,7 +422,9 @@ func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey,
AuthSigBytes: testSig.Serialize(), AuthSigBytes: testSig.Serialize(),
} }
graphNode.AddPubKey(pub) graphNode.AddPubKey(pub)
err := d.db.AddLightningNode(graphNode) err := d.db.AddLightningNode(
context.Background(), graphNode,
)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -450,7 +452,9 @@ func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey,
AuthSigBytes: testSig.Serialize(), AuthSigBytes: testSig.Serialize(),
} }
dbNode.AddPubKey(nodeKey) dbNode.AddPubKey(nodeKey)
if err := d.db.AddLightningNode(dbNode); err != nil { if err := d.db.AddLightningNode(
context.Background(), dbNode,
); err != nil {
return nil, err return nil, err
} }
@@ -554,7 +558,8 @@ func (d *testDBGraph) addRandNode() (*btcec.PublicKey, error) {
AuthSigBytes: testSig.Serialize(), AuthSigBytes: testSig.Serialize(),
} }
dbNode.AddPubKey(nodeKey) dbNode.AddPubKey(nodeKey)
if err := d.db.AddLightningNode(dbNode); err != nil { err = d.db.AddLightningNode(context.Background(), dbNode)
if err != nil {
return nil, err return nil, err
} }
+4 -2
View File
@@ -2022,7 +2022,7 @@ func (d *AuthenticatedGossiper) fetchPKScript(chanID *lnwire.ShortChannelID) (
// addNode processes the given node announcement, and adds it to our channel // addNode processes the given node announcement, and adds it to our channel
// graph. // graph.
func (d *AuthenticatedGossiper) addNode(_ context.Context, func (d *AuthenticatedGossiper) addNode(ctx context.Context,
msg *lnwire.NodeAnnouncement, op ...batch.SchedulerOption) error { msg *lnwire.NodeAnnouncement, op ...batch.SchedulerOption) error {
if err := netann.ValidateNodeAnn(msg); err != nil { if err := netann.ValidateNodeAnn(msg); err != nil {
@@ -2030,7 +2030,9 @@ func (d *AuthenticatedGossiper) addNode(_ context.Context,
err) err)
} }
return d.cfg.Graph.AddNode(models.NodeFromWireAnnouncement(msg), op...) return d.cfg.Graph.AddNode(
ctx, models.NodeFromWireAnnouncement(msg), op...,
)
} }
// isPremature decides whether a given network message has a block height+delta // isPremature decides whether a given network message has a block height+delta
+1 -1
View File
@@ -109,7 +109,7 @@ func newMockRouter(t *testing.T, height uint32) *mockGraphSource {
var _ graph.ChannelGraphSource = (*mockGraphSource)(nil) var _ graph.ChannelGraphSource = (*mockGraphSource)(nil)
func (r *mockGraphSource) AddNode(node *models.LightningNode, func (r *mockGraphSource) AddNode(_ context.Context, node *models.LightningNode,
_ ...batch.SchedulerOption) error { _ ...batch.SchedulerOption) error {
r.mu.Lock() r.mu.Lock()
+5 -4
View File
@@ -1,6 +1,7 @@
package graph package graph
import ( import (
"context"
"fmt" "fmt"
"sync" "sync"
"sync/atomic" "sync/atomic"
@@ -972,10 +973,10 @@ func (b *Builder) ApplyChannelUpdate(msg *lnwire.ChannelUpdate1) bool {
// be ignored. // be ignored.
// //
// NOTE: This method is part of the ChannelGraphSource interface. // NOTE: This method is part of the ChannelGraphSource interface.
func (b *Builder) AddNode(node *models.LightningNode, func (b *Builder) AddNode(ctx context.Context, node *models.LightningNode,
op ...batch.SchedulerOption) error { op ...batch.SchedulerOption) error {
err := b.addNode(node, op...) err := b.addNode(ctx, node, op...)
if err != nil { if err != nil {
logNetworkMsgProcessError(err) logNetworkMsgProcessError(err)
@@ -989,7 +990,7 @@ func (b *Builder) AddNode(node *models.LightningNode,
// currently have persisted in the graph, and then adds it to the graph. If we // currently have persisted in the graph, and then adds it to the graph. If we
// already know about the node, then we only update our DB if the new update // already know about the node, then we only update our DB if the new update
// has a newer timestamp than the last one we received. // has a newer timestamp than the last one we received.
func (b *Builder) addNode(node *models.LightningNode, func (b *Builder) addNode(ctx context.Context, node *models.LightningNode,
op ...batch.SchedulerOption) error { op ...batch.SchedulerOption) error {
// Before we add the node to the database, we'll check to see if the // Before we add the node to the database, we'll check to see if the
@@ -1000,7 +1001,7 @@ func (b *Builder) addNode(node *models.LightningNode,
return err return err
} }
if err := b.cfg.Graph.AddLightningNode(node, op...); err != nil { if err := b.cfg.Graph.AddLightningNode(ctx, node, op...); err != nil {
return errors.Errorf("unable to add node %x to the "+ return errors.Errorf("unable to add node %x to the "+
"graph: %v", node.PubKeyBytes, err) "graph: %v", node.PubKeyBytes, err)
} }
+9 -4
View File
@@ -2,6 +2,7 @@ package graph
import ( import (
"bytes" "bytes"
"context"
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
@@ -106,7 +107,7 @@ func TestIgnoreNodeAnnouncement(t *testing.T) {
} }
copy(node.PubKeyBytes[:], pub.SerializeCompressed()) copy(node.PubKeyBytes[:], pub.SerializeCompressed())
err := ctx.builder.AddNode(node) err := ctx.builder.AddNode(context.Background(), node)
if !IsError(err, ErrIgnored) { if !IsError(err, ErrIgnored) {
t.Fatalf("expected to get ErrIgnore, instead got: %v", err) t.Fatalf("expected to get ErrIgnore, instead got: %v", err)
} }
@@ -1068,7 +1069,7 @@ func TestIsStaleNode(t *testing.T) {
Features: testFeatures, Features: testFeatures,
} }
copy(n1.PubKeyBytes[:], priv1.PubKey().SerializeCompressed()) copy(n1.PubKeyBytes[:], priv1.PubKey().SerializeCompressed())
if err := ctx.builder.AddNode(n1); err != nil { if err := ctx.builder.AddNode(context.Background(), n1); err != nil {
t.Fatalf("could not add node: %v", err) t.Fatalf("could not add node: %v", err)
} }
@@ -1328,6 +1329,8 @@ func createTestCtxFromFile(t *testing.T,
func parseTestGraph(t *testing.T, useCache bool, path string) ( func parseTestGraph(t *testing.T, useCache bool, path string) (
*testGraphInstance, error) { *testGraphInstance, error) {
ctx := context.Background()
graphJSON, err := os.ReadFile(path) graphJSON, err := os.ReadFile(path)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -1434,7 +1437,7 @@ func parseTestGraph(t *testing.T, useCache bool, path string) (
// With the node fully parsed, add it as a vertex within the // With the node fully parsed, add it as a vertex within the
// graph. // graph.
if err := graph.AddLightningNode(dbNode); err != nil { if err := graph.AddLightningNode(ctx, dbNode); err != nil {
return nil, err return nil, err
} }
} }
@@ -1715,6 +1718,8 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
testChannels []*testChannel, source string) (*testGraphInstance, testChannels []*testChannel, source string) (*testGraphInstance,
error) { error) {
ctx := context.Background()
// We'll use this fake address for the IP address of all the nodes in // We'll use this fake address for the IP address of all the nodes in
// our tests. This value isn't needed for path finding so it doesn't // our tests. This value isn't needed for path finding so it doesn't
// need to be unique. // need to be unique.
@@ -1763,7 +1768,7 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
// With the node fully parsed, add it as a vertex within the // With the node fully parsed, add it as a vertex within the
// graph. // graph.
if err := graph.AddLightningNode(dbNode); err != nil { if err := graph.AddLightningNode(ctx, dbNode); err != nil {
return nil, err return nil, err
} }
+4 -3
View File
@@ -1,6 +1,7 @@
package graphdb package graphdb
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
"sync" "sync"
@@ -260,10 +261,10 @@ func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex,
// information. Note that this method is expected to only be called to update an // information. Note that this method is expected to only be called to update an
// already present node from a node announcement, or to insert a node found in a // already present node from a node announcement, or to insert a node found in a
// channel update. // channel update.
func (c *ChannelGraph) AddLightningNode(node *models.LightningNode, func (c *ChannelGraph) AddLightningNode(ctx context.Context,
op ...batch.SchedulerOption) error { node *models.LightningNode, op ...batch.SchedulerOption) error {
err := c.V1Store.AddLightningNode(node, op...) err := c.V1Store.AddLightningNode(ctx, node, op...)
if err != nil { if err != nil {
return err return err
} }
+66 -43
View File
@@ -2,6 +2,7 @@ package graphdb
import ( import (
"bytes" "bytes"
"context"
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"errors" "errors"
@@ -99,6 +100,7 @@ func createTestVertex(t testing.TB) *models.LightningNode {
func TestNodeInsertionAndDeletion(t *testing.T) { func TestNodeInsertionAndDeletion(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraphNew(t) graph := MakeTestGraphNew(t)
@@ -123,7 +125,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
// First, insert the node into the graph DB. This should succeed // First, insert the node into the graph DB. This should succeed
// without any errors. // without any errors.
node := nodeWithAddrs(testAddrs) node := nodeWithAddrs(testAddrs)
if err := graph.AddLightningNode(node); err != nil { if err := graph.AddLightningNode(ctx, node); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
assertNodeInCache(t, graph, node, testFeatures) assertNodeInCache(t, graph, node, testFeatures)
@@ -187,7 +189,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
// Add the node without any addresses. // Add the node without any addresses.
node = nodeWithAddrs(nil) node = nodeWithAddrs(nil)
require.NoError(t, graph.AddLightningNode(node)) require.NoError(t, graph.AddLightningNode(ctx, node))
// Fetch the node and assert the empty addresses. // Fetch the node and assert the empty addresses.
dbNode, err = graph.FetchLightningNode(testPub) dbNode, err = graph.FetchLightningNode(testPub)
@@ -214,7 +216,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
testOpaqueAddr, testOpaqueAddr,
} }
node = nodeWithAddrs(expAddrs) node = nodeWithAddrs(expAddrs)
require.NoError(t, graph.AddLightningNode(node)) require.NoError(t, graph.AddLightningNode(ctx, node))
// Fetch the node and assert the updated addresses. // Fetch the node and assert the updated addresses.
dbNode, err = graph.FetchLightningNode(testPub) dbNode, err = graph.FetchLightningNode(testPub)
@@ -235,7 +237,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
testIPV6Addr, testIPV6Addr,
} }
node = nodeWithAddrs(expAddrs) node = nodeWithAddrs(expAddrs)
require.NoError(t, graph.AddLightningNode(node)) require.NoError(t, graph.AddLightningNode(ctx, node))
// Fetch the node and assert the updated addresses. // Fetch the node and assert the updated addresses.
dbNode, err = graph.FetchLightningNode(testPub) dbNode, err = graph.FetchLightningNode(testPub)
@@ -248,7 +250,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
testOnionV3Addr, testOnionV3Addr,
} }
node = nodeWithAddrs(expAddrs) node = nodeWithAddrs(expAddrs)
require.NoError(t, graph.AddLightningNode(node)) require.NoError(t, graph.AddLightningNode(ctx, node))
// Fetch the node and assert the updated addresses. // Fetch the node and assert the updated addresses.
dbNode, err = graph.FetchLightningNode(testPub) dbNode, err = graph.FetchLightningNode(testPub)
@@ -327,6 +329,7 @@ func TestPartialNode(t *testing.T) {
// TestAliasLookup tests the alias lookup functionality of the graph store. // TestAliasLookup tests the alias lookup functionality of the graph store.
func TestAliasLookup(t *testing.T) { func TestAliasLookup(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraphNew(t) graph := MakeTestGraphNew(t)
@@ -336,7 +339,7 @@ func TestAliasLookup(t *testing.T) {
// Add the node to the graph's database, this should also insert an // Add the node to the graph's database, this should also insert an
// entry into the alias index for this node. // entry into the alias index for this node.
require.NoError(t, graph.AddLightningNode(testNode)) require.NoError(t, graph.AddLightningNode(ctx, testNode))
// Next, attempt to lookup the alias. The alias should exactly match // Next, attempt to lookup the alias. The alias should exactly match
// the one which the test node was assigned. // the one which the test node was assigned.
@@ -801,18 +804,19 @@ func createChannelEdge(node1, node2 *models.LightningNode,
func TestEdgeInfoUpdates(t *testing.T) { func TestEdgeInfoUpdates(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
// We'd like to test the update of edges inserted into the database, so // We'd like to test the update of edges inserted into the database, so
// we create two vertexes to connect. // we create two vertexes to connect.
node1 := createTestVertex(t) node1 := createTestVertex(t)
if err := graph.AddLightningNode(node1); err != nil { if err := graph.AddLightningNode(ctx, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
assertNodeInCache(t, graph, node1, testFeatures) assertNodeInCache(t, graph, node1, testFeatures)
node2 := createTestVertex(t) node2 := createTestVertex(t)
if err := graph.AddLightningNode(node2); err != nil { if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
assertNodeInCache(t, graph, node2, testFeatures) assertNodeInCache(t, graph, node2, testFeatures)
@@ -1483,6 +1487,8 @@ func TestGraphCacheTraversal(t *testing.T) {
func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes,
numChannels int) (map[uint64]struct{}, []*models.LightningNode) { numChannels int) (map[uint64]struct{}, []*models.LightningNode) {
ctx := context.Background()
nodes := make([]*models.LightningNode, numNodes) nodes := make([]*models.LightningNode, numNodes)
nodeIndex := map[string]struct{}{} nodeIndex := map[string]struct{}{}
for i := 0; i < numNodes; i++ { for i := 0; i < numNodes; i++ {
@@ -1495,7 +1501,7 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes,
// Add each of the nodes into the graph, they should be inserted // Add each of the nodes into the graph, they should be inserted
// without error. // without error.
for _, node := range nodes { for _, node := range nodes {
require.NoError(t, graph.AddLightningNode(node)) require.NoError(t, graph.AddLightningNode(ctx, node))
} }
// Iterate over each node as returned by the graph, if all nodes are // Iterate over each node as returned by the graph, if all nodes are
@@ -1671,6 +1677,7 @@ func assertChanViewEqualChanPoints(t *testing.T, a []EdgePoint,
func TestGraphPruning(t *testing.T) { func TestGraphPruning(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
@@ -1687,7 +1694,7 @@ func TestGraphPruning(t *testing.T) {
for i := 0; i < numNodes; i++ { for i := 0; i < numNodes; i++ {
node := createTestVertex(t) node := createTestVertex(t)
if err := graph.AddLightningNode(node); err != nil { if err := graph.AddLightningNode(ctx, node); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -1919,6 +1926,7 @@ func TestHighestChanID(t *testing.T) {
// insertion of a new edge, the edge update index is updated properly. // insertion of a new edge, the edge update index is updated properly.
func TestChanUpdatesInHorizon(t *testing.T) { func TestChanUpdatesInHorizon(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
@@ -1935,11 +1943,11 @@ func TestChanUpdatesInHorizon(t *testing.T) {
// We'll start by creating two nodes which will seed our test graph. // We'll start by creating two nodes which will seed our test graph.
node1 := createTestVertex(t) node1 := createTestVertex(t)
if err := graph.AddLightningNode(node1); err != nil { if err := graph.AddLightningNode(ctx, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
node2 := createTestVertex(t) node2 := createTestVertex(t)
if err := graph.AddLightningNode(node2); err != nil { if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -2077,6 +2085,7 @@ func TestChanUpdatesInHorizon(t *testing.T) {
// the most recent node updates within a particular time horizon. // the most recent node updates within a particular time horizon.
func TestNodeUpdatesInHorizon(t *testing.T) { func TestNodeUpdatesInHorizon(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraphNew(t) graph := MakeTestGraphNew(t)
@@ -2108,7 +2117,7 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
nodeAnns = append(nodeAnns, *nodeAnn) nodeAnns = append(nodeAnns, *nodeAnn)
require.NoError(t, graph.AddLightningNode(nodeAnn)) require.NoError(t, graph.AddLightningNode(ctx, nodeAnn))
} }
queryCases := []struct { queryCases := []struct {
@@ -2245,6 +2254,7 @@ func TestFilterKnownChanIDsZombieRevival(t *testing.T) {
// know of on disk. // know of on disk.
func TestFilterKnownChanIDs(t *testing.T) { func TestFilterKnownChanIDs(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
@@ -2277,11 +2287,11 @@ func TestFilterKnownChanIDs(t *testing.T) {
// We'll start by creating two nodes which will seed our test graph. // We'll start by creating two nodes which will seed our test graph.
node1 := createTestVertex(t) node1 := createTestVertex(t)
if err := graph.AddLightningNode(node1); err != nil { if err := graph.AddLightningNode(ctx, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
node2 := createTestVertex(t) node2 := createTestVertex(t)
if err := graph.AddLightningNode(node2); err != nil { if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -2422,14 +2432,15 @@ func TestFilterKnownChanIDs(t *testing.T) {
// methods that acquire the cache mutex along with the DB mutex. // methods that acquire the cache mutex along with the DB mutex.
func TestStressTestChannelGraphAPI(t *testing.T) { func TestStressTestChannelGraphAPI(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
node1 := createTestVertex(t) node1 := createTestVertex(t)
require.NoError(t, graph.AddLightningNode(node1)) require.NoError(t, graph.AddLightningNode(ctx, node1))
node2 := createTestVertex(t) node2 := createTestVertex(t)
require.NoError(t, graph.AddLightningNode(node2)) require.NoError(t, graph.AddLightningNode(ctx, node2))
require.NoError(t, graph.SetSourceNode(node1)) require.NoError(t, graph.SetSourceNode(node1))
@@ -2706,16 +2717,17 @@ func TestStressTestChannelGraphAPI(t *testing.T) {
// set of short channel ID's for a given block range. // set of short channel ID's for a given block range.
func TestFilterChannelRange(t *testing.T) { func TestFilterChannelRange(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
// We'll first populate our graph with two nodes. All channels created // We'll first populate our graph with two nodes. All channels created
// below will be made between these two nodes. // below will be made between these two nodes.
node1 := createTestVertex(t) node1 := createTestVertex(t)
require.NoError(t, graph.AddLightningNode(node1)) require.NoError(t, graph.AddLightningNode(ctx, node1))
node2 := createTestVertex(t) node2 := createTestVertex(t)
require.NoError(t, graph.AddLightningNode(node2)) require.NoError(t, graph.AddLightningNode(ctx, node2))
// If we try to filter a channel range before we have any channels // If we try to filter a channel range before we have any channels
// inserted, we should get an empty slice of results. // inserted, we should get an empty slice of results.
@@ -2922,17 +2934,18 @@ func TestFilterChannelRange(t *testing.T) {
// of ChannelEdge structs for a given set of short channel ID's. // of ChannelEdge structs for a given set of short channel ID's.
func TestFetchChanInfos(t *testing.T) { func TestFetchChanInfos(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
// We'll first populate our graph with two nodes. All channels created // We'll first populate our graph with two nodes. All channels created
// below will be made between these two nodes. // below will be made between these two nodes.
node1 := createTestVertex(t) node1 := createTestVertex(t)
if err := graph.AddLightningNode(node1); err != nil { if err := graph.AddLightningNode(ctx, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
node2 := createTestVertex(t) node2 := createTestVertex(t)
if err := graph.AddLightningNode(node2); err != nil { if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -3024,16 +3037,17 @@ func TestFetchChanInfos(t *testing.T) {
// both sides. // both sides.
func TestIncompleteChannelPolicies(t *testing.T) { func TestIncompleteChannelPolicies(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
// Create two nodes. // Create two nodes.
node1 := createTestVertex(t) node1 := createTestVertex(t)
if err := graph.AddLightningNode(node1); err != nil { if err := graph.AddLightningNode(ctx, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
node2 := createTestVertex(t) node2 := createTestVertex(t)
if err := graph.AddLightningNode(node2); err != nil { if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -3119,6 +3133,7 @@ func TestIncompleteChannelPolicies(t *testing.T) {
// up. // up.
func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) { func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
@@ -3136,11 +3151,11 @@ func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) {
// We'll first populate our graph with two nodes. All channels created // We'll first populate our graph with two nodes. All channels created
// below will be made between these two nodes. // below will be made between these two nodes.
node1 := createTestVertex(t) node1 := createTestVertex(t)
if err := graph.AddLightningNode(node1); err != nil { if err := graph.AddLightningNode(ctx, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
node2 := createTestVertex(t) node2 := createTestVertex(t)
if err := graph.AddLightningNode(node2); err != nil { if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -3266,6 +3281,7 @@ func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) {
// PruneSyncState method. // PruneSyncState method.
func TestPruneGraphNodes(t *testing.T) { func TestPruneGraphNodes(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
@@ -3280,15 +3296,15 @@ func TestPruneGraphNodes(t *testing.T) {
// channel graph, at the end of the scenario, only two of these nodes // channel graph, at the end of the scenario, only two of these nodes
// should still be in the graph. // should still be in the graph.
node1 := createTestVertex(t) node1 := createTestVertex(t)
if err := graph.AddLightningNode(node1); err != nil { if err := graph.AddLightningNode(ctx, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
node2 := createTestVertex(t) node2 := createTestVertex(t)
if err := graph.AddLightningNode(node2); err != nil { if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
node3 := createTestVertex(t) node3 := createTestVertex(t)
if err := graph.AddLightningNode(node3); err != nil { if err := graph.AddLightningNode(ctx, node3); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -3331,6 +3347,7 @@ func TestPruneGraphNodes(t *testing.T) {
// database, then shell edges are created for each node if needed. // database, then shell edges are created for each node if needed.
func TestAddChannelEdgeShellNodes(t *testing.T) { func TestAddChannelEdgeShellNodes(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraphNew(t) graph := MakeTestGraphNew(t)
@@ -3361,7 +3378,7 @@ func TestAddChannelEdgeShellNodes(t *testing.T) {
require.ErrorIs(t, err, ErrEdgeAlreadyExist) require.ErrorIs(t, err, ErrEdgeAlreadyExist)
// Show that updating the shell node to a full node record works. // Show that updating the shell node to a full node record works.
require.NoError(t, graph.AddLightningNode(node2)) require.NoError(t, graph.AddLightningNode(ctx, node2))
} }
// TestNodePruningUpdateIndexDeletion tests that once a node has been removed // TestNodePruningUpdateIndexDeletion tests that once a node has been removed
@@ -3369,13 +3386,14 @@ func TestAddChannelEdgeShellNodes(t *testing.T) {
// well. // well.
func TestNodePruningUpdateIndexDeletion(t *testing.T) { func TestNodePruningUpdateIndexDeletion(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraphNew(t) graph := MakeTestGraphNew(t)
// We'll first populate our graph with a single node that will be // We'll first populate our graph with a single node that will be
// removed shortly. // removed shortly.
node1 := createTestVertex(t) node1 := createTestVertex(t)
if err := graph.AddLightningNode(node1); err != nil { if err := graph.AddLightningNode(ctx, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -3416,6 +3434,7 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
// public within the network graph. // public within the network graph.
func TestNodeIsPublic(t *testing.T) { func TestNodeIsPublic(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
// We'll start off the test by creating a small network of 3 // We'll start off the test by creating a small network of 3
// participants with the following graph: // participants with the following graph:
@@ -3453,9 +3472,8 @@ func TestNodeIsPublic(t *testing.T) {
graphs := []*ChannelGraph{aliceGraph, bobGraph, carolGraph} graphs := []*ChannelGraph{aliceGraph, bobGraph, carolGraph}
for _, graph := range graphs { for _, graph := range graphs {
for _, node := range nodes { for _, node := range nodes {
if err := graph.AddLightningNode(node); err != nil { err := graph.AddLightningNode(ctx, node)
t.Fatalf("unable to add node: %v", err) require.NoError(t, err)
}
} }
for _, edge := range edges { for _, edge := range edges {
if err := graph.AddChannelEdge(edge); err != nil { if err := graph.AddChannelEdge(edge); err != nil {
@@ -3551,24 +3569,25 @@ func TestNodeIsPublic(t *testing.T) {
// DisabledChannelIDs is correct. // DisabledChannelIDs is correct.
func TestDisabledChannelIDs(t *testing.T) { func TestDisabledChannelIDs(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
// Create first node and add it to the graph. // Create first node and add it to the graph.
node1 := createTestVertex(t) node1 := createTestVertex(t)
if err := graph.AddLightningNode(node1); err != nil { if err := graph.AddLightningNode(ctx, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
// Create second node and add it to the graph. // Create second node and add it to the graph.
node2 := createTestVertex(t) node2 := createTestVertex(t)
if err := graph.AddLightningNode(node2); err != nil { if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
// Adding a new channel edge to the graph. // Adding a new channel edge to the graph.
edgeInfo, edge1, edge2 := createChannelEdge(node1, node2) edgeInfo, edge1, edge2 := createChannelEdge(node1, node2)
if err := graph.AddLightningNode(node2); err != nil { if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -3634,6 +3653,7 @@ func TestDisabledChannelIDs(t *testing.T) {
// receive the proper update. // receive the proper update.
func TestEdgePolicyMissingMaxHtcl(t *testing.T) { func TestEdgePolicyMissingMaxHtcl(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
@@ -3646,13 +3666,13 @@ func TestEdgePolicyMissingMaxHtcl(t *testing.T) {
// We'd like to test the update of edges inserted into the database, so // We'd like to test the update of edges inserted into the database, so
// we create two vertexes to connect. // we create two vertexes to connect.
node1 := createTestVertex(t) node1 := createTestVertex(t)
if err := graph.AddLightningNode(node1); err != nil { if err := graph.AddLightningNode(ctx, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
node2 := createTestVertex(t) node2 := createTestVertex(t)
edgeInfo, edge1, edge2 := createChannelEdge(node1, node2) edgeInfo, edge1, edge2 := createChannelEdge(node1, node2)
if err := graph.AddLightningNode(node2); err != nil { if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
if err := graph.AddChannelEdge(edgeInfo); err != nil { if err := graph.AddChannelEdge(edgeInfo); err != nil {
@@ -4037,15 +4057,16 @@ func TestBatchedAddChannelEdge(t *testing.T) {
// executes multiple UpdateEdgePolicy requests in a single txn. // executes multiple UpdateEdgePolicy requests in a single txn.
func TestBatchedUpdateEdgePolicy(t *testing.T) { func TestBatchedUpdateEdgePolicy(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraphNew(t) graph := MakeTestGraphNew(t)
// We'd like to test the update of edges inserted into the database, so // We'd like to test the update of edges inserted into the database, so
// we create two vertexes to connect. // we create two vertexes to connect.
node1 := createTestVertex(t) node1 := createTestVertex(t)
require.NoError(t, graph.AddLightningNode(node1)) require.NoError(t, graph.AddLightningNode(ctx, node1))
node2 := createTestVertex(t) node2 := createTestVertex(t)
require.NoError(t, graph.AddLightningNode(node2)) require.NoError(t, graph.AddLightningNode(ctx, node2))
// Create an edge and add it to the db. // Create an edge and add it to the db.
edgeInfo, edge1, edge2 := createChannelEdge(node1, node2) edgeInfo, edge1, edge2 := createChannelEdge(node1, node2)
@@ -4139,6 +4160,7 @@ func BenchmarkForEachChannel(b *testing.B) {
// method 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) {
t.Parallel() t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t) graph := MakeTestGraph(t)
@@ -4147,9 +4169,9 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) {
graph.graphCache = nil graph.graphCache = nil
node1 := createTestVertex(t) node1 := createTestVertex(t)
require.NoError(t, graph.AddLightningNode(node1)) require.NoError(t, graph.AddLightningNode(ctx, node1))
node2 := createTestVertex(t) node2 := createTestVertex(t)
require.NoError(t, graph.AddLightningNode(node2)) require.NoError(t, graph.AddLightningNode(ctx, node2))
// Create an edge and add it to the db. // Create an edge and add it to the db.
edgeInfo, e1, e2 := createChannelEdge(node1, node2) edgeInfo, e1, e2 := createChannelEdge(node1, node2)
@@ -4305,6 +4327,7 @@ var testNodeAnn = "01012674c2e7ef68c73a086b7de2603f4ef1567358df84bb4edaa06c" +
// that the two messages are equal. // that the two messages are equal.
func TestLightningNodePersistence(t *testing.T) { func TestLightningNodePersistence(t *testing.T) {
t.Parallel() t.Parallel()
ctx := context.Background()
// Create a new test graph instance. // Create a new test graph instance.
graph := MakeTestGraphNew(t) graph := MakeTestGraphNew(t)
@@ -4323,7 +4346,7 @@ func TestLightningNodePersistence(t *testing.T) {
node := models.NodeFromWireAnnouncement(na) node := models.NodeFromWireAnnouncement(na)
// Persist the node to disk. // Persist the node to disk.
err = graph.AddLightningNode(node) err = graph.AddLightningNode(ctx, node)
require.NoError(t, err) require.NoError(t, err)
// Read the node from disk. // Read the node from disk.
+2 -1
View File
@@ -1,6 +1,7 @@
package graphdb package graphdb
import ( import (
"context"
"net" "net"
"time" "time"
@@ -56,7 +57,7 @@ type V1Store interface { //nolint:interfacebloat
// update that node's information. Note that this method is expected to // update that node's information. Note that this method is expected to
// only be called to update an already present node from a node // only be called to update an already present node from a node
// announcement, or to insert a node found in a channel update. // announcement, or to insert a node found in a channel update.
AddLightningNode(node *models.LightningNode, AddLightningNode(ctx context.Context, node *models.LightningNode,
op ...batch.SchedulerOption) error op ...batch.SchedulerOption) error
// AddrsForNode returns all known addresses for the target node public // AddrsForNode returns all known addresses for the target node public
+2 -4
View File
@@ -957,10 +957,8 @@ func (c *KVStore) SetSourceNode(node *models.LightningNode) error {
// channel update. // channel update.
// //
// TODO(roasbeef): also need sig of announcement. // TODO(roasbeef): also need sig of announcement.
func (c *KVStore) AddLightningNode(node *models.LightningNode, func (c *KVStore) AddLightningNode(ctx context.Context,
opts ...batch.SchedulerOption) error { node *models.LightningNode, opts ...batch.SchedulerOption) error {
ctx := context.TODO()
r := &batch.Request[kvdb.RwTx]{ r := &batch.Request[kvdb.RwTx]{
Opts: batch.NewSchedulerOptions(opts...), Opts: batch.NewSchedulerOptions(opts...),
+2 -4
View File
@@ -174,10 +174,8 @@ func NewSQLStore(cfg *SQLStoreConfig, db BatchedSQLQueries, kvStore *KVStore,
// information. // information.
// //
// NOTE: part of the V1Store interface. // NOTE: part of the V1Store interface.
func (s *SQLStore) AddLightningNode(node *models.LightningNode, func (s *SQLStore) AddLightningNode(ctx context.Context,
opts ...batch.SchedulerOption) error { node *models.LightningNode, opts ...batch.SchedulerOption) error {
ctx := context.TODO()
r := &batch.Request[SQLQueries]{ r := &batch.Request[SQLQueries]{
Opts: batch.NewSchedulerOptions(opts...), Opts: batch.NewSchedulerOptions(opts...),
+3 -2
View File
@@ -1,6 +1,7 @@
package graph package graph
import ( import (
"context"
"time" "time"
"github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/chaincfg/chainhash"
@@ -22,7 +23,7 @@ type ChannelGraphSource interface {
// AddNode is used to add information about a node to the router // AddNode is used to add information about a node to the router
// database. If the node with this pubkey is not present in an existing // database. If the node with this pubkey is not present in an existing
// channel, it will be ignored. // channel, it will be ignored.
AddNode(node *models.LightningNode, AddNode(ctx context.Context, node *models.LightningNode,
op ...batch.SchedulerOption) error op ...batch.SchedulerOption) error
// AddEdge is used to add edge/channel to the topology of the router, // AddEdge is used to add edge/channel to the topology of the router,
@@ -202,7 +203,7 @@ type DB interface {
// update that node's information. Note that this method is expected to // update that node's information. Note that this method is expected to
// only be called to update an already present node from a node // only be called to update an already present node from a node
// announcement, or to insert a node found in a channel update. // announcement, or to insert a node found in a channel update.
AddLightningNode(node *models.LightningNode, AddLightningNode(ctx context.Context, node *models.LightningNode,
op ...batch.SchedulerOption) error op ...batch.SchedulerOption) error
// AddChannelEdge adds a new (undirected, blank) edge to the graph // AddChannelEdge adds a new (undirected, blank) edge to the graph
+8 -5
View File
@@ -2,6 +2,7 @@ package graph
import ( import (
"bytes" "bytes"
"context"
"encoding/hex" "encoding/hex"
"fmt" "fmt"
"image/color" "image/color"
@@ -608,6 +609,7 @@ func TestEdgeUpdateNotification(t *testing.T) {
// attributes with new data. // attributes with new data.
func TestNodeUpdateNotification(t *testing.T) { func TestNodeUpdateNotification(t *testing.T) {
t.Parallel() t.Parallel()
ctxb := context.Background()
const startingBlockHeight = 101 const startingBlockHeight = 101
ctx := createTestCtxSingleNode(t, startingBlockHeight) ctx := createTestCtxSingleNode(t, startingBlockHeight)
@@ -665,10 +667,10 @@ func TestNodeUpdateNotification(t *testing.T) {
// Change network topology by adding the updated info for the two nodes // Change network topology by adding the updated info for the two nodes
// to the channel router. // to the channel router.
if err := ctx.builder.AddNode(node1); err != nil { if err := ctx.builder.AddNode(ctxb, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
if err := ctx.builder.AddNode(node2); err != nil { if err := ctx.builder.AddNode(ctxb, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -763,7 +765,7 @@ func TestNodeUpdateNotification(t *testing.T) {
nodeUpdateAnn.LastUpdate = node1.LastUpdate.Add(300 * time.Millisecond) nodeUpdateAnn.LastUpdate = node1.LastUpdate.Add(300 * time.Millisecond)
// Add new node topology update to the channel router. // Add new node topology update to the channel router.
if err := ctx.builder.AddNode(&nodeUpdateAnn); err != nil { if err := ctx.builder.AddNode(ctxb, &nodeUpdateAnn); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
@@ -790,6 +792,7 @@ func TestNodeUpdateNotification(t *testing.T) {
// when the client wishes to exit. // when the client wishes to exit.
func TestNotificationCancellation(t *testing.T) { func TestNotificationCancellation(t *testing.T) {
t.Parallel() t.Parallel()
ctxb := context.Background()
const startingBlockHeight = 101 const startingBlockHeight = 101
ctx := createTestCtxSingleNode(t, startingBlockHeight) ctx := createTestCtxSingleNode(t, startingBlockHeight)
@@ -845,11 +848,11 @@ func TestNotificationCancellation(t *testing.T) {
t.Fatalf("unable to add edge: %v", err) t.Fatalf("unable to add edge: %v", err)
} }
if err := ctx.builder.AddNode(node1); err != nil { if err := ctx.builder.AddNode(ctxb, node1); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
if err := ctx.builder.AddNode(node2); err != nil { if err := ctx.builder.AddNode(ctxb, node2); err != nil {
t.Fatalf("unable to add node: %v", err) t.Fatalf("unable to add node: %v", err)
} }
+1 -1
View File
@@ -260,7 +260,7 @@ func (s *Server) ImportGraph(ctx context.Context,
return nil, err return nil, err
} }
if err := graphDB.AddLightningNode(node); err != nil { if err := graphDB.AddLightningNode(ctx, node); err != nil {
return nil, fmt.Errorf("unable to add node %v: %w", return nil, fmt.Errorf("unable to add node %v: %w",
rpcNode.PubKey, err) rpcNode.PubKey, err)
} }
+7 -2
View File
@@ -2,6 +2,7 @@ package routing
import ( import (
"bytes" "bytes"
"context"
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
@@ -181,6 +182,8 @@ func makeTestGraph(t *testing.T, useCache bool) (*graphdb.ChannelGraph,
func parseTestGraph(t *testing.T, useCache bool, path string) ( func parseTestGraph(t *testing.T, useCache bool, path string) (
*testGraphInstance, error) { *testGraphInstance, error) {
ctx := context.Background()
graphJSON, err := os.ReadFile(path) graphJSON, err := os.ReadFile(path)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -290,7 +293,7 @@ func parseTestGraph(t *testing.T, useCache bool, path string) (
// With the node fully parsed, add it as a vertex within the // With the node fully parsed, add it as a vertex within the
// graph. // graph.
if err := graph.AddLightningNode(dbNode); err != nil { if err := graph.AddLightningNode(ctx, dbNode); err != nil {
return nil, err return nil, err
} }
} }
@@ -515,6 +518,8 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
testChannels []*testChannel, source string, testChannels []*testChannel, source string,
sourceFeatureBits ...lnwire.FeatureBit) (*testGraphInstance, error) { sourceFeatureBits ...lnwire.FeatureBit) (*testGraphInstance, error) {
ctx := context.Background()
// We'll use this fake address for the IP address of all the nodes in // We'll use this fake address for the IP address of all the nodes in
// our tests. This value isn't needed for path finding so it doesn't // our tests. This value isn't needed for path finding so it doesn't
// need to be unique. // need to be unique.
@@ -566,7 +571,7 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
// With the node fully parsed, add it as a vertex within the // With the node fully parsed, add it as a vertex within the
// graph. // graph.
if err := graph.AddLightningNode(dbNode); err != nil { if err := graph.AddLightningNode(ctx, dbNode); err != nil {
return nil, err return nil, err
} }
+4 -2
View File
@@ -2,6 +2,7 @@ package routing
import ( import (
"bytes" "bytes"
"context"
"fmt" "fmt"
"image/color" "image/color"
"math" "math"
@@ -2706,6 +2707,7 @@ func TestNewRouteRequest(t *testing.T) {
// announcements for the channel vertexes to be able to use the channel. // announcements for the channel vertexes to be able to use the channel.
func TestAddEdgeUnknownVertexes(t *testing.T) { func TestAddEdgeUnknownVertexes(t *testing.T) {
t.Parallel() t.Parallel()
ctxb := context.Background()
const startingBlockHeight = 101 const startingBlockHeight = 101
ctx := createTestCtxFromFile(t, startingBlockHeight, basicGraphFilePath) ctx := createTestCtxFromFile(t, startingBlockHeight, basicGraphFilePath)
@@ -2879,7 +2881,7 @@ func TestAddEdgeUnknownVertexes(t *testing.T) {
} }
copy(n1.PubKeyBytes[:], priv1.PubKey().SerializeCompressed()) copy(n1.PubKeyBytes[:], priv1.PubKey().SerializeCompressed())
require.NoError(t, ctx.graph.AddLightningNode(n1)) require.NoError(t, ctx.graph.AddLightningNode(ctxb, n1))
n2 := &models.LightningNode{ n2 := &models.LightningNode{
HaveNodeAnnouncement: true, HaveNodeAnnouncement: true,
@@ -2892,7 +2894,7 @@ func TestAddEdgeUnknownVertexes(t *testing.T) {
} }
copy(n2.PubKeyBytes[:], priv2.PubKey().SerializeCompressed()) copy(n2.PubKeyBytes[:], priv2.PubKey().SerializeCompressed())
require.NoError(t, ctx.graph.AddLightningNode(n2)) require.NoError(t, ctx.graph.AddLightningNode(ctxb, n2))
// Should still be able to find the route, and the info should be // Should still be able to find the route, and the info should be
// updated. // updated.