graph/db: thread context through to FetchLightingNode
This commit is contained in:
12 files changed
+47
-39
No files matched your search
@@ -395,6 +395,8 @@ func TestPrefAttachmentSelectSkipNodes(t *testing.T) {
|
||||
func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey,
|
||||
capacity btcutil.Amount) (*ChannelEdge, *ChannelEdge, error) {
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
fetchNode := func(pub *btcec.PublicKey) (*models.LightningNode, error) {
|
||||
if pub != nil {
|
||||
vertex, err := route.NewVertexFromBytes(
|
||||
@@ -404,7 +406,7 @@ func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey,
|
||||
return nil, err
|
||||
}
|
||||
|
||||
dbNode, err := d.db.FetchLightningNode(vertex)
|
||||
dbNode, err := d.db.FetchLightningNode(ctx, vertex)
|
||||
switch {
|
||||
case errors.Is(err, graphdb.ErrGraphNodeNotFound):
|
||||
fallthrough
|
||||
@@ -737,7 +739,7 @@ func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo,
|
||||
}
|
||||
|
||||
func (t *testNodeTx) FetchNode(pub route.Vertex) (graphdb.NodeRTx, error) {
|
||||
node, err := t.db.db.FetchLightningNode(pub)
|
||||
node, err := t.db.db.FetchLightningNode(context.Background(), pub)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -2206,10 +2206,10 @@ func (d *AuthenticatedGossiper) processZombieUpdate(_ context.Context,
|
||||
|
||||
// fetchNodeAnn fetches the latest signed node announcement from our point of
|
||||
// view for the node with the given public key.
|
||||
func (d *AuthenticatedGossiper) fetchNodeAnn(_ context.Context,
|
||||
func (d *AuthenticatedGossiper) fetchNodeAnn(ctx context.Context,
|
||||
pubKey [33]byte) (*lnwire.NodeAnnouncement, error) {
|
||||
|
||||
node, err := d.cfg.Graph.FetchLightningNode(pubKey)
|
||||
node, err := d.cfg.Graph.FetchLightningNode(ctx, pubKey)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -294,7 +294,7 @@ func (r *mockGraphSource) GetChannelByID(chanID lnwire.ShortChannelID) (
|
||||
return &chanInfo, edge1, edge2, nil
|
||||
}
|
||||
|
||||
func (r *mockGraphSource) FetchLightningNode(
|
||||
func (r *mockGraphSource) FetchLightningNode(_ context.Context,
|
||||
nodePub route.Vertex) (*models.LightningNode, error) {
|
||||
|
||||
for _, node := range r.nodes {
|
||||
|
||||
+2
-2
@@ -1265,10 +1265,10 @@ func (b *Builder) GetChannelByID(chanID lnwire.ShortChannelID) (
|
||||
// within the graph.
|
||||
//
|
||||
// NOTE: This method is part of the ChannelGraphSource interface.
|
||||
func (b *Builder) FetchLightningNode(
|
||||
func (b *Builder) FetchLightningNode(ctx context.Context,
|
||||
node route.Vertex) (*models.LightningNode, error) {
|
||||
|
||||
return b.cfg.Graph.FetchLightningNode(node)
|
||||
return b.cfg.Graph.FetchLightningNode(ctx, node)
|
||||
}
|
||||
|
||||
// ForAllOutgoingChannels is used to iterate over all outgoing channels owned by
|
||||
|
||||
+14
-13
@@ -132,7 +132,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
|
||||
|
||||
// Next, fetch the node from the database to ensure everything was
|
||||
// serialized properly.
|
||||
dbNode, err := graph.FetchLightningNode(testPub)
|
||||
dbNode, err := graph.FetchLightningNode(ctx, testPub)
|
||||
require.NoError(t, err, "unable to locate node")
|
||||
|
||||
_, exists, err := graph.HasLightningNode(dbNode.PubKeyBytes)
|
||||
@@ -170,7 +170,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
|
||||
|
||||
// Finally, attempt to fetch the node again. This should fail as the
|
||||
// node should have been deleted from the database.
|
||||
_, err = graph.FetchLightningNode(testPub)
|
||||
_, err = graph.FetchLightningNode(ctx, testPub)
|
||||
require.ErrorIs(t, err, ErrGraphNodeNotFound)
|
||||
|
||||
// Now, we'll specifically test the updating of addresses of a node
|
||||
@@ -192,7 +192,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
|
||||
require.NoError(t, graph.AddLightningNode(ctx, node))
|
||||
|
||||
// Fetch the node and assert the empty addresses.
|
||||
dbNode, err = graph.FetchLightningNode(testPub)
|
||||
dbNode, err = graph.FetchLightningNode(ctx, testPub)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, dbNode.Addresses)
|
||||
|
||||
@@ -219,7 +219,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
|
||||
require.NoError(t, graph.AddLightningNode(ctx, node))
|
||||
|
||||
// Fetch the node and assert the updated addresses.
|
||||
dbNode, err = graph.FetchLightningNode(testPub)
|
||||
dbNode, err = graph.FetchLightningNode(ctx, testPub)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, expAddrs, dbNode.Addresses)
|
||||
|
||||
@@ -240,7 +240,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
|
||||
require.NoError(t, graph.AddLightningNode(ctx, node))
|
||||
|
||||
// Fetch the node and assert the updated addresses.
|
||||
dbNode, err = graph.FetchLightningNode(testPub)
|
||||
dbNode, err = graph.FetchLightningNode(ctx, testPub)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, expAddrs, dbNode.Addresses)
|
||||
|
||||
@@ -253,7 +253,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
|
||||
require.NoError(t, graph.AddLightningNode(ctx, node))
|
||||
|
||||
// Fetch the node and assert the updated addresses.
|
||||
dbNode, err = graph.FetchLightningNode(testPub)
|
||||
dbNode, err = graph.FetchLightningNode(ctx, testPub)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, expAddrs, dbNode.Addresses)
|
||||
}
|
||||
@@ -262,6 +262,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
|
||||
// only the pubkey is known to the database.
|
||||
func TestPartialNode(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
|
||||
graph := MakeTestGraphNew(t)
|
||||
|
||||
@@ -282,9 +283,9 @@ func TestPartialNode(t *testing.T) {
|
||||
|
||||
// Next, fetch the node2 from the database to ensure everything was
|
||||
// serialized properly.
|
||||
dbNode1, err := graph.FetchLightningNode(pubKey1)
|
||||
dbNode1, err := graph.FetchLightningNode(ctx, pubKey1)
|
||||
require.NoError(t, err)
|
||||
dbNode2, err := graph.FetchLightningNode(pubKey2)
|
||||
dbNode2, err := graph.FetchLightningNode(ctx, pubKey2)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, exists, err := graph.HasLightningNode(dbNode1.PubKeyBytes)
|
||||
@@ -322,7 +323,7 @@ func TestPartialNode(t *testing.T) {
|
||||
|
||||
// Finally, attempt to fetch the node again. This should fail as the
|
||||
// node should have been deleted from the database.
|
||||
_, err = graph.FetchLightningNode(testPub)
|
||||
_, err = graph.FetchLightningNode(ctx, testPub)
|
||||
require.ErrorIs(t, err, ErrGraphNodeNotFound)
|
||||
}
|
||||
|
||||
@@ -3338,7 +3339,7 @@ func TestPruneGraphNodes(t *testing.T) {
|
||||
|
||||
// Finally, we'll ensure that node3, the only fully unconnected node as
|
||||
// properly deleted from the graph and not another node in its place.
|
||||
_, err := graph.FetchLightningNode(node3.PubKeyBytes)
|
||||
_, err := graph.FetchLightningNode(ctx, node3.PubKeyBytes)
|
||||
require.NotNil(t, err)
|
||||
}
|
||||
|
||||
@@ -3364,11 +3365,11 @@ func TestAddChannelEdgeShellNodes(t *testing.T) {
|
||||
|
||||
// Ensure that node1 was inserted as a full node, while node2 only has
|
||||
// a shell node present.
|
||||
node1, err := graph.FetchLightningNode(node1.PubKeyBytes)
|
||||
node1, err := graph.FetchLightningNode(ctx, node1.PubKeyBytes)
|
||||
require.NoError(t, err, "unable to fetch node1")
|
||||
require.True(t, node1.HaveNodeAnnouncement)
|
||||
|
||||
node2, err = graph.FetchLightningNode(node2.PubKeyBytes)
|
||||
node2, err = graph.FetchLightningNode(ctx, node2.PubKeyBytes)
|
||||
require.NoError(t, err, "unable to fetch node2")
|
||||
require.False(t, node2.HaveNodeAnnouncement)
|
||||
|
||||
@@ -4350,7 +4351,7 @@ func TestLightningNodePersistence(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
|
||||
// Read the node from disk.
|
||||
diskNode, err := graph.FetchLightningNode(node.PubKeyBytes)
|
||||
diskNode, err := graph.FetchLightningNode(ctx, node.PubKeyBytes)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Convert it back to a wire message.
|
||||
|
||||
@@ -126,8 +126,8 @@ type V1Store interface { //nolint:interfacebloat
|
||||
// FetchLightningNode attempts to look up a target node by its identity
|
||||
// public key. If the node isn't found in the database, then
|
||||
// ErrGraphNodeNotFound is returned.
|
||||
FetchLightningNode(nodePub route.Vertex) (
|
||||
*models.LightningNode, error)
|
||||
FetchLightningNode(ctx context.Context,
|
||||
nodePub route.Vertex) (*models.LightningNode, error)
|
||||
|
||||
// HasLightningNode determines if the graph has a vertex identified by
|
||||
// the target node identity public key. If the node exists in the
|
||||
|
||||
@@ -379,12 +379,14 @@ func initKVStore(db kvdb.Backend) error {
|
||||
func (c *KVStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr,
|
||||
error) {
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
pubKey, err := route.NewVertexFromBytes(nodePub.SerializeCompressed())
|
||||
if err != nil {
|
||||
return false, nil, err
|
||||
}
|
||||
|
||||
node, err := c.FetchLightningNode(pubKey)
|
||||
node, err := c.FetchLightningNode(ctx, pubKey)
|
||||
// We don't consider it an error if the graph is unaware of the node.
|
||||
switch {
|
||||
case err != nil && !errors.Is(err, ErrGraphNodeNotFound):
|
||||
@@ -2995,8 +2997,8 @@ func (c *KVStore) FetchLightningNodeTx(tx kvdb.RTx, nodePub route.Vertex) (
|
||||
// FetchLightningNode attempts to look up a target node by its identity public
|
||||
// key. If the node isn't found in the database, then ErrGraphNodeNotFound is
|
||||
// returned.
|
||||
func (c *KVStore) FetchLightningNode(nodePub route.Vertex) (
|
||||
*models.LightningNode, error) {
|
||||
func (c *KVStore) FetchLightningNode(_ context.Context,
|
||||
nodePub route.Vertex) (*models.LightningNode, error) {
|
||||
|
||||
return c.fetchLightningNode(nil, nodePub)
|
||||
}
|
||||
|
||||
@@ -193,10 +193,8 @@ func (s *SQLStore) AddLightningNode(ctx context.Context,
|
||||
// returned.
|
||||
//
|
||||
// NOTE: part of the V1Store interface.
|
||||
func (s *SQLStore) FetchLightningNode(pubKey route.Vertex) (
|
||||
*models.LightningNode, error) {
|
||||
|
||||
ctx := context.TODO()
|
||||
func (s *SQLStore) FetchLightningNode(ctx context.Context,
|
||||
pubKey route.Vertex) (*models.LightningNode, error) {
|
||||
|
||||
var node *models.LightningNode
|
||||
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
||||
|
||||
+4
-3
@@ -84,7 +84,8 @@ type ChannelGraphSource interface {
|
||||
// FetchLightningNode attempts to look up a target node by its identity
|
||||
// public key. channeldb.ErrGraphNodeNotFound is returned if the node
|
||||
// doesn't exist within the graph.
|
||||
FetchLightningNode(route.Vertex) (*models.LightningNode, error)
|
||||
FetchLightningNode(context.Context,
|
||||
route.Vertex) (*models.LightningNode, error)
|
||||
|
||||
// MarkZombieEdge marks the channel with the given ID as a zombie edge.
|
||||
MarkZombieEdge(chanID uint64) error
|
||||
@@ -242,8 +243,8 @@ type DB interface {
|
||||
// FetchLightningNode attempts to look up a target node by its identity
|
||||
// public key. If the node isn't found in the database, then
|
||||
// ErrGraphNodeNotFound is returned.
|
||||
FetchLightningNode(nodePub route.Vertex) (*models.LightningNode,
|
||||
error)
|
||||
FetchLightningNode(ctx context.Context,
|
||||
nodePub route.Vertex) (*models.LightningNode, error)
|
||||
|
||||
// ForEachNodeChannel iterates through all channels of the given node,
|
||||
// executing the passed callback with an edge info structure and the
|
||||
|
||||
@@ -2907,12 +2907,12 @@ func TestAddEdgeUnknownVertexes(t *testing.T) {
|
||||
_, _, err = ctx.router.FindRoute(req)
|
||||
require.NoError(t, err, "unable to find any routes")
|
||||
|
||||
copy1, err := ctx.graph.FetchLightningNode(pub1)
|
||||
copy1, err := ctx.graph.FetchLightningNode(ctxb, pub1)
|
||||
require.NoError(t, err, "unable to fetch node")
|
||||
|
||||
require.Equal(t, n1.Alias, copy1.Alias)
|
||||
|
||||
copy2, err := ctx.graph.FetchLightningNode(pub2)
|
||||
copy2, err := ctx.graph.FetchLightningNode(ctxb, pub2)
|
||||
require.NoError(t, err, "unable to fetch node")
|
||||
|
||||
require.Equal(t, n2.Alias, copy2.Alias)
|
||||
|
||||
+2
-2
@@ -6963,7 +6963,7 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context,
|
||||
// With the public key decoded, attempt to fetch the node corresponding
|
||||
// to this public key. If the node cannot be found, then an error will
|
||||
// be returned.
|
||||
node, err := graph.FetchLightningNode(pubKey)
|
||||
node, err := graph.FetchLightningNode(ctx, pubKey)
|
||||
switch {
|
||||
case errors.Is(err, graphdb.ErrGraphNodeNotFound):
|
||||
return nil, status.Error(codes.NotFound, err.Error())
|
||||
@@ -8040,7 +8040,7 @@ func (r *rpcServer) ForwardingHistory(ctx context.Context,
|
||||
return "", err
|
||||
}
|
||||
|
||||
peer, err := r.server.graphDB.FetchLightningNode(vertex)
|
||||
peer, err := r.server.graphDB.FetchLightningNode(ctx, vertex)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
@@ -4637,6 +4637,8 @@ func (s *server) peerInitializer(p *peer.Brontide) {
|
||||
func (s *server) peerTerminationWatcher(p *peer.Brontide, ready chan struct{}) {
|
||||
defer s.wg.Done()
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
p.WaitForDisconnect(ready)
|
||||
|
||||
srvrLog.Debugf("Peer %v has been disconnected", p)
|
||||
@@ -4723,7 +4725,7 @@ func (s *server) peerTerminationWatcher(p *peer.Brontide, ready chan struct{}) {
|
||||
|
||||
// We'll ensure that we locate all the peers advertised addresses for
|
||||
// reconnection purposes.
|
||||
advertisedAddrs, err := s.fetchNodeAdvertisedAddrs(pubKey)
|
||||
advertisedAddrs, err := s.fetchNodeAdvertisedAddrs(ctx, pubKey)
|
||||
switch {
|
||||
// We found advertised addresses, so use them.
|
||||
case err == nil:
|
||||
@@ -5216,13 +5218,15 @@ func computeNextBackoff(currBackoff, maxBackoff time.Duration) time.Duration {
|
||||
var errNoAdvertisedAddr = errors.New("no advertised address found")
|
||||
|
||||
// fetchNodeAdvertisedAddrs attempts to fetch the advertised addresses of a node.
|
||||
func (s *server) fetchNodeAdvertisedAddrs(pub *btcec.PublicKey) ([]net.Addr, error) {
|
||||
func (s *server) fetchNodeAdvertisedAddrs(ctx context.Context,
|
||||
pub *btcec.PublicKey) ([]net.Addr, error) {
|
||||
|
||||
vertex, err := route.NewVertexFromBytes(pub.SerializeCompressed())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
node, err := s.graphDB.FetchLightningNode(vertex)
|
||||
node, err := s.graphDB.FetchLightningNode(ctx, vertex)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user