diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index e0baf4971..8d785a81a 100644 --- a/autopilot/prefattach_test.go +++ b/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 } diff --git a/chanbackup/backup.go b/chanbackup/backup.go index afffe5a2e..cf7217ae3 100644 --- a/chanbackup/backup.go +++ b/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 } diff --git a/chanbackup/backup_test.go b/chanbackup/backup_test.go index d36fa385b..d73b3769f 100644 --- a/chanbackup/backup_test.go +++ b/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, - []net.Addr, error) { +func (m *mockChannelSource) AddrsForNode(_ context.Context, + 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") } diff --git a/chanbackup/pubsub.go b/chanbackup/pubsub.go index 8fa1d5f34..43c368aaa 100644 --- a/chanbackup/pubsub.go +++ b/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, - keyRing keychain.KeyRing, backupSwapper Swapper) (*SubSwapper, error) { +func NewSubSwapper(ctx context.Context, startingChans []Single, + 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 } diff --git a/chanbackup/pubsub_test.go b/chanbackup/pubsub_test.go index 32694e5a7..49cfd3fb1 100644 --- a/chanbackup/pubsub_test.go +++ b/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{}) ( - *ChannelSubscription, error) { +func (m *mockChannelNotifier) SubscribeChans(_ context.Context, + _ 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) - if err == nil { - t.Fatalf("expected fail due to lack of subscription") - } + _, err := NewSubSwapper(ctx, nil, &chanNotifier, keyRing, &swapper) + require.Error(t, err) } 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 { diff --git a/channel_notifier.go b/channel_notifier.go index 88a05ac4c..8affd48f0 100644 --- a/channel_notifier.go +++ b/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 diff --git a/channeldb/addr_source.go b/channeldb/addr_source.go index de933ed49..99dc7f7e3 100644 --- a/channeldb/addr_source.go +++ b/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, - []net.Addr, error) { +func (c *multiAddrSource) AddrsForNode(ctx context.Context, + 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 } diff --git a/channeldb/addr_source_test.go b/channeldb/addr_source_test.go index 85ee30bf5..7562a2dd1 100644 --- a/channeldb/addr_source_test.go +++ b/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() - src2.On("AddrsForNode", pk1).Return(false, nil, nil).Once() + src1.On("AddrsForNode", ctx, 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, - error) { +func (m *mockAddrSource) AddrsForNode(ctx context.Context, + 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) } diff --git a/channeldb/db.go b/channeldb/db.go index b617d1008..90530459e 100644 --- a/channeldb/db.go +++ b/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. diff --git a/channeldb/db_test.go b/channeldb/db_test.go index 9c8c34d58..b010d1b65 100644 --- a/channeldb/db_test.go +++ b/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) diff --git a/discovery/gossiper.go b/discovery/gossiper.go index d6855ae45..2f390da5b 100644 --- a/discovery/gossiper.go +++ b/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 diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index 399010991..fb219f41c 100644 --- a/discovery/gossiper_test.go +++ b/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() diff --git a/graph/builder.go b/graph/builder.go index 99fc73c17..ce8dff179 100644 --- a/graph/builder.go +++ b/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 diff --git a/graph/builder_test.go b/graph/builder_test.go index fa7fc722d..8795fc857 100644 --- a/graph/builder_test.go +++ b/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 } diff --git a/graph/db/graph.go b/graph/db/graph.go index a68368e3d..fb2a9975a 100644 --- a/graph/db/graph.go +++ b/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, - op ...batch.SchedulerOption) error { +func (c *ChannelGraph) AddLightningNode(ctx context.Context, + 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 { - err := c.V1Store.DeleteLightningNode(nodePub) +func (c *ChannelGraph) DeleteLightningNode(ctx context.Context, + nodePub route.Vertex) error { + + err := c.V1Store.DeleteLightningNode(ctx, nodePub) if err != nil { return err } diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 18d3e52a5..b279f5476 100644 --- a/graph/db/graph_test.go +++ b/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 { - t.Fatalf("unable to delete node: %v", err) - } + err = graph.DeleteLightningNode(ctx, node1.PubKeyBytes) + 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 { - t.Fatalf("unable to add node: %v", err) - } + err := graph.AddLightningNode(ctx, node) + 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. diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 0f7d499fe..c1b3dd578 100644 --- a/graph/db/interfaces.go +++ b/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) ( - *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 // 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) // 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 diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index bbac55f5d..d722e4fe0 100644 --- a/graph/db/kv_store.go +++ b/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, - error) { +func (c *KVStore) AddrsForNode(ctx context.Context, + 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, - opts ...batch.SchedulerOption) error { - - ctx := context.TODO() +func (c *KVStore) AddLightningNode(ctx context.Context, + node *models.LightningNode, opts ...batch.SchedulerOption) error { 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) ( - *models.LightningNode, error) { +func (c *KVStore) FetchLightningNode(_ context.Context, + 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, - error) { +func (c *KVStore) HasLightningNode(_ context.Context, + nodePub [33]byte) (time.Time, bool, error) { var ( updateTime time.Time diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 2ae133480..1c3c1d116 100644 --- a/graph/db/sql_store.go +++ b/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, - opts ...batch.SchedulerOption) error { - - ctx := context.TODO() +func (s *SQLStore) AddLightningNode(ctx context.Context, + node *models.LightningNode, opts ...batch.SchedulerOption) error { 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) ( - *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 { @@ -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, - error) { - - ctx := context.TODO() +func (s *SQLStore) HasLightningNode(ctx context.Context, + pubKey [33]byte) (time.Time, bool, error) { 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, - error) { - - ctx := context.TODO() +func (s *SQLStore) AddrsForNode(ctx context.Context, + nodePub *btcec.PublicKey) (bool, []net.Addr, error) { 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 { - ctx := context.TODO() +func (s *SQLStore) DeleteLightningNode(ctx context.Context, + pubKey route.Vertex) error { err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { res, err := db.DeleteNodeByPubKey( diff --git a/graph/interfaces.go b/graph/interfaces.go index 52897d8f5..e9f894041 100644 --- a/graph/interfaces.go +++ b/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, - 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 diff --git a/graph/notifications_test.go b/graph/notifications_test.go index 2f2ec1b4b..099c2244b 100644 --- a/graph/notifications_test.go +++ b/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) } diff --git a/invoices/testdata/channel.db b/invoices/testdata/channel.db deleted file mode 100644 index 69397f529..000000000 Binary files a/invoices/testdata/channel.db and /dev/null differ diff --git a/lnrpc/devrpc/dev_server.go b/lnrpc/devrpc/dev_server.go index 71f51b085..a28a2500a 100644 --- a/lnrpc/devrpc/dev_server.go +++ b/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) } diff --git a/routing/pathfind_test.go b/routing/pathfind_test.go index 361fb83d4..0c483f0d9 100644 --- a/routing/pathfind_test.go +++ b/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 } diff --git a/routing/router_test.go b/routing/router_test.go index 0f81bea37..04f42c83d 100644 --- a/routing/router_test.go +++ b/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) diff --git a/rpcserver.go b/rpcserver.go index cd2367446..3c72d3459 100644 --- a/rpcserver.go +++ b/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 "+ diff --git a/server.go b/server.go index 9c6a08395..2c985b2cc 100644 --- a/server.go +++ b/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 }