GlcoinCore/lnd
1
0
Fork 0
mirror of https://github.com/lightningnetwork/lnd.git synced 2026-08-13 12:32:48 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

multi: thread context through to AddrsForNode

Browse source
This commit is contained in:
Elle Mouton 2025-06-17 12:06:03 +02:00
parent d1fa5701eb
commit e724e1c3e4
No known key found for this signature in database
GPG key ID: D7D916376026F177
15 changed files with 88 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

16
chanbackup/backup.go
View file

@ -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
}

15
chanbackup/backup_test.go
View file

@ -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")
}

11
chanbackup/pubsub.go
View file

@ -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
}

19
chanbackup/pubsub_test.go
View file

@ -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 {

6
channel_notifier.go
View file

@ -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

10
channeldb/addr_source.go
View file

@ -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
}

26
channeldb/addr_source_test.go
View file

@ -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)
}

5
channeldb/db.go
View file

@ -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.

6
channeldb/db_test.go
View file

@ -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)

6
graph/db/graph_test.go
View file

@ -183,7 +183,7 @@ 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)
@ -197,7 +197,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
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)
@ -224,7 +224,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
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)

3
graph/db/interfaces.go
View file

@ -63,7 +63,8 @@ type V1Store interface { //nolint:interfacebloat
// 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

6
graph/db/kv_store.go
View file

@ -376,10 +376,8 @@ 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) {
ctx := context.TODO()
func (c *KVStore) AddrsForNode(ctx context.Context,
nodePub *btcec.PublicKey) (bool, []net.Addr, error) {
pubKey, err := route.NewVertexFromBytes(nodePub.SerializeCompressed())
if err != nil {

6
graph/db/sql_store.go
View file

@ -258,10 +258,8 @@ func (s *SQLStore) HasLightningNode(ctx context.Context,
// 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

7
rpcserver.go
View file

@ -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 "+

10
server.go
View file

@ -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",