Merge pull request #9956 from ellemouton/chanGraphContext

multi: add `context.Context` param to some `graphdb.V1Store` methods

autopilot/prefattach_test.go

@@ -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
@@ -422,7 +424,9 @@ func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey,
 					AuthSigBytes: testSig.Serialize(),
 				}
 				graphNode.AddPubKey(pub)
-				err := d.db.AddLightningNode(graphNode)
+				err := d.db.AddLightningNode(
+					context.Background(), graphNode,
+				)
 				if err != nil {
 					return nil, err
 				}
@@ -450,7 +454,9 @@ func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey,
 			AuthSigBytes: testSig.Serialize(),
 		}
 		dbNode.AddPubKey(nodeKey)
-		if err := d.db.AddLightningNode(dbNode); err != nil {
+		if err := d.db.AddLightningNode(
+			context.Background(), dbNode,
+		); err != nil {
 			return nil, err
 		}
 
@@ -554,7 +560,8 @@ func (d *testDBGraph) addRandNode() (*btcec.PublicKey, error) {
 		AuthSigBytes: testSig.Serialize(),
 	}
 	dbNode.AddPubKey(nodeKey)
-	if err := d.db.AddLightningNode(dbNode); err != nil {
+	err = d.db.AddLightningNode(context.Background(), dbNode)
+	if err != nil {
 		return nil, err
 	}
 
@@ -732,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
 	}

chanbackup/backup.go

@@ -1,6 +1,7 @@
 package chanbackup
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/btcsuite/btcd/wire"
@@ -24,7 +25,7 @@ type LiveChannelSource interface {
 // passed open channel. The backup includes all information required to restore
 // the channel, as well as addressing information so we can find the peer and
 // reconnect to them to initiate the protocol.
-func assembleChanBackup(addrSource channeldb.AddrSource,
+func assembleChanBackup(ctx context.Context, addrSource channeldb.AddrSource,
 	openChan *channeldb.OpenChannel) (*Single, error) {
 
 	log.Debugf("Crafting backup for ChannelPoint(%v)",
@@ -32,7 +33,9 @@ func assembleChanBackup(addrSource channeldb.AddrSource,
 
 	// First, we'll query the channel source to obtain all the addresses
 	// that are associated with the peer for this channel.
-	known, nodeAddrs, err := addrSource.AddrsForNode(openChan.IdentityPub)
+	known, nodeAddrs, err := addrSource.AddrsForNode(
+		ctx, openChan.IdentityPub,
+	)
 	if err != nil {
 		return nil, err
 	}
@@ -90,7 +93,8 @@ func buildCloseTxInputs(
 // FetchBackupForChan attempts to create a plaintext static channel backup for
 // the target channel identified by its channel point. If we're unable to find
 // the target channel, then an error will be returned.
-func FetchBackupForChan(chanPoint wire.OutPoint, chanSource LiveChannelSource,
+func FetchBackupForChan(ctx context.Context, chanPoint wire.OutPoint,
+	chanSource LiveChannelSource,
 	addrSource channeldb.AddrSource) (*Single, error) {
 
 	// First, we'll query the channel source to see if the channel is known
@@ -104,7 +108,7 @@ func FetchBackupForChan(chanPoint wire.OutPoint, chanSource LiveChannelSource,
 
 	// Once we have the target channel, we can assemble the backup using
 	// the source to obtain any extra information that we may need.
-	staticChanBackup, err := assembleChanBackup(addrSource, targetChan)
+	staticChanBackup, err := assembleChanBackup(ctx, addrSource, targetChan)
 	if err != nil {
 		return nil, fmt.Errorf("unable to create chan backup: %w", err)
 	}
@@ -114,7 +118,7 @@ func FetchBackupForChan(chanPoint wire.OutPoint, chanSource LiveChannelSource,
 
 // FetchStaticChanBackups will return a plaintext static channel back up for
 // all known active/open channels within the passed channel source.
-func FetchStaticChanBackups(chanSource LiveChannelSource,
+func FetchStaticChanBackups(ctx context.Context, chanSource LiveChannelSource,
 	addrSource channeldb.AddrSource) ([]Single, error) {
 
 	// First, we'll query the backup source for information concerning all
@@ -129,7 +133,7 @@ func FetchStaticChanBackups(chanSource LiveChannelSource,
 	// channel.
 	staticChanBackups := make([]Single, 0, len(openChans))
 	for _, openChan := range openChans {
-		chanBackup, err := assembleChanBackup(addrSource, openChan)
+		chanBackup, err := assembleChanBackup(ctx, addrSource, openChan)
 		if err != nil {
 			return nil, err
 		}

chanbackup/backup_test.go

@@ -1,6 +1,7 @@
 package chanbackup
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"testing"
@@ -61,8 +62,8 @@ func (m *mockChannelSource) addAddrsForNode(nodePub *btcec.PublicKey, addrs []ne
 	m.addrs[nodeKey] = addrs
 }
 
-func (m *mockChannelSource) AddrsForNode(nodePub *btcec.PublicKey) (bool,
+func (m *mockChannelSource) AddrsForNode(_ context.Context,
-	[]net.Addr, error) {
+	nodePub *btcec.PublicKey) (bool, []net.Addr, error) {
 
 	if m.failQuery {
 		return false, nil, fmt.Errorf("fail")
@@ -120,7 +121,8 @@ func TestFetchBackupForChan(t *testing.T) {
 	}
 	for i, testCase := range testCases {
 		_, err := FetchBackupForChan(
-			testCase.chanPoint, chanSource, chanSource,
+			context.Background(), testCase.chanPoint, chanSource,
+			chanSource,
 		)
 		switch {
 		// If this is a valid test case, and we failed, then we'll
@@ -141,6 +143,7 @@ func TestFetchBackupForChan(t *testing.T) {
 // channel source for all channels and construct a Single for each channel.
 func TestFetchStaticChanBackups(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	// First, we'll make the set of channels that we want to seed the
 	// channel source with. Both channels will be fully populated in the
@@ -162,7 +165,7 @@ func TestFetchStaticChanBackups(t *testing.T) {
 	// With the channel source populated, we'll now attempt to create a set
 	// of backups for all the channels. This should succeed, as all items
 	// are populated within the channel source.
-	backups, err := FetchStaticChanBackups(chanSource, chanSource)
+	backups, err := FetchStaticChanBackups(ctx, chanSource, chanSource)
 	require.NoError(t, err, "unable to create chan back ups")
 
 	if len(backups) != numChans {
@@ -177,7 +180,7 @@ func TestFetchStaticChanBackups(t *testing.T) {
 	copy(n[:], randomChan2.IdentityPub.SerializeCompressed())
 	delete(chanSource.addrs, n)
 
-	_, err = FetchStaticChanBackups(chanSource, chanSource)
+	_, err = FetchStaticChanBackups(ctx, chanSource, chanSource)
 	if err == nil {
 		t.Fatalf("query with incomplete information should fail")
 	}
@@ -186,7 +189,7 @@ func TestFetchStaticChanBackups(t *testing.T) {
 	// source at all, then we'll fail as well.
 	chanSource = newMockChannelSource()
 	chanSource.failQuery = true
-	_, err = FetchStaticChanBackups(chanSource, chanSource)
+	_, err = FetchStaticChanBackups(ctx, chanSource, chanSource)
 	if err == nil {
 		t.Fatalf("query should fail")
 	}

chanbackup/pubsub.go

@@ -2,6 +2,7 @@ package chanbackup
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"net"
 	"os"
@@ -81,7 +82,8 @@ type ChannelNotifier interface {
 	// synchronization point to ensure that the chanbackup.SubSwapper does
 	// not miss any channel open or close events in the period between when
 	// it's created, and when it requests the channel subscription.
-	SubscribeChans(map[wire.OutPoint]struct{}) (*ChannelSubscription, error)
+	SubscribeChans(context.Context,
+		map[wire.OutPoint]struct{}) (*ChannelSubscription, error)
 }
 
 // SubSwapper subscribes to new updates to the open channel state, and then
@@ -119,8 +121,9 @@ type SubSwapper struct {
 // set of channels, and the required interfaces to be notified of new channel
 // updates, pack a multi backup, and swap the current best backup from its
 // storage location.
-func NewSubSwapper(startingChans []Single, chanNotifier ChannelNotifier,
+func NewSubSwapper(ctx context.Context, startingChans []Single,
-	keyRing keychain.KeyRing, backupSwapper Swapper) (*SubSwapper, error) {
+	chanNotifier ChannelNotifier, keyRing keychain.KeyRing,
+	backupSwapper Swapper) (*SubSwapper, error) {
 
 	// First, we'll subscribe to the latest set of channel updates given
 	// the set of channels we already know of.
@@ -128,7 +131,7 @@ func NewSubSwapper(startingChans []Single, chanNotifier ChannelNotifier,
 	for _, chanBackup := range startingChans {
 		knownChans[chanBackup.FundingOutpoint] = struct{}{}
 	}
-	chanEvents, err := chanNotifier.SubscribeChans(knownChans)
+	chanEvents, err := chanNotifier.SubscribeChans(ctx, knownChans)
 	if err != nil {
 		return nil, err
 	}

chanbackup/pubsub_test.go

@@ -1,6 +1,7 @@
 package chanbackup
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
@@ -62,8 +63,8 @@ func newMockChannelNotifier() *mockChannelNotifier {
 	}
 }
 
-func (m *mockChannelNotifier) SubscribeChans(chans map[wire.OutPoint]struct{}) (
+func (m *mockChannelNotifier) SubscribeChans(_ context.Context,
-	*ChannelSubscription, error) {
+	_ map[wire.OutPoint]struct{}) (*ChannelSubscription, error) {
 
 	if m.fail {
 		return nil, fmt.Errorf("fail")
@@ -80,6 +81,7 @@ func (m *mockChannelNotifier) SubscribeChans(chans map[wire.OutPoint]struct{}) (
 // channel subscription, then the entire sub-swapper will fail to start.
 func TestNewSubSwapperSubscribeFail(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	keyRing := &lnencrypt.MockKeyRing{}
 
@@ -88,10 +90,8 @@ func TestNewSubSwapperSubscribeFail(t *testing.T) {
 		fail: true,
 	}
 
-	_, err := NewSubSwapper(nil, &chanNotifier, keyRing, &swapper)
+	_, err := NewSubSwapper(ctx, nil, &chanNotifier, keyRing, &swapper)
-	if err == nil {
+	require.Error(t, err)
-		t.Fatalf("expected fail due to lack of subscription")
-	}
 }
 
 func assertExpectedBackupSwap(t *testing.T, swapper *mockSwapper,
@@ -158,7 +158,9 @@ func TestSubSwapperIdempotentStartStop(t *testing.T) {
 	var chanNotifier mockChannelNotifier
 
 	swapper := newMockSwapper(keyRing)
-	subSwapper, err := NewSubSwapper(nil, &chanNotifier, keyRing, swapper)
+	subSwapper, err := NewSubSwapper(
+		context.Background(), nil, &chanNotifier, keyRing, swapper,
+	)
 	require.NoError(t, err, "unable to init subSwapper")
 
 	if err := subSwapper.Start(); err != nil {
@@ -224,7 +226,8 @@ func TestSubSwapperUpdater(t *testing.T) {
 	// With our channel set created, we'll make a fresh sub swapper
 	// instance to begin our test.
 	subSwapper, err := NewSubSwapper(
-		initialChanSet, chanNotifier, keyRing, swapper,
+		context.Background(), initialChanSet, chanNotifier, keyRing,
+		swapper,
 	)
 	require.NoError(t, err, "unable to make swapper")
 	if err := subSwapper.Start(); err != nil {

channel_notifier.go

@@ -1,6 +1,7 @@
 package lnd
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/btcsuite/btcd/wire"
@@ -31,7 +32,8 @@ type channelNotifier struct {
 // the channel subscription.
 //
 // NOTE: This is part of the chanbackup.ChannelNotifier interface.
-func (c *channelNotifier) SubscribeChans(startingChans map[wire.OutPoint]struct{}) (
+func (c *channelNotifier) SubscribeChans(ctx context.Context,
+	startingChans map[wire.OutPoint]struct{}) (
 	*chanbackup.ChannelSubscription, error) {
 
 	ltndLog.Infof("Channel backup proxy channel notifier starting")
@@ -46,7 +48,7 @@ func (c *channelNotifier) SubscribeChans(startingChans map[wire.OutPoint]struct{
 	// confirmed channels.
 	sendChanOpenUpdate := func(newOrPendingChan *channeldb.OpenChannel) {
 		_, nodeAddrs, err := c.addrs.AddrsForNode(
-			newOrPendingChan.IdentityPub,
+			ctx, newOrPendingChan.IdentityPub,
 		)
 		if err != nil {
 			pub := newOrPendingChan.IdentityPub

channeldb/addr_source.go

@@ -1,6 +1,7 @@
 package channeldb
 
 import (
+	"context"
 	"errors"
 	"net"
 
@@ -13,7 +14,8 @@ type AddrSource interface {
 	// AddrsForNode returns all known addresses for the target node public
 	// key. The returned boolean must indicate if the given node is unknown
 	// to the backing source.
-	AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, error)
+	AddrsForNode(ctx context.Context,
+		nodePub *btcec.PublicKey) (bool, []net.Addr, error)
 }
 
 // multiAddrSource is an implementation of AddrSource which gathers all the
@@ -38,8 +40,8 @@ func NewMultiAddrSource(sources ...AddrSource) AddrSource {
 // node.
 //
 // NOTE: this implements the AddrSource interface.
-func (c *multiAddrSource) AddrsForNode(nodePub *btcec.PublicKey) (bool,
+func (c *multiAddrSource) AddrsForNode(ctx context.Context,
-	[]net.Addr, error) {
+	nodePub *btcec.PublicKey) (bool, []net.Addr, error) {
 
 	if len(c.sources) == 0 {
 		return false, nil, errors.New("no address sources")
@@ -55,7 +57,7 @@ func (c *multiAddrSource) AddrsForNode(nodePub *btcec.PublicKey) (bool,
 	// Iterate over all the address sources and query each one for the
 	// addresses it has for the node in question.
 	for _, src := range c.sources {
-		isKnown, addrs, err := src.AddrsForNode(nodePub)
+		isKnown, addrs, err := src.AddrsForNode(ctx, nodePub)
 		if err != nil {
 			return false, nil, err
 		}

channeldb/addr_source_test.go

@@ -1,6 +1,7 @@
 package channeldb
 
 import (
+	"context"
 	"net"
 	"testing"
 
@@ -19,6 +20,7 @@ var (
 // deduplicates the results of a set of AddrSource implementations.
 func TestMultiAddrSource(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	var pk1 = newTestPubKey(t)
 
@@ -35,12 +37,12 @@ func TestMultiAddrSource(t *testing.T) {
 		})
 
 		// Let source 1 know of 2 addresses (addr 1 and 2) for node 1.
-		src1.On("AddrsForNode", pk1).Return(
+		src1.On("AddrsForNode", ctx, pk1).Return(
 			true, []net.Addr{addr1, addr2}, nil,
 		).Once()
 
 		// Let source 2 know of 2 addresses (addr 2 and 3) for node 1.
-		src2.On("AddrsForNode", pk1).Return(
+		src2.On("AddrsForNode", ctx, pk1).Return(
 			true, []net.Addr{addr2, addr3}, nil,
 			[]net.Addr{addr2, addr3}, nil,
 		).Once()
@@ -51,7 +53,7 @@ func TestMultiAddrSource(t *testing.T) {
 
 		// Query it for the addresses known for node 1. The results
 		// should contain addr 1, 2 and 3.
-		known, addrs, err := multiSrc.AddrsForNode(pk1)
+		known, addrs, err := multiSrc.AddrsForNode(ctx, pk1)
 		require.NoError(t, err)
 		require.True(t, known)
 		require.ElementsMatch(t, addrs, []net.Addr{addr1, addr2, addr3})
@@ -70,10 +72,10 @@ func TestMultiAddrSource(t *testing.T) {
 		})
 
 		// Let source 1 know of address 1 for node 1.
-		src1.On("AddrsForNode", pk1).Return(
+		src1.On("AddrsForNode", ctx, pk1).Return(
 			true, []net.Addr{addr1}, nil,
 		).Once()
-		src2.On("AddrsForNode", pk1).Return(false, nil, nil).Once()
+		src2.On("AddrsForNode", ctx, pk1).Return(false, nil, nil).Once()
 
 		// Create a multi-addr source that consists of both source 1
 		// and 2.
@@ -81,7 +83,7 @@ func TestMultiAddrSource(t *testing.T) {
 
 		// Query it for the addresses known for node 1. The results
 		// should contain addr 1.
-		known, addrs, err := multiSrc.AddrsForNode(pk1)
+		known, addrs, err := multiSrc.AddrsForNode(ctx, pk1)
 		require.NoError(t, err)
 		require.True(t, known)
 		require.ElementsMatch(t, addrs, []net.Addr{addr1})
@@ -103,13 +105,13 @@ func TestMultiAddrSource(t *testing.T) {
 		// and 2. Neither source known of node 1.
 		multiSrc := NewMultiAddrSource(src1, src2)
 
-		src1.On("AddrsForNode", pk1).Return(false, nil, nil).Once()
+		src1.On("AddrsForNode", ctx, pk1).Return(false, nil, nil).Once()
-		src2.On("AddrsForNode", pk1).Return(false, nil, nil).Once()
+		src2.On("AddrsForNode", ctx, pk1).Return(false, nil, nil).Once()
 
 		// Query it for the addresses known for node 1. It should return
 		// false to indicate that the node is unknown to all backing
 		// sources.
-		known, addrs, err := multiSrc.AddrsForNode(pk1)
+		known, addrs, err := multiSrc.AddrsForNode(ctx, pk1)
 		require.NoError(t, err)
 		require.False(t, known)
 		require.Empty(t, addrs)
@@ -127,10 +129,10 @@ func newMockAddrSource(t *testing.T) *mockAddrSource {
 	return &mockAddrSource{t: t}
 }
 
-func (m *mockAddrSource) AddrsForNode(pub *btcec.PublicKey) (bool, []net.Addr,
+func (m *mockAddrSource) AddrsForNode(ctx context.Context,
-	error) {
+	pub *btcec.PublicKey) (bool, []net.Addr, error) {
 
-	args := m.Called(pub)
+	args := m.Called(ctx, pub)
 	if args.Get(1) == nil {
 		return args.Bool(0), nil, args.Error(2)
 	}

channeldb/db.go

@@ -2,6 +2,7 @@ package channeldb
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
 	"fmt"
 	"net"
@@ -1532,7 +1533,9 @@ func (c *ChannelStateDB) RestoreChannelShells(channelShells ...*ChannelShell) er
 // unknown to the channel DB or not.
 //
 // NOTE: this is part of the AddrSource interface.
-func (d *DB) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, error) {
+func (d *DB) AddrsForNode(_ context.Context, nodePub *btcec.PublicKey) (bool,
+	[]net.Addr, error) {
+
 	linkNode, err := d.channelStateDB.linkNodeDB.FetchLinkNode(nodePub)
 	// Only if the error is something other than ErrNodeNotFound do we
 	// return it.

channeldb/db_test.go

@@ -1,6 +1,7 @@
 package channeldb
 
 import (
+	"context"
 	"image/color"
 	"math"
 	"math/rand"
@@ -180,6 +181,7 @@ func TestFetchClosedChannelForID(t *testing.T) {
 // channel db and graph db.
 func TestMultiSourceAddrsForNode(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	fullDB, err := MakeTestDB(t)
 	require.NoError(t, err, "unable to make test database")
@@ -194,7 +196,7 @@ func TestMultiSourceAddrsForNode(t *testing.T) {
 	testNode := createTestVertex(t)
 	nodePub, err := testNode.PubKey()
 	require.NoError(t, err)
-	graph.On("AddrsForNode", nodePub).Return(
+	graph.On("AddrsForNode", ctx, nodePub).Return(
 		true, []net.Addr{testAddr}, nil,
 	).Once()
 
@@ -212,7 +214,7 @@ func TestMultiSourceAddrsForNode(t *testing.T) {
 
 	// Now that we've created a link node, as well as a vertex for the
 	// node, we'll query for all its addresses.
-	known, nodeAddrs, err := addrSource.AddrsForNode(nodePub)
+	known, nodeAddrs, err := addrSource.AddrsForNode(ctx, nodePub)
 	require.NoError(t, err, "unable to obtain node addrs")
 	require.True(t, known)
 

discovery/gossiper.go

@@ -2022,7 +2022,7 @@ func (d *AuthenticatedGossiper) fetchPKScript(chanID *lnwire.ShortChannelID) (
 
 // addNode processes the given node announcement, and adds it to our channel
 // graph.
-func (d *AuthenticatedGossiper) addNode(_ context.Context,
+func (d *AuthenticatedGossiper) addNode(ctx context.Context,
 	msg *lnwire.NodeAnnouncement, op ...batch.SchedulerOption) error {
 
 	if err := netann.ValidateNodeAnn(msg); err != nil {
@@ -2030,7 +2030,9 @@ func (d *AuthenticatedGossiper) addNode(_ context.Context,
 			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
@@ -2204,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
 	}
@@ -2441,7 +2443,7 @@ func (d *AuthenticatedGossiper) handleNodeAnnouncement(ctx context.Context,
 
 	// We'll quickly ask the router if it already has a newer update for
 	// this node so we can skip validating signatures if not required.
-	if d.cfg.Graph.IsStaleNode(nodeAnn.NodeID, timestamp) {
+	if d.cfg.Graph.IsStaleNode(ctx, nodeAnn.NodeID, timestamp) {
 		log.Debugf("Skipped processing stale node: %x", nodeAnn.NodeID)
 		nMsg.err <- nil
 		return nil, true

discovery/gossiper_test.go

@@ -109,7 +109,7 @@ func newMockRouter(t *testing.T, height uint32) *mockGraphSource {
 
 var _ graph.ChannelGraphSource = (*mockGraphSource)(nil)
 
-func (r *mockGraphSource) AddNode(node *models.LightningNode,
+func (r *mockGraphSource) AddNode(_ context.Context, node *models.LightningNode,
 	_ ...batch.SchedulerOption) error {
 
 	r.mu.Lock()
@@ -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 {
@@ -308,7 +308,9 @@ func (r *mockGraphSource) FetchLightningNode(
 
 // IsStaleNode returns true if the graph source has a node announcement for the
 // target node with a more recent timestamp.
-func (r *mockGraphSource) IsStaleNode(nodePub route.Vertex, timestamp time.Time) bool {
+func (r *mockGraphSource) IsStaleNode(_ context.Context,
+	nodePub route.Vertex, timestamp time.Time) bool {
+
 	r.mu.Lock()
 	defer r.mu.Unlock()
 

graph/builder.go

@@ -1,6 +1,7 @@
 package graph
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -864,7 +865,7 @@ func (b *Builder) updateGraphWithClosedChannels(
 // timestamp. ErrIgnored will be returned if we already have the node, and
 // ErrOutdated will be returned if we have a timestamp that's after the new
 // timestamp.
-func (b *Builder) assertNodeAnnFreshness(node route.Vertex,
+func (b *Builder) assertNodeAnnFreshness(ctx context.Context, node route.Vertex,
 	msgTimestamp time.Time) error {
 
 	// If we are not already aware of this node, it means that we don't
@@ -872,7 +873,7 @@ func (b *Builder) assertNodeAnnFreshness(node route.Vertex,
 	// node announcements, we will ignore such nodes. If we do know about
 	// this node, check that this update brings info newer than what we
 	// already have.
-	lastUpdate, exists, err := b.cfg.Graph.HasLightningNode(node)
+	lastUpdate, exists, err := b.cfg.Graph.HasLightningNode(ctx, node)
 	if err != nil {
 		return errors.Errorf("unable to query for the "+
 			"existence of node: %v", err)
@@ -972,10 +973,10 @@ func (b *Builder) ApplyChannelUpdate(msg *lnwire.ChannelUpdate1) bool {
 // be ignored.
 //
 // 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 {
 
-	err := b.addNode(node, op...)
+	err := b.addNode(ctx, node, op...)
 	if err != nil {
 		logNetworkMsgProcessError(err)
 
@@ -989,18 +990,18 @@ func (b *Builder) AddNode(node *models.LightningNode,
 // 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
 // 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 {
 
 	// Before we add the node to the database, we'll check to see if the
 	// announcement is "fresh" or not. If it isn't, then we'll return an
 	// error.
-	err := b.assertNodeAnnFreshness(node.PubKeyBytes, node.LastUpdate)
+	err := b.assertNodeAnnFreshness(ctx, node.PubKeyBytes, node.LastUpdate)
 	if err != nil {
 		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 "+
 			"graph: %v", node.PubKeyBytes, err)
 	}
@@ -1264,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
@@ -1305,12 +1306,12 @@ func (b *Builder) AddProof(chanID lnwire.ShortChannelID,
 // target node with a more recent timestamp.
 //
 // NOTE: This method is part of the ChannelGraphSource interface.
-func (b *Builder) IsStaleNode(node route.Vertex,
+func (b *Builder) IsStaleNode(ctx context.Context, node route.Vertex,
 	timestamp time.Time) bool {
 
 	// If our attempt to assert that the node announcement is fresh fails,
 	// then we know that this is actually a stale announcement.
-	err := b.assertNodeAnnFreshness(node, timestamp)
+	err := b.assertNodeAnnFreshness(ctx, node, timestamp)
 	if err != nil {
 		log.Debugf("Checking stale node %x got %v", node, err)
 		return true

graph/builder_test.go

@@ -2,6 +2,7 @@ package graph
 
 import (
 	"bytes"
+	"context"
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
@@ -106,7 +107,7 @@ func TestIgnoreNodeAnnouncement(t *testing.T) {
 	}
 	copy(node.PubKeyBytes[:], pub.SerializeCompressed())
 
-	err := ctx.builder.AddNode(node)
+	err := ctx.builder.AddNode(context.Background(), node)
 	if !IsError(err, ErrIgnored) {
 		t.Fatalf("expected to get ErrIgnore, instead got: %v", err)
 	}
@@ -1012,6 +1013,7 @@ func testPruneChannelGraphDoubleDisabled(t *testing.T, assumeValid bool) {
 // node announcements.
 func TestIsStaleNode(t *testing.T) {
 	t.Parallel()
+	ctxb := context.Background()
 
 	const startingBlockHeight = 101
 	ctx := createTestCtxSingleNode(t, startingBlockHeight)
@@ -1052,7 +1054,7 @@ func TestIsStaleNode(t *testing.T) {
 	// Before we add the node, if we query for staleness, we should get
 	// false, as we haven't added the full node.
 	updateTimeStamp := time.Unix(123, 0)
-	if ctx.builder.IsStaleNode(pub1, updateTimeStamp) {
+	if ctx.builder.IsStaleNode(ctxb, pub1, updateTimeStamp) {
 		t.Fatalf("incorrectly detected node as stale")
 	}
 
@@ -1068,20 +1070,20 @@ func TestIsStaleNode(t *testing.T) {
 		Features:             testFeatures,
 	}
 	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)
 	}
 
 	// If we use the same timestamp and query for staleness, we should get
 	// true.
-	if !ctx.builder.IsStaleNode(pub1, updateTimeStamp) {
+	if !ctx.builder.IsStaleNode(ctxb, pub1, updateTimeStamp) {
 		t.Fatalf("failure to detect stale node update")
 	}
 
 	// If we update the timestamp and once again query for staleness, it
 	// should report false.
 	newTimeStamp := time.Unix(1234, 0)
-	if ctx.builder.IsStaleNode(pub1, newTimeStamp) {
+	if ctx.builder.IsStaleNode(ctxb, pub1, newTimeStamp) {
 		t.Fatalf("incorrectly detected node as stale")
 	}
 }
@@ -1328,6 +1330,8 @@ func createTestCtxFromFile(t *testing.T,
 func parseTestGraph(t *testing.T, useCache bool, path string) (
 	*testGraphInstance, error) {
 
+	ctx := context.Background()
+
 	graphJSON, err := os.ReadFile(path)
 	if err != nil {
 		return nil, err
@@ -1434,7 +1438,7 @@ func parseTestGraph(t *testing.T, useCache bool, path string) (
 
 		// With the node fully parsed, add it as a vertex within the
 		// graph.
-		if err := graph.AddLightningNode(dbNode); err != nil {
+		if err := graph.AddLightningNode(ctx, dbNode); err != nil {
 			return nil, err
 		}
 	}
@@ -1715,6 +1719,8 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
 	testChannels []*testChannel, source string) (*testGraphInstance,
 	error) {
 
+	ctx := context.Background()
+
 	// 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
 	// need to be unique.
@@ -1763,7 +1769,7 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
 
 		// With the node fully parsed, add it as a vertex within the
 		// graph.
-		if err := graph.AddLightningNode(dbNode); err != nil {
+		if err := graph.AddLightningNode(ctx, dbNode); err != nil {
 			return nil, err
 		}
 

graph/db/graph.go

@@ -1,6 +1,7 @@
 package graphdb
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"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
 // already present node from a node announcement, or to insert a node found in a
 // 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 {
 		return err
 	}
@@ -285,8 +286,10 @@ func (c *ChannelGraph) AddLightningNode(node *models.LightningNode,
 
 // DeleteLightningNode starts a new database transaction to remove a vertex/node
 // from the database according to the node's public key.
-func (c *ChannelGraph) DeleteLightningNode(nodePub route.Vertex) error {
+func (c *ChannelGraph) DeleteLightningNode(ctx context.Context,
-	err := c.V1Store.DeleteLightningNode(nodePub)
+	nodePub route.Vertex) error {
+
+	err := c.V1Store.DeleteLightningNode(ctx, nodePub)
 	if err != nil {
 		return err
 	}

graph/db/graph_test.go

@@ -2,6 +2,7 @@ package graphdb
 
 import (
 	"bytes"
+	"context"
 	"crypto/sha256"
 	"encoding/hex"
 	"errors"
@@ -99,6 +100,7 @@ func createTestVertex(t testing.TB) *models.LightningNode {
 
 func TestNodeInsertionAndDeletion(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraphNew(t)
 
@@ -123,17 +125,17 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 	// First, insert the node into the graph DB. This should succeed
 	// without any errors.
 	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)
 	}
 	assertNodeInCache(t, graph, node, testFeatures)
 
 	// 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)
+	_, exists, err := graph.HasLightningNode(ctx, dbNode.PubKeyBytes)
 	if err != nil {
 		t.Fatalf("unable to query for node: %v", err)
 	} else if !exists {
@@ -157,18 +159,19 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 
 	// Next, delete the node from the graph, this should purge all data
 	// related to the node.
-	require.NoError(t, graph.DeleteLightningNode(testPub))
+	require.NoError(t, graph.DeleteLightningNode(ctx, testPub))
 	assertNodeNotInCache(t, graph, testPub)
 
 	// Attempting to delete the node again should return an error since
 	// the node is no longer known.
 	require.ErrorIs(
-		t, graph.DeleteLightningNode(testPub), ErrGraphNodeNotFound,
+		t, graph.DeleteLightningNode(ctx, testPub),
+		ErrGraphNodeNotFound,
 	)
 
 	// 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
@@ -180,21 +183,21 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 
 	// Initially, the node is unknown to the graph and there should be no
 	// addresses for it.
-	known, addrs, err := graph.AddrsForNode(pub)
+	known, addrs, err := graph.AddrsForNode(ctx, pub)
 	require.NoError(t, err)
 	require.False(t, known)
 	require.Empty(t, addrs)
 
 	// Add the node without any addresses.
 	node = nodeWithAddrs(nil)
-	require.NoError(t, graph.AddLightningNode(node))
+	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)
 
-	known, addrs, err = graph.AddrsForNode(pub)
+	known, addrs, err = graph.AddrsForNode(ctx, pub)
 	require.NoError(t, err)
 	require.True(t, known)
 	require.Empty(t, addrs)
@@ -214,14 +217,14 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 		testOpaqueAddr,
 	}
 	node = nodeWithAddrs(expAddrs)
-	require.NoError(t, graph.AddLightningNode(node))
+	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)
 
-	known, addrs, err = graph.AddrsForNode(pub)
+	known, addrs, err = graph.AddrsForNode(ctx, pub)
 	require.NoError(t, err)
 	require.True(t, known)
 	require.EqualValues(t, expAddrs, addrs)
@@ -235,10 +238,10 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 		testIPV6Addr,
 	}
 	node = nodeWithAddrs(expAddrs)
-	require.NoError(t, graph.AddLightningNode(node))
+	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)
 
@@ -248,10 +251,10 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 		testOnionV3Addr,
 	}
 	node = nodeWithAddrs(expAddrs)
-	require.NoError(t, graph.AddLightningNode(node))
+	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)
 }
@@ -260,6 +263,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)
 
@@ -280,12 +284,12 @@ 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)
+	_, exists, err := graph.HasLightningNode(ctx, dbNode1.PubKeyBytes)
 	require.NoError(t, err)
 	require.True(t, exists)
 
@@ -299,7 +303,7 @@ func TestPartialNode(t *testing.T) {
 	}
 	compareNodes(t, expectedNode1, dbNode1)
 
-	_, exists, err = graph.HasLightningNode(dbNode2.PubKeyBytes)
+	_, exists, err = graph.HasLightningNode(ctx, dbNode2.PubKeyBytes)
 	require.NoError(t, err)
 	require.True(t, exists)
 
@@ -315,18 +319,19 @@ func TestPartialNode(t *testing.T) {
 
 	// Next, delete the node from the graph, this should purge all data
 	// related to the node.
-	require.NoError(t, graph.DeleteLightningNode(pubKey1))
+	require.NoError(t, graph.DeleteLightningNode(ctx, pubKey1))
 	assertNodeNotInCache(t, graph, testPub)
 
 	// 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)
 }
 
 // TestAliasLookup tests the alias lookup functionality of the graph store.
 func TestAliasLookup(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraphNew(t)
 
@@ -336,7 +341,7 @@ func TestAliasLookup(t *testing.T) {
 
 	// Add the node to the graph's database, this should also insert an
 	// 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
 	// the one which the test node was assigned.
@@ -801,18 +806,19 @@ func createChannelEdge(node1, node2 *models.LightningNode,
 
 func TestEdgeInfoUpdates(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
 	// We'd like to test the update of edges inserted into the database, so
 	// we create two vertexes to connect.
 	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)
 	}
 	assertNodeInCache(t, graph, node1, testFeatures)
 	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)
 	}
 	assertNodeInCache(t, graph, node2, testFeatures)
@@ -1483,6 +1489,8 @@ func TestGraphCacheTraversal(t *testing.T) {
 func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes,
 	numChannels int) (map[uint64]struct{}, []*models.LightningNode) {
 
+	ctx := context.Background()
+
 	nodes := make([]*models.LightningNode, numNodes)
 	nodeIndex := map[string]struct{}{}
 	for i := 0; i < numNodes; i++ {
@@ -1495,7 +1503,7 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes,
 	// Add each of the nodes into the graph, they should be inserted
 	// without error.
 	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
@@ -1671,6 +1679,7 @@ func assertChanViewEqualChanPoints(t *testing.T, a []EdgePoint,
 
 func TestGraphPruning(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
@@ -1687,7 +1696,7 @@ func TestGraphPruning(t *testing.T) {
 	for i := 0; i < numNodes; i++ {
 		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)
 		}
 
@@ -1919,6 +1928,7 @@ func TestHighestChanID(t *testing.T) {
 // insertion of a new edge, the edge update index is updated properly.
 func TestChanUpdatesInHorizon(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
@@ -1935,11 +1945,11 @@ func TestChanUpdatesInHorizon(t *testing.T) {
 
 	// We'll start by creating two nodes which will seed our test graph.
 	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)
 	}
 	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)
 	}
 
@@ -2077,6 +2087,7 @@ func TestChanUpdatesInHorizon(t *testing.T) {
 // the most recent node updates within a particular time horizon.
 func TestNodeUpdatesInHorizon(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraphNew(t)
 
@@ -2108,7 +2119,7 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
 
 		nodeAnns = append(nodeAnns, *nodeAnn)
 
-		require.NoError(t, graph.AddLightningNode(nodeAnn))
+		require.NoError(t, graph.AddLightningNode(ctx, nodeAnn))
 	}
 
 	queryCases := []struct {
@@ -2245,6 +2256,7 @@ func TestFilterKnownChanIDsZombieRevival(t *testing.T) {
 // know of on disk.
 func TestFilterKnownChanIDs(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
@@ -2277,11 +2289,11 @@ func TestFilterKnownChanIDs(t *testing.T) {
 
 	// We'll start by creating two nodes which will seed our test graph.
 	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)
 	}
 	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)
 	}
 
@@ -2422,14 +2434,15 @@ func TestFilterKnownChanIDs(t *testing.T) {
 // methods that acquire the cache mutex along with the DB mutex.
 func TestStressTestChannelGraphAPI(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
 	node1 := createTestVertex(t)
-	require.NoError(t, graph.AddLightningNode(node1))
+	require.NoError(t, graph.AddLightningNode(ctx, node1))
 
 	node2 := createTestVertex(t)
-	require.NoError(t, graph.AddLightningNode(node2))
+	require.NoError(t, graph.AddLightningNode(ctx, node2))
 
 	require.NoError(t, graph.SetSourceNode(node1))
 
@@ -2706,16 +2719,17 @@ func TestStressTestChannelGraphAPI(t *testing.T) {
 // set of short channel ID's for a given block range.
 func TestFilterChannelRange(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
 	// We'll first populate our graph with two nodes. All channels created
 	// below will be made between these two nodes.
 	node1 := createTestVertex(t)
-	require.NoError(t, graph.AddLightningNode(node1))
+	require.NoError(t, graph.AddLightningNode(ctx, node1))
 
 	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
 	// inserted, we should get an empty slice of results.
@@ -2922,17 +2936,18 @@ func TestFilterChannelRange(t *testing.T) {
 // of ChannelEdge structs for a given set of short channel ID's.
 func TestFetchChanInfos(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
 	// We'll first populate our graph with two nodes. All channels created
 	// below will be made between these two nodes.
 	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)
 	}
 	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)
 	}
 
@@ -3024,16 +3039,17 @@ func TestFetchChanInfos(t *testing.T) {
 // both sides.
 func TestIncompleteChannelPolicies(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
 	// Create two nodes.
 	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)
 	}
 	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)
 	}
 
@@ -3119,6 +3135,7 @@ func TestIncompleteChannelPolicies(t *testing.T) {
 // up.
 func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
@@ -3136,11 +3153,11 @@ func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) {
 	// We'll first populate our graph with two nodes. All channels created
 	// below will be made between these two nodes.
 	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)
 	}
 	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)
 	}
 
@@ -3266,6 +3283,7 @@ func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) {
 // PruneSyncState method.
 func TestPruneGraphNodes(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
@@ -3280,15 +3298,15 @@ func TestPruneGraphNodes(t *testing.T) {
 	// channel graph, at the end of the scenario, only two of these nodes
 	// should still be in the graph.
 	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)
 	}
 	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)
 	}
 	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)
 	}
 
@@ -3322,7 +3340,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)
 }
 
@@ -3331,6 +3349,7 @@ func TestPruneGraphNodes(t *testing.T) {
 // database, then shell edges are created for each node if needed.
 func TestAddChannelEdgeShellNodes(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraphNew(t)
 
@@ -3347,11 +3366,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)
 
@@ -3361,7 +3380,7 @@ func TestAddChannelEdgeShellNodes(t *testing.T) {
 	require.ErrorIs(t, err, ErrEdgeAlreadyExist)
 
 	// 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
@@ -3369,13 +3388,14 @@ func TestAddChannelEdgeShellNodes(t *testing.T) {
 // well.
 func TestNodePruningUpdateIndexDeletion(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraphNew(t)
 
 	// We'll first populate our graph with a single node that will be
 	// removed shortly.
 	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)
 	}
 
@@ -3397,9 +3417,8 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
 
 	// We'll now delete the node from the graph, this should result in it
 	// being removed from the update index as well.
-	if err := graph.DeleteLightningNode(node1.PubKeyBytes); err != nil {
+	err = graph.DeleteLightningNode(ctx, node1.PubKeyBytes)
-		t.Fatalf("unable to delete node: %v", err)
+	require.NoError(t, err)
-	}
 
 	// Now that the node has been deleted, we'll again query the nodes in
 	// the horizon. This time we should have no nodes at all.
@@ -3416,6 +3435,7 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
 // public within the network graph.
 func TestNodeIsPublic(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	// We'll start off the test by creating a small network of 3
 	// participants with the following graph:
@@ -3453,9 +3473,8 @@ func TestNodeIsPublic(t *testing.T) {
 	graphs := []*ChannelGraph{aliceGraph, bobGraph, carolGraph}
 	for _, graph := range graphs {
 		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 {
 			if err := graph.AddChannelEdge(edge); err != nil {
@@ -3551,24 +3570,25 @@ func TestNodeIsPublic(t *testing.T) {
 // DisabledChannelIDs is correct.
 func TestDisabledChannelIDs(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
 	// Create first node and add it to the graph.
 	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)
 	}
 
 	// Create second node and add it to the graph.
 	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)
 	}
 
 	// Adding a new channel edge to the graph.
 	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)
 	}
 
@@ -3634,6 +3654,7 @@ func TestDisabledChannelIDs(t *testing.T) {
 // receive the proper update.
 func TestEdgePolicyMissingMaxHtcl(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
@@ -3646,13 +3667,13 @@ func TestEdgePolicyMissingMaxHtcl(t *testing.T) {
 	// We'd like to test the update of edges inserted into the database, so
 	// we create two vertexes to connect.
 	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)
 	}
 	node2 := createTestVertex(t)
 
 	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)
 	}
 	if err := graph.AddChannelEdge(edgeInfo); err != nil {
@@ -4037,15 +4058,16 @@ func TestBatchedAddChannelEdge(t *testing.T) {
 // executes multiple UpdateEdgePolicy requests in a single txn.
 func TestBatchedUpdateEdgePolicy(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraphNew(t)
 
 	// We'd like to test the update of edges inserted into the database, so
 	// we create two vertexes to connect.
 	node1 := createTestVertex(t)
-	require.NoError(t, graph.AddLightningNode(node1))
+	require.NoError(t, graph.AddLightningNode(ctx, node1))
 	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.
 	edgeInfo, edge1, edge2 := createChannelEdge(node1, node2)
@@ -4139,6 +4161,7 @@ func BenchmarkForEachChannel(b *testing.B) {
 // method works as expected, and is able to handle nil self edges.
 func TestGraphCacheForEachNodeChannel(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
@@ -4147,9 +4170,9 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) {
 	graph.graphCache = nil
 
 	node1 := createTestVertex(t)
-	require.NoError(t, graph.AddLightningNode(node1))
+	require.NoError(t, graph.AddLightningNode(ctx, node1))
 	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.
 	edgeInfo, e1, e2 := createChannelEdge(node1, node2)
@@ -4305,6 +4328,7 @@ var testNodeAnn = "01012674c2e7ef68c73a086b7de2603f4ef1567358df84bb4edaa06c" +
 // that the two messages are equal.
 func TestLightningNodePersistence(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	// Create a new test graph instance.
 	graph := MakeTestGraphNew(t)
@@ -4323,11 +4347,11 @@ func TestLightningNodePersistence(t *testing.T) {
 	node := models.NodeFromWireAnnouncement(na)
 
 	// Persist the node to disk.
-	err = graph.AddLightningNode(node)
+	err = graph.AddLightningNode(ctx, node)
 	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.

graph/db/interfaces.go

@@ -1,6 +1,7 @@
 package graphdb
 
 import (
+	"context"
 	"net"
 	"time"
 
@@ -56,13 +57,14 @@ type V1Store interface { //nolint:interfacebloat
 	// update that node's 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 channel update.
-	AddLightningNode(node *models.LightningNode,
+	AddLightningNode(ctx context.Context, node *models.LightningNode,
 		op ...batch.SchedulerOption) error
 
 	// AddrsForNode returns all known addresses for the target node public
 	// key that the graph DB is aware of. The returned boolean indicates if
 	// the given node is unknown to the graph DB or not.
-	AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, error)
+	AddrsForNode(ctx context.Context,
+		nodePub *btcec.PublicKey) (bool, []net.Addr, error)
 
 	// ForEachSourceNodeChannel iterates through all channels of the source
 	// node, executing the passed callback on each. The call-back is
@@ -113,7 +115,7 @@ type V1Store interface { //nolint:interfacebloat
 
 	// DeleteLightningNode starts a new database transaction to remove a
 	// vertex/node from the database according to the node's public key.
-	DeleteLightningNode(nodePub route.Vertex) error
+	DeleteLightningNode(ctx context.Context, nodePub route.Vertex) error
 
 	// NodeUpdatesInHorizon returns all the known lightning node which have
 	// an update timestamp within the passed range. This method can be used
@@ -125,16 +127,16 @@ 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) (
+	FetchLightningNode(ctx context.Context,
-		*models.LightningNode, error)
+		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
 	// database, a timestamp of when the data for the node was lasted
 	// updated is returned along with a true boolean. Otherwise, an empty
 	// time.Time is returned with a false boolean.
-	HasLightningNode(nodePub [33]byte) (time.Time, bool,
+	HasLightningNode(ctx context.Context, nodePub [33]byte) (time.Time,
-		error)
+		bool, error)
 
 	// IsPublicNode is a helper method that determines whether the node with
 	// the given public key is seen as a public node in the graph from the

graph/db/kv_store.go

@@ -376,15 +376,15 @@ func initKVStore(db kvdb.Backend) error {
 // unknown to the graph DB or not.
 //
 // NOTE: this is part of the channeldb.AddrSource interface.
-func (c *KVStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr,
+func (c *KVStore) AddrsForNode(ctx context.Context,
-	error) {
+	nodePub *btcec.PublicKey) (bool, []net.Addr, error) {
 
 	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):
@@ -957,10 +957,8 @@ func (c *KVStore) SetSourceNode(node *models.LightningNode) error {
 // channel update.
 //
 // 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]{
 		Opts: batch.NewSchedulerOptions(opts...),
@@ -1032,7 +1030,9 @@ func (c *KVStore) LookupAlias(pub *btcec.PublicKey) (string, error) {
 
 // DeleteLightningNode starts a new database transaction to remove a vertex/node
 // from the database according to the node's public key.
-func (c *KVStore) DeleteLightningNode(nodePub route.Vertex) error {
+func (c *KVStore) DeleteLightningNode(_ context.Context,
+	nodePub route.Vertex) error {
+
 	// TODO(roasbeef): ensure dangling edges are removed...
 	return kvdb.Update(c.db, func(tx kvdb.RwTx) error {
 		nodes := tx.ReadWriteBucket(nodeBucket)
@@ -2997,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) (
+func (c *KVStore) FetchLightningNode(_ context.Context,
-	*models.LightningNode, error) {
+	nodePub route.Vertex) (*models.LightningNode, error) {
 
 	return c.fetchLightningNode(nil, nodePub)
 }
@@ -3065,8 +3065,8 @@ func (c *KVStore) fetchLightningNode(tx kvdb.RTx,
 // timestamp of when the data for the node was lasted updated is returned along
 // with a true boolean. Otherwise, an empty time.Time is returned with a false
 // boolean.
-func (c *KVStore) HasLightningNode(nodePub [33]byte) (time.Time, bool,
+func (c *KVStore) HasLightningNode(_ context.Context,
-	error) {
+	nodePub [33]byte) (time.Time, bool, error) {
 
 	var (
 		updateTime time.Time

graph/db/sql_store.go

@@ -174,10 +174,8 @@ func NewSQLStore(cfg *SQLStoreConfig, db BatchedSQLQueries, kvStore *KVStore,
 // information.
 //
 // 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]{
 		Opts: batch.NewSchedulerOptions(opts...),
@@ -195,10 +193,8 @@ func (s *SQLStore) AddLightningNode(node *models.LightningNode,
 // returned.
 //
 // NOTE: part of the V1Store interface.
-func (s *SQLStore) FetchLightningNode(pubKey route.Vertex) (
+func (s *SQLStore) FetchLightningNode(ctx context.Context,
-	*models.LightningNode, error) {
+	pubKey route.Vertex) (*models.LightningNode, error) {
-
-	ctx := context.TODO()
 
 	var node *models.LightningNode
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
@@ -221,10 +217,8 @@ func (s *SQLStore) FetchLightningNode(pubKey route.Vertex) (
 // boolean.
 //
 // NOTE: part of the V1Store interface.
-func (s *SQLStore) HasLightningNode(pubKey [33]byte) (time.Time, bool,
+func (s *SQLStore) HasLightningNode(ctx context.Context,
-	error) {
+	pubKey [33]byte) (time.Time, bool, error) {
-
-	ctx := context.TODO()
 
 	var (
 		exists     bool
@@ -264,10 +258,8 @@ func (s *SQLStore) HasLightningNode(pubKey [33]byte) (time.Time, bool,
 // given node is unknown to the graph DB or not.
 //
 // NOTE: part of the V1Store interface.
-func (s *SQLStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr,
+func (s *SQLStore) AddrsForNode(ctx context.Context,
-	error) {
+	nodePub *btcec.PublicKey) (bool, []net.Addr, error) {
-
-	ctx := context.TODO()
 
 	var (
 		addresses []net.Addr
@@ -297,8 +289,8 @@ func (s *SQLStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr,
 // from the database according to the node's public key.
 //
 // NOTE: part of the V1Store interface.
-func (s *SQLStore) DeleteLightningNode(pubKey route.Vertex) error {
+func (s *SQLStore) DeleteLightningNode(ctx context.Context,
-	ctx := context.TODO()
+	pubKey route.Vertex) error {
 
 	err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
 		res, err := db.DeleteNodeByPubKey(

graph/interfaces.go

@@ -1,6 +1,7 @@
 package graph
 
 import (
+	"context"
 	"time"
 
 	"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
 	// database. If the node with this pubkey is not present in an existing
 	// channel, it will be ignored.
-	AddNode(node *models.LightningNode,
+	AddNode(ctx context.Context, node *models.LightningNode,
 		op ...batch.SchedulerOption) error
 
 	// AddEdge is used to add edge/channel to the topology of the router,
@@ -45,7 +46,8 @@ type ChannelGraphSource interface {
 	// for the target node with a more recent timestamp. This method will
 	// also return true if we don't have an active channel announcement for
 	// the target node.
-	IsStaleNode(node route.Vertex, timestamp time.Time) bool
+	IsStaleNode(ctx context.Context, node route.Vertex,
+		timestamp time.Time) bool
 
 	// IsPublicNode determines whether the given vertex is seen as a public
 	// node in the graph from the graph's source node's point of view.
@@ -83,7 +85,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
@@ -202,7 +205,7 @@ type DB interface {
 	// update that node's 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 channel update.
-	AddLightningNode(node *models.LightningNode,
+	AddLightningNode(ctx context.Context, node *models.LightningNode,
 		op ...batch.SchedulerOption) error
 
 	// AddChannelEdge adds a new (undirected, blank) edge to the graph
@@ -236,13 +239,14 @@ type DB interface {
 	// database, a timestamp of when the data for the node was lasted
 	// updated is returned along with a true boolean. Otherwise, an empty
 	// time.Time is returned with a false boolean.
-	HasLightningNode(nodePub [33]byte) (time.Time, bool, error)
+	HasLightningNode(ctx context.Context, nodePub [33]byte) (time.Time,
+		bool, error)
 
 	// 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,
+	FetchLightningNode(ctx context.Context,
-		error)
+		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

graph/notifications_test.go

@@ -2,6 +2,7 @@ package graph
 
 import (
 	"bytes"
+	"context"
 	"encoding/hex"
 	"fmt"
 	"image/color"
@@ -608,6 +609,7 @@ func TestEdgeUpdateNotification(t *testing.T) {
 // attributes with new data.
 func TestNodeUpdateNotification(t *testing.T) {
 	t.Parallel()
+	ctxb := context.Background()
 
 	const startingBlockHeight = 101
 	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
 	// 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)
 	}
-	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)
 	}
 
@@ -763,7 +765,7 @@ func TestNodeUpdateNotification(t *testing.T) {
 	nodeUpdateAnn.LastUpdate = node1.LastUpdate.Add(300 * time.Millisecond)
 
 	// 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)
 	}
 
@@ -790,6 +792,7 @@ func TestNodeUpdateNotification(t *testing.T) {
 // when the client wishes to exit.
 func TestNotificationCancellation(t *testing.T) {
 	t.Parallel()
+	ctxb := context.Background()
 
 	const startingBlockHeight = 101
 	ctx := createTestCtxSingleNode(t, startingBlockHeight)
@@ -845,11 +848,11 @@ func TestNotificationCancellation(t *testing.T) {
 		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)
 	}
 
-	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)
 	}
 

lnrpc/devrpc/dev_server.go

@@ -260,7 +260,7 @@ func (s *Server) ImportGraph(ctx context.Context,
 			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",
 				rpcNode.PubKey, err)
 		}

routing/pathfind_test.go

@@ -2,6 +2,7 @@ package routing
 
 import (
 	"bytes"
+	"context"
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
@@ -181,6 +182,8 @@ func makeTestGraph(t *testing.T, useCache bool) (*graphdb.ChannelGraph,
 func parseTestGraph(t *testing.T, useCache bool, path string) (
 	*testGraphInstance, error) {
 
+	ctx := context.Background()
+
 	graphJSON, err := os.ReadFile(path)
 	if err != nil {
 		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
 		// graph.
-		if err := graph.AddLightningNode(dbNode); err != nil {
+		if err := graph.AddLightningNode(ctx, dbNode); err != nil {
 			return nil, err
 		}
 	}
@@ -515,6 +518,8 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
 	testChannels []*testChannel, source string,
 	sourceFeatureBits ...lnwire.FeatureBit) (*testGraphInstance, error) {
 
+	ctx := context.Background()
+
 	// 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
 	// 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
 		// graph.
-		if err := graph.AddLightningNode(dbNode); err != nil {
+		if err := graph.AddLightningNode(ctx, dbNode); err != nil {
 			return nil, err
 		}
 

routing/router_test.go

@@ -2,6 +2,7 @@ package routing
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"image/color"
 	"math"
@@ -2706,6 +2707,7 @@ func TestNewRouteRequest(t *testing.T) {
 // announcements for the channel vertexes to be able to use the channel.
 func TestAddEdgeUnknownVertexes(t *testing.T) {
 	t.Parallel()
+	ctxb := context.Background()
 
 	const startingBlockHeight = 101
 	ctx := createTestCtxFromFile(t, startingBlockHeight, basicGraphFilePath)
@@ -2717,11 +2719,11 @@ func TestAddEdgeUnknownVertexes(t *testing.T) {
 	copy(pub2[:], priv2.PubKey().SerializeCompressed())
 
 	// The two nodes we are about to add should not exist yet.
-	_, exists1, err := ctx.graph.HasLightningNode(pub1)
+	_, exists1, err := ctx.graph.HasLightningNode(ctxb, pub1)
 	require.NoError(t, err, "unable to query graph")
 	require.False(t, exists1)
 
-	_, exists2, err := ctx.graph.HasLightningNode(pub2)
+	_, exists2, err := ctx.graph.HasLightningNode(ctxb, pub2)
 	require.NoError(t, err, "unable to query graph")
 	require.False(t, exists2)
 
@@ -2777,11 +2779,11 @@ func TestAddEdgeUnknownVertexes(t *testing.T) {
 
 	// After adding the edge between the two previously unknown nodes, they
 	// should have been added to the graph.
-	_, exists1, err = ctx.graph.HasLightningNode(pub1)
+	_, exists1, err = ctx.graph.HasLightningNode(ctxb, pub1)
 	require.NoError(t, err, "unable to query graph")
 	require.True(t, exists1)
 
-	_, exists2, err = ctx.graph.HasLightningNode(pub2)
+	_, exists2, err = ctx.graph.HasLightningNode(ctxb, pub2)
 	require.NoError(t, err, "unable to query graph")
 	require.True(t, exists2)
 
@@ -2879,7 +2881,7 @@ func TestAddEdgeUnknownVertexes(t *testing.T) {
 	}
 	copy(n1.PubKeyBytes[:], priv1.PubKey().SerializeCompressed())
 
-	require.NoError(t, ctx.graph.AddLightningNode(n1))
+	require.NoError(t, ctx.graph.AddLightningNode(ctxb, n1))
 
 	n2 := &models.LightningNode{
 		HaveNodeAnnouncement: true,
@@ -2892,7 +2894,7 @@ func TestAddEdgeUnknownVertexes(t *testing.T) {
 	}
 	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
 	// updated.
@@ -2905,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)

rpcserver.go

@@ -1755,7 +1755,7 @@ func (r *rpcServer) VerifyMessage(ctx context.Context,
 	//
 	// TODO(phlip9): Require valid nodes to have capital in active channels.
 	graph := r.server.graphDB
-	_, active, err := graph.HasLightningNode(pub)
+	_, active, err := graph.HasLightningNode(ctx, pub)
 	if err != nil {
 		return nil, fmt.Errorf("failed to query graph: %w", err)
 	}
@@ -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
 		}
@@ -8136,7 +8136,7 @@ func (r *rpcServer) ExportChannelBackup(ctx context.Context,
 	// the database. If this channel has been closed, or the outpoint is
 	// unknown, then we'll return an error
 	unpackedBackup, err := chanbackup.FetchBackupForChan(
-		chanPoint, r.server.chanStateDB, r.server.addrSource,
+		ctx, chanPoint, r.server.chanStateDB, r.server.addrSource,
 	)
 	if err != nil {
 		return nil, err
@@ -8316,7 +8316,7 @@ func (r *rpcServer) ExportAllChannelBackups(ctx context.Context,
 	// First, we'll attempt to read back ups for ALL currently opened
 	// channels from disk.
 	allUnpackedBackups, err := chanbackup.FetchStaticChanBackups(
-		r.server.chanStateDB, r.server.addrSource,
+		ctx, r.server.chanStateDB, r.server.addrSource,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("unable to fetch all static chan "+
@@ -8451,7 +8451,8 @@ func (r *rpcServer) SubscribeChannelBackups(req *lnrpc.ChannelBackupSubscription
 			// we'll obtains the current set of single channel
 			// backups from disk.
 			chanBackups, err := chanbackup.FetchStaticChanBackups(
-				r.server.chanStateDB, r.server.addrSource,
+				updateStream.Context(), r.server.chanStateDB,
+				r.server.addrSource,
 			)
 			if err != nil {
 				return fmt.Errorf("unable to fetch all "+

server.go

@@ -555,7 +555,7 @@ func noiseDial(idKey keychain.SingleKeyECDH,
 // passed listener address.
 //
 //nolint:funlen
-func newServer(_ context.Context, cfg *Config, listenAddrs []net.Addr,
+func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
 	dbs *DatabaseInstances, cc *chainreg.ChainControl,
 	nodeKeyDesc *keychain.KeyDescriptor,
 	chansToRestore walletunlocker.ChannelsToRecover,
@@ -1715,13 +1715,13 @@ func newServer(_ context.Context, cfg *Config, listenAddrs []net.Addr,
 		cfg.BackupFilePath, cfg.NoBackupArchive,
 	)
 	startingChans, err := chanbackup.FetchStaticChanBackups(
-		s.chanStateDB, s.addrSource,
+		ctx, s.chanStateDB, s.addrSource,
 	)
 	if err != nil {
 		return nil, err
 	}
 	s.chanSubSwapper, err = chanbackup.NewSubSwapper(
-		startingChans, chanNotifier, s.cc.KeyRing, backupFile,
+		ctx, startingChans, chanNotifier, s.cc.KeyRing, backupFile,
 	)
 	if err != nil {
 		return nil, err
@@ -2664,6 +2664,8 @@ func (s *server) Stop() error {
 	s.stop.Do(func() {
 		atomic.StoreInt32(&s.stopping, 1)
 
+		ctx := context.Background()
+
 		close(s.quit)
 
 		// Shutdown connMgr first to prevent conns during shutdown.
@@ -2737,7 +2739,7 @@ func (s *server) Stop() error {
 		// Update channel.backup file. Make sure to do it before
 		// stopping chanSubSwapper.
 		singles, err := chanbackup.FetchStaticChanBackups(
-			s.chanStateDB, s.addrSource,
+			ctx, s.chanStateDB, s.addrSource,
 		)
 		if err != nil {
 			srvrLog.Warnf("failed to fetch channel states: %v",
@@ -4637,6 +4639,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 +4727,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 +5220,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
 	}