package lnd
import (
"bytes"
"context"
"crypto/rand"
"encoding/hex"
"fmt"
"math/big"
prand "math/rand"
"net"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/btcsuite/btcd/btcec/v2"
"github.com/btcsuite/btcd/btcec/v2/ecdsa"
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/connmgr"
"github.com/btcsuite/btcd/txscript"
"github.com/btcsuite/btcd/wire"
"github.com/go-errors/errors"
sphinx "github.com/lightningnetwork/lightning-onion"
"github.com/lightningnetwork/lnd/aliasmgr"
"github.com/lightningnetwork/lnd/autopilot"
"github.com/lightningnetwork/lnd/brontide"
"github.com/lightningnetwork/lnd/chainreg"
"github.com/lightningnetwork/lnd/chanacceptor"
"github.com/lightningnetwork/lnd/chanbackup"
"github.com/lightningnetwork/lnd/chanfitness"
"github.com/lightningnetwork/lnd/channeldb"
"github.com/lightningnetwork/lnd/channelnotifier"
"github.com/lightningnetwork/lnd/clock"
"github.com/lightningnetwork/lnd/contractcourt"
"github.com/lightningnetwork/lnd/discovery"
"github.com/lightningnetwork/lnd/feature"
"github.com/lightningnetwork/lnd/funding"
"github.com/lightningnetwork/lnd/healthcheck"
"github.com/lightningnetwork/lnd/htlcswitch"
"github.com/lightningnetwork/lnd/htlcswitch/hop"
"github.com/lightningnetwork/lnd/input"
"github.com/lightningnetwork/lnd/invoices"
"github.com/lightningnetwork/lnd/keychain"
"github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/lncfg"
"github.com/lightningnetwork/lnd/lnencrypt"
"github.com/lightningnetwork/lnd/lnpeer"
"github.com/lightningnetwork/lnd/lnrpc"
"github.com/lightningnetwork/lnd/lnrpc/routerrpc"
"github.com/lightningnetwork/lnd/lnwallet"
"github.com/lightningnetwork/lnd/lnwallet/chainfee"
"github.com/lightningnetwork/lnd/lnwallet/rpcwallet"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/nat"
"github.com/lightningnetwork/lnd/netann"
"github.com/lightningnetwork/lnd/peer"
"github.com/lightningnetwork/lnd/peernotifier"
"github.com/lightningnetwork/lnd/pool"
"github.com/lightningnetwork/lnd/queue"
"github.com/lightningnetwork/lnd/routing"
"github.com/lightningnetwork/lnd/routing/localchans"
"github.com/lightningnetwork/lnd/routing/route"
"github.com/lightningnetwork/lnd/subscribe"
"github.com/lightningnetwork/lnd/sweep"
"github.com/lightningnetwork/lnd/ticker"
"github.com/lightningnetwork/lnd/tor"
"github.com/lightningnetwork/lnd/walletunlocker"
"github.com/lightningnetwork/lnd/watchtower/blob"
"github.com/lightningnetwork/lnd/watchtower/wtclient"
"github.com/lightningnetwork/lnd/watchtower/wtpolicy"
"github.com/lightningnetwork/lnd/watchtower/wtserver"
)
const (
// defaultMinPeers is the minimum number of peers nodes should always be
// connected to.
defaultMinPeers = 3
// defaultStableConnDuration is a floor under which all reconnection
// attempts will apply exponential randomized backoff. Connections
// durations exceeding this value will be eligible to have their
// backoffs reduced.
defaultStableConnDuration = 10 * time.Minute
// numInstantInitReconnect specifies how many persistent peers we should
// always attempt outbound connections to immediately. After this value
// is surpassed, the remaining peers will be randomly delayed using
// maxInitReconnectDelay.
numInstantInitReconnect = 10
// maxInitReconnectDelay specifies the maximum delay in seconds we will
// apply in attempting to reconnect to persistent peers on startup. The
// value used or a particular peer will be chosen between 0s and this
// value.
maxInitReconnectDelay = 30
// multiAddrConnectionStagger is the number of seconds to wait between
// attempting to a peer with each of its advertised addresses.
multiAddrConnectionStagger = 10 * time.Second
)
var (
// ErrPeerNotConnected signals that the server has no connection to the
// given peer.
ErrPeerNotConnected = errors.New("peer is not connected")
// ErrServerNotActive indicates that the server has started but hasn't
// fully finished the startup process.
ErrServerNotActive = errors.New("server is still in the process of " +
"starting")
// ErrServerShuttingDown indicates that the server is in the process of
// gracefully exiting.
ErrServerShuttingDown = errors.New("server is shutting down")
// MaxFundingAmount is a soft-limit of the maximum channel size
// currently accepted within the Lightning Protocol. This is
// defined in BOLT-0002, and serves as an initial precautionary limit
// while implementations are battle tested in the real world.
//
// At the moment, this value depends on which chain is active. It is set
// to the value under the Bitcoin chain as default.
//
// TODO(roasbeef): add command line param to modify.
MaxFundingAmount = funding.MaxBtcFundingAmount
)
// errPeerAlreadyConnected is an error returned by the server when we're
// commanded to connect to a peer, but they're already connected.
type errPeerAlreadyConnected struct {
peer *peer.Brontide
}
// Error returns the human readable version of this error type.
//
// NOTE: Part of the error interface.
func (e *errPeerAlreadyConnected) Error() string {
return fmt.Sprintf("already connected to peer: %v", e.peer)
}
// server is the main server of the Lightning Network Daemon. The server houses
// global state pertaining to the wallet, database, and the rpcserver.
// Additionally, the server is also used as a central messaging bus to interact
// with any of its companion objects.
type server struct {
active int32 // atomic
stopping int32 // atomic
start sync.Once
stop sync.Once
cfg *Config
// identityECDH is an ECDH capable wrapper for the private key used
// to authenticate any incoming connections.
identityECDH keychain.SingleKeyECDH
// identityKeyLoc is the key locator for the above wrapped identity key.
identityKeyLoc keychain.KeyLocator
// nodeSigner is an implementation of the MessageSigner implementation
// that's backed by the identity private key of the running lnd node.
nodeSigner *netann.NodeSigner
chanStatusMgr *netann.ChanStatusManager
// listenAddrs is the list of addresses the server is currently
// listening on.
listenAddrs []net.Addr
// torController is a client that will communicate with a locally
// running Tor server. This client will handle initiating and
// authenticating the connection to the Tor server, automatically
// creating and setting up onion services, etc.
torController *tor.Controller
// natTraversal is the specific NAT traversal technique used to
// automatically set up port forwarding rules in order to advertise to
// the network that the node is accepting inbound connections.
natTraversal nat.Traversal
// lastDetectedIP is the last IP detected by the NAT traversal technique
// above. This IP will be watched periodically in a goroutine in order
// to handle dynamic IP changes.
lastDetectedIP net.IP
mu sync.RWMutex
peersByPub map[string]*peer.Brontide
inboundPeers map[string]*peer.Brontide
outboundPeers map[string]*peer.Brontide
peerConnectedListeners map[string][]chan<- lnpeer.Peer
peerDisconnectedListeners map[string][]chan<- struct{}
// TODO(yy): the Brontide.Start doesn't know this value, which means it
// will continue to send messages even if there are no active channels
// and the value below is false. Once it's pruned, all its connections
// will be closed, thus the Brontide.Start will return an error.
persistentPeers map[string]bool
persistentPeersBackoff map[string]time.Duration
persistentPeerAddrs map[string][]*lnwire.NetAddress
persistentConnReqs map[string][]*connmgr.ConnReq
persistentRetryCancels map[string]chan struct{}
// peerErrors keeps a set of peer error buffers for peers that have
// disconnected from us. This allows us to track historic peer errors
// over connections. The string of the peer's compressed pubkey is used
// as a key for this map.
peerErrors map[string]*queue.CircularBuffer
// ignorePeerTermination tracks peers for which the server has initiated
// a disconnect. Adding a peer to this map causes the peer termination
// watcher to short circuit in the event that peers are purposefully
// disconnected.
ignorePeerTermination map[*peer.Brontide]struct{}
// scheduledPeerConnection maps a pubkey string to a callback that
// should be executed in the peerTerminationWatcher the prior peer with
// the same pubkey exits. This allows the server to wait until the
// prior peer has cleaned up successfully, before adding the new peer
// intended to replace it.
scheduledPeerConnection map[string]func()
// pongBuf is a shared pong reply buffer we'll use across all active
// peer goroutines. We know the max size of a pong message
// (lnwire.MaxPongBytes), so we can allocate this ahead of time, and
// avoid allocations each time we need to send a pong message.
pongBuf []byte
cc *chainreg.ChainControl
fundingMgr *funding.Manager
graphDB *channeldb.ChannelGraph
chanStateDB *channeldb.ChannelStateDB
addrSource chanbackup.AddressSource
// miscDB is the DB that contains all "other" databases within the main
// channel DB that haven't been separated out yet.
miscDB *channeldb.DB
aliasMgr *aliasmgr.Manager
htlcSwitch *htlcswitch.Switch
interceptableSwitch *htlcswitch.InterceptableSwitch
invoices *invoices.InvoiceRegistry
channelNotifier *channelnotifier.ChannelNotifier
peerNotifier *peernotifier.PeerNotifier
htlcNotifier *htlcswitch.HtlcNotifier
witnessBeacon contractcourt.WitnessBeacon
breachArbiter *contractcourt.BreachArbiter
missionControl *routing.MissionControl
chanRouter *routing.ChannelRouter
controlTower routing.ControlTower
authGossiper *discovery.AuthenticatedGossiper
localChanMgr *localchans.Manager
utxoNursery *contractcourt.UtxoNursery
sweeper *sweep.UtxoSweeper
chainArb *contractcourt.ChainArbitrator
sphinx *hop.OnionProcessor
towerClient wtclient.Client
anchorTowerClient wtclient.Client
connMgr *connmgr.ConnManager
sigPool *lnwallet.SigPool
writePool *pool.Write
readPool *pool.Read
tlsManager *TLSManager
// featureMgr dispatches feature vectors for various contexts within the
// daemon.
featureMgr *feature.Manager
// currentNodeAnn is the node announcement that has been broadcast to
// the network upon startup, if the attributes of the node (us) has
// changed since last start.
currentNodeAnn *lnwire.NodeAnnouncement
// chansToRestore is the set of channels that upon starting, the server
// should attempt to restore/recover.
chansToRestore walletunlocker.ChannelsToRecover
// chanSubSwapper is a sub-system that will ensure our on-disk channel
// backups are consistent at all times. It interacts with the
// channelNotifier to be notified of newly opened and closed channels.
chanSubSwapper *chanbackup.SubSwapper
// chanEventStore tracks the behaviour of channels and their remote peers to
// provide insights into their health and performance.
chanEventStore *chanfitness.ChannelEventStore
hostAnn *netann.HostAnnouncer
// livenessMonitor monitors that lnd has access to critical resources.
livenessMonitor *healthcheck.Monitor
customMessageServer *subscribe.Server
quit chan struct{}
wg sync.WaitGroup
}
// updatePersistentPeerAddrs subscribes to topology changes and stores
// advertised addresses for any NodeAnnouncements from our persisted peers.
func (s *server) updatePersistentPeerAddrs() error {
graphSub, err := s.chanRouter.SubscribeTopology()
if err != nil {
return err
}
s.wg.Add(1)
go func() {
defer func() {
graphSub.Cancel()
s.wg.Done()
}()
for {
select {
case <-s.quit:
return
case topChange, ok := <-graphSub.TopologyChanges:
// If the router is shutting down, then we will
// as well.
if !ok {
return
}
for _, update := range topChange.NodeUpdates {
pubKeyStr := string(
update.IdentityKey.
SerializeCompressed(),
)
// We only care about updates from
// our persistentPeers.
s.mu.RLock()
_, ok := s.persistentPeers[pubKeyStr]
s.mu.RUnlock()
if !ok {
continue
}
addrs := make([]*lnwire.NetAddress, 0,
len(update.Addresses))
for _, addr := range update.Addresses {
addrs = append(addrs,
&lnwire.NetAddress{
IdentityKey: update.IdentityKey,
Address: addr,
ChainNet: s.cfg.ActiveNetParams.Net,
},
)
}
s.mu.Lock()
// Update the stored addresses for this
// to peer to reflect the new set.
s.persistentPeerAddrs[pubKeyStr] = addrs
// If there are no outstanding
// connection requests for this peer
// then our work is done since we are
// not currently trying to connect to
// them.
if len(s.persistentConnReqs[pubKeyStr]) == 0 {
s.mu.Unlock()
continue
}
s.mu.Unlock()
s.connectToPersistentPeer(pubKeyStr)
}
}
}
}()
return nil
}
// CustomMessage is a custom message that is received from a peer.
type CustomMessage struct {
// Peer is the peer pubkey
Peer [33]byte
// Msg is the custom wire message.
Msg *lnwire.Custom
}
// parseAddr parses an address from its string format to a net.Addr.
func parseAddr(address string, netCfg tor.Net) (net.Addr, error) {
var (
host string
port int
)
// Split the address into its host and port components.
h, p, err := net.SplitHostPort(address)
if err != nil {
// If a port wasn't specified, we'll assume the address only
// contains the host so we'll use the default port.
host = address
port = defaultPeerPort
} else {
// Otherwise, we'll note both the host and ports.
host = h
portNum, err := strconv.Atoi(p)
if err != nil {
return nil, err
}
port = portNum
}
if tor.IsOnionHost(host) {
return &tor.OnionAddr{OnionService: host, Port: port}, nil
}
// If the host is part of a TCP address, we'll use the network
// specific ResolveTCPAddr function in order to resolve these
// addresses over Tor in order to prevent leaking your real IP
// address.
hostPort := net.JoinHostPort(host, strconv.Itoa(port))
return netCfg.ResolveTCPAddr("tcp", hostPort)
}
// noiseDial is a factory function which creates a connmgr compliant dialing
// function by returning a closure which includes the server's identity key.
func noiseDial(idKey keychain.SingleKeyECDH,
netCfg tor.Net, timeout time.Duration) func(net.Addr) (net.Conn, error) {
return func(a net.Addr) (net.Conn, error) {
lnAddr := a.(*lnwire.NetAddress)
return brontide.Dial(idKey, lnAddr, timeout, netCfg.Dial)
}
}
// newServer creates a new instance of the server which is to listen using the
// passed listener address.
func newServer(cfg *Config, listenAddrs []net.Addr,
dbs *DatabaseInstances, cc *chainreg.ChainControl,
nodeKeyDesc *keychain.KeyDescriptor,
chansToRestore walletunlocker.ChannelsToRecover,
chanPredicate chanacceptor.ChannelAcceptor,
torController *tor.Controller, tlsManager *TLSManager) (*server,
error) {
var (
err error
nodeKeyECDH = keychain.NewPubKeyECDH(*nodeKeyDesc, cc.KeyRing)
// We just derived the full descriptor, so we know the public
// key is set on it.
nodeKeySigner = keychain.NewPubKeyMessageSigner(
nodeKeyDesc.PubKey, nodeKeyDesc.KeyLocator, cc.KeyRing,
)
)
listeners := make([]net.Listener, len(listenAddrs))
for i, listenAddr := range listenAddrs {
// Note: though brontide.NewListener uses ResolveTCPAddr, it
// doesn't need to call the general lndResolveTCP function
// since we are resolving a local address.
listeners[i], err = brontide.NewListener(
nodeKeyECDH, listenAddr.String(),
)
if err != nil {
return nil, err
}
}
var serializedPubKey [33]byte
copy(serializedPubKey[:], nodeKeyDesc.PubKey.SerializeCompressed())
// Initialize the sphinx router.
replayLog := htlcswitch.NewDecayedLog(
dbs.DecayedLogDB, cc.ChainNotifier,
)
sphinxRouter := sphinx.NewRouter(
nodeKeyECDH, cfg.ActiveNetParams.Params, replayLog,
)
writeBufferPool := pool.NewWriteBuffer(
pool.DefaultWriteBufferGCInterval,
pool.DefaultWriteBufferExpiryInterval,
)
writePool := pool.NewWrite(
writeBufferPool, cfg.Workers.Write, pool.DefaultWorkerTimeout,
)
readBufferPool := pool.NewReadBuffer(
pool.DefaultReadBufferGCInterval,
pool.DefaultReadBufferExpiryInterval,
)
readPool := pool.NewRead(
readBufferPool, cfg.Workers.Read, pool.DefaultWorkerTimeout,
)
//nolint:lll
featureMgr, err := feature.NewManager(feature.Config{
NoTLVOnion: cfg.ProtocolOptions.LegacyOnion(),
NoStaticRemoteKey: cfg.ProtocolOptions.NoStaticRemoteKey(),
NoAnchors: cfg.ProtocolOptions.NoAnchorCommitments(),
NoWumbo: !cfg.ProtocolOptions.Wumbo(),
NoScriptEnforcementLease: cfg.ProtocolOptions.NoScriptEnforcementLease(),
NoKeysend: !cfg.AcceptKeySend,
NoOptionScidAlias: !cfg.ProtocolOptions.ScidAlias(),
NoZeroConf: !cfg.ProtocolOptions.ZeroConf(),
NoAnySegwit: cfg.ProtocolOptions.NoAnySegwit(),
CustomFeatures: cfg.ProtocolOptions.ExperimentalProtocol.CustomFeatures(),
NoTaprootChans: !cfg.ProtocolOptions.TaprootChans,
})
if err != nil {
return nil, err
}
registryConfig := invoices.RegistryConfig{
FinalCltvRejectDelta: lncfg.DefaultFinalCltvRejectDelta,
HtlcHoldDuration: invoices.DefaultHtlcHoldDuration,
Clock: clock.NewDefaultClock(),
AcceptKeySend: cfg.AcceptKeySend,
AcceptAMP: cfg.AcceptAMP,
GcCanceledInvoicesOnStartup: cfg.GcCanceledInvoicesOnStartup,
GcCanceledInvoicesOnTheFly: cfg.GcCanceledInvoicesOnTheFly,
KeysendHoldTime: cfg.KeysendHoldTime,
}
s := &server{
cfg: cfg,
graphDB: dbs.GraphDB.ChannelGraph(),
chanStateDB: dbs.ChanStateDB.ChannelStateDB(),
addrSource: dbs.ChanStateDB,
miscDB: dbs.ChanStateDB,
cc: cc,
sigPool: lnwallet.NewSigPool(cfg.Workers.Sig, cc.Signer),
writePool: writePool,
readPool: readPool,
chansToRestore: chansToRestore,
channelNotifier: channelnotifier.New(
dbs.ChanStateDB.ChannelStateDB(),
),
identityECDH: nodeKeyECDH,
identityKeyLoc: nodeKeyDesc.KeyLocator,
nodeSigner: netann.NewNodeSigner(nodeKeySigner),
listenAddrs: listenAddrs,
// TODO(roasbeef): derive proper onion key based on rotation
// schedule
sphinx: hop.NewOnionProcessor(sphinxRouter),
torController: torController,
persistentPeers: make(map[string]bool),
persistentPeersBackoff: make(map[string]time.Duration),
persistentConnReqs: make(map[string][]*connmgr.ConnReq),
persistentPeerAddrs: make(map[string][]*lnwire.NetAddress),
persistentRetryCancels: make(map[string]chan struct{}),
peerErrors: make(map[string]*queue.CircularBuffer),
ignorePeerTermination: make(map[*peer.Brontide]struct{}),
scheduledPeerConnection: make(map[string]func()),
pongBuf: make([]byte, lnwire.MaxPongBytes),
peersByPub: make(map[string]*peer.Brontide),
inboundPeers: make(map[string]*peer.Brontide),
outboundPeers: make(map[string]*peer.Brontide),
peerConnectedListeners: make(map[string][]chan<- lnpeer.Peer),
peerDisconnectedListeners: make(map[string][]chan<- struct{}),
customMessageServer: subscribe.NewServer(),
tlsManager: tlsManager,
featureMgr: featureMgr,
quit: make(chan struct{}),
}
currentHash, currentHeight, err := s.cc.ChainIO.GetBestBlock()
if err != nil {
return nil, err
}
expiryWatcher := invoices.NewInvoiceExpiryWatcher(
clock.NewDefaultClock(), cfg.Invoices.HoldExpiryDelta,
uint32(currentHeight), currentHash, cc.ChainNotifier,
)
s.invoices = invoices.NewRegistry(
dbs.InvoiceDB, expiryWatcher, ®istryConfig,
)
s.htlcNotifier = htlcswitch.NewHtlcNotifier(time.Now)
thresholdSats := btcutil.Amount(cfg.DustThreshold)
thresholdMSats := lnwire.NewMSatFromSatoshis(thresholdSats)
s.aliasMgr, err = aliasmgr.NewManager(dbs.ChanStateDB)
if err != nil {
return nil, err
}
s.htlcSwitch, err = htlcswitch.New(htlcswitch.Config{
DB: dbs.ChanStateDB,
FetchAllOpenChannels: s.chanStateDB.FetchAllOpenChannels,
FetchAllChannels: s.chanStateDB.FetchAllChannels,
FetchClosedChannels: s.chanStateDB.FetchClosedChannels,
LocalChannelClose: func(pubKey []byte,
request *htlcswitch.ChanClose) {
peer, err := s.FindPeerByPubStr(string(pubKey))
if err != nil {
srvrLog.Errorf("unable to close channel, peer"+
" with %v id can't be found: %v",
pubKey, err,
)
return
}
peer.HandleLocalCloseChanReqs(request)
},
FwdingLog: dbs.ChanStateDB.ForwardingLog(),
SwitchPackager: channeldb.NewSwitchPackager(),
ExtractErrorEncrypter: s.sphinx.ExtractErrorEncrypter,
FetchLastChannelUpdate: s.fetchLastChanUpdate(),
Notifier: s.cc.ChainNotifier,
HtlcNotifier: s.htlcNotifier,
FwdEventTicker: ticker.New(htlcswitch.DefaultFwdEventInterval),
LogEventTicker: ticker.New(htlcswitch.DefaultLogInterval),
AckEventTicker: ticker.New(htlcswitch.DefaultAckInterval),
AllowCircularRoute: cfg.AllowCircularRoute,
RejectHTLC: cfg.RejectHTLC,
Clock: clock.NewDefaultClock(),
MailboxDeliveryTimeout: cfg.Htlcswitch.MailboxDeliveryTimeout,
DustThreshold: thresholdMSats,
SignAliasUpdate: s.signAliasUpdate,
IsAlias: aliasmgr.IsAlias,
}, uint32(currentHeight))
if err != nil {
return nil, err
}
s.interceptableSwitch, err = htlcswitch.NewInterceptableSwitch(
&htlcswitch.InterceptableSwitchConfig{
Switch: s.htlcSwitch,
CltvRejectDelta: lncfg.DefaultFinalCltvRejectDelta,
CltvInterceptDelta: lncfg.DefaultCltvInterceptDelta,
RequireInterceptor: s.cfg.RequireInterceptor,
Notifier: s.cc.ChainNotifier,
},
)
if err != nil {
return nil, err
}
s.witnessBeacon = newPreimageBeacon(
dbs.ChanStateDB.NewWitnessCache(),
s.interceptableSwitch.ForwardPacket,
)
chanStatusMgrCfg := &netann.ChanStatusConfig{
ChanStatusSampleInterval: cfg.ChanStatusSampleInterval,
ChanEnableTimeout: cfg.ChanEnableTimeout,
ChanDisableTimeout: cfg.ChanDisableTimeout,
OurPubKey: nodeKeyDesc.PubKey,
OurKeyLoc: nodeKeyDesc.KeyLocator,
MessageSigner: s.nodeSigner,
IsChannelActive: s.htlcSwitch.HasActiveLink,
ApplyChannelUpdate: s.applyChannelUpdate,
DB: s.chanStateDB,
Graph: dbs.GraphDB.ChannelGraph(),
}
chanStatusMgr, err := netann.NewChanStatusManager(chanStatusMgrCfg)
if err != nil {
return nil, err
}
s.chanStatusMgr = chanStatusMgr
// If enabled, use either UPnP or NAT-PMP to automatically configure
// port forwarding for users behind a NAT.
if cfg.NAT {
srvrLog.Info("Scanning local network for a UPnP enabled device")
discoveryTimeout := time.Duration(10 * time.Second)
ctx, cancel := context.WithTimeout(
context.Background(), discoveryTimeout,
)
defer cancel()
upnp, err := nat.DiscoverUPnP(ctx)
if err == nil {
s.natTraversal = upnp
} else {
// If we were not able to discover a UPnP enabled device
// on the local network, we'll fall back to attempting
// to discover a NAT-PMP enabled device.
srvrLog.Errorf("Unable to discover a UPnP enabled "+
"device on the local network: %v", err)
srvrLog.Info("Scanning local network for a NAT-PMP " +
"enabled device")
pmp, err := nat.DiscoverPMP(discoveryTimeout)
if err != nil {
err := fmt.Errorf("unable to discover a "+
"NAT-PMP enabled device on the local "+
"network: %v", err)
srvrLog.Error(err)
return nil, err
}
s.natTraversal = pmp
}
}
// If we were requested to automatically configure port forwarding,
// we'll use the ports that the server will be listening on.
externalIPStrings := make([]string, len(cfg.ExternalIPs))
for idx, ip := range cfg.ExternalIPs {
externalIPStrings[idx] = ip.String()
}
if s.natTraversal != nil {
listenPorts := make([]uint16, 0, len(listenAddrs))
for _, listenAddr := range listenAddrs {
// At this point, the listen addresses should have
// already been normalized, so it's safe to ignore the
// errors.
_, portStr, _ := net.SplitHostPort(listenAddr.String())
port, _ := strconv.Atoi(portStr)
listenPorts = append(listenPorts, uint16(port))
}
ips, err := s.configurePortForwarding(listenPorts...)
if err != nil {
srvrLog.Errorf("Unable to automatically set up port "+
"forwarding using %s: %v",
s.natTraversal.Name(), err)
} else {
srvrLog.Infof("Automatically set up port forwarding "+
"using %s to advertise external IP",
s.natTraversal.Name())
externalIPStrings = append(externalIPStrings, ips...)
}
}
// If external IP addresses have been specified, add those to the list
// of this server's addresses.
externalIPs, err := lncfg.NormalizeAddresses(
externalIPStrings, strconv.Itoa(defaultPeerPort),
cfg.net.ResolveTCPAddr,
)
if err != nil {
return nil, err
}
selfAddrs := make([]net.Addr, 0, len(externalIPs))
selfAddrs = append(selfAddrs, externalIPs...)
// As the graph can be obtained at anytime from the network, we won't
// replicate it, and instead it'll only be stored locally.
chanGraph := dbs.GraphDB.ChannelGraph()
// We'll now reconstruct a node announcement based on our current
// configuration so we can send it out as a sort of heart beat within
// the network.
//
// We'll start by parsing the node color from configuration.
color, err := lncfg.ParseHexColor(cfg.Color)
if err != nil {
srvrLog.Errorf("unable to parse color: %v\n", err)
return nil, err
}
// If no alias is provided, default to first 10 characters of public
// key.
alias := cfg.Alias
if alias == "" {
alias = hex.EncodeToString(serializedPubKey[:10])
}
nodeAlias, err := lnwire.NewNodeAlias(alias)
if err != nil {
return nil, err
}
selfNode := &channeldb.LightningNode{
HaveNodeAnnouncement: true,
LastUpdate: time.Now(),
Addresses: selfAddrs,
Alias: nodeAlias.String(),
Features: s.featureMgr.Get(feature.SetNodeAnn),
Color: color,
}
copy(selfNode.PubKeyBytes[:], nodeKeyDesc.PubKey.SerializeCompressed())
// Based on the disk representation of the node announcement generated
// above, we'll generate a node announcement that can go out on the
// network so we can properly sign it.
nodeAnn, err := selfNode.NodeAnnouncement(false)
if err != nil {
return nil, fmt.Errorf("unable to gen self node ann: %v", err)
}
// With the announcement generated, we'll sign it to properly
// authenticate the message on the network.
authSig, err := netann.SignAnnouncement(
s.nodeSigner, nodeKeyDesc.KeyLocator, nodeAnn,
)
if err != nil {
return nil, fmt.Errorf("unable to generate signature for "+
"self node announcement: %v", err)
}
selfNode.AuthSigBytes = authSig.Serialize()
nodeAnn.Signature, err = lnwire.NewSigFromECDSARawSignature(
selfNode.AuthSigBytes,
)
if err != nil {
return nil, err
}
// Finally, we'll update the representation on disk, and update our
// cached in-memory version as well.
if err := chanGraph.SetSourceNode(selfNode); err != nil {
return nil, fmt.Errorf("can't set self node: %v", err)
}
s.currentNodeAnn = nodeAnn
// The router will get access to the payment ID sequencer, such that it
// can generate unique payment IDs.
sequencer, err := htlcswitch.NewPersistentSequencer(dbs.ChanStateDB)
if err != nil {
return nil, err
}
// Instantiate mission control with config from the sub server.
//
// TODO(joostjager): When we are further in the process of moving to sub
// servers, the mission control instance itself can be moved there too.
routingConfig := routerrpc.GetRoutingConfig(cfg.SubRPCServers.RouterRPC)
// We only initialize a probability estimator if there's no custom one.
var estimator routing.Estimator
if cfg.Estimator != nil {
estimator = cfg.Estimator
} else {
switch routingConfig.ProbabilityEstimatorType {
case routing.AprioriEstimatorName:
aCfg := routingConfig.AprioriConfig
aprioriConfig := routing.AprioriConfig{
AprioriHopProbability: aCfg.HopProbability,
PenaltyHalfLife: aCfg.PenaltyHalfLife,
AprioriWeight: aCfg.Weight,
CapacityFraction: aCfg.CapacityFraction,
}
estimator, err = routing.NewAprioriEstimator(
aprioriConfig,
)
if err != nil {
return nil, err
}
case routing.BimodalEstimatorName:
bCfg := routingConfig.BimodalConfig
bimodalConfig := routing.BimodalConfig{
BimodalNodeWeight: bCfg.NodeWeight,
BimodalScaleMsat: lnwire.MilliSatoshi(
bCfg.Scale,
),
BimodalDecayTime: bCfg.DecayTime,
}
estimator, err = routing.NewBimodalEstimator(
bimodalConfig,
)
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unknown estimator type %v",
routingConfig.ProbabilityEstimatorType)
}
}
mcCfg := &routing.MissionControlConfig{
Estimator: estimator,
MaxMcHistory: routingConfig.MaxMcHistory,
McFlushInterval: routingConfig.McFlushInterval,
MinFailureRelaxInterval: routing.DefaultMinFailureRelaxInterval,
}
s.missionControl, err = routing.NewMissionControl(
dbs.ChanStateDB, selfNode.PubKeyBytes, mcCfg,
)
if err != nil {
return nil, fmt.Errorf("can't create mission control: %v", err)
}
srvrLog.Debugf("Instantiating payment session source with config: "+
"AttemptCost=%v + %v%%, MinRouteProbability=%v",
int64(routingConfig.AttemptCost),
float64(routingConfig.AttemptCostPPM)/10000,
routingConfig.MinRouteProbability)
pathFindingConfig := routing.PathFindingConfig{
AttemptCost: lnwire.NewMSatFromSatoshis(
routingConfig.AttemptCost,
),
AttemptCostPPM: routingConfig.AttemptCostPPM,
MinProbability: routingConfig.MinRouteProbability,
}
sourceNode, err := chanGraph.SourceNode()
if err != nil {
return nil, fmt.Errorf("error getting source node: %v", err)
}
paymentSessionSource := &routing.SessionSource{
Graph: chanGraph,
SourceNode: sourceNode,
MissionControl: s.missionControl,
GetLink: s.htlcSwitch.GetLinkByShortID,
PathFindingConfig: pathFindingConfig,
}
paymentControl := channeldb.NewPaymentControl(dbs.ChanStateDB)
s.controlTower = routing.NewControlTower(paymentControl)
strictPruning := (cfg.Bitcoin.Node == "neutrino" ||
cfg.Routing.StrictZombiePruning)
s.chanRouter, err = routing.New(routing.Config{
Graph: chanGraph,
Chain: cc.ChainIO,
ChainView: cc.ChainView,
Notifier: cc.ChainNotifier,
Payer: s.htlcSwitch,
Control: s.controlTower,
MissionControl: s.missionControl,
SessionSource: paymentSessionSource,
ChannelPruneExpiry: routing.DefaultChannelPruneExpiry,
GraphPruneInterval: time.Hour,
FirstTimePruneDelay: routing.DefaultFirstTimePruneDelay,
GetLink: s.htlcSwitch.GetLinkByShortID,
AssumeChannelValid: cfg.Routing.AssumeChannelValid,
NextPaymentID: sequencer.NextID,
PathFindingConfig: pathFindingConfig,
Clock: clock.NewDefaultClock(),
StrictZombiePruning: strictPruning,
IsAlias: aliasmgr.IsAlias,
})
if err != nil {
return nil, fmt.Errorf("can't create router: %v", err)
}
chanSeries := discovery.NewChanSeries(s.graphDB)
gossipMessageStore, err := discovery.NewMessageStore(dbs.ChanStateDB)
if err != nil {
return nil, err
}
waitingProofStore, err := channeldb.NewWaitingProofStore(dbs.ChanStateDB)
if err != nil {
return nil, err
}
s.authGossiper = discovery.New(discovery.Config{
Router: s.chanRouter,
Notifier: s.cc.ChainNotifier,
ChainHash: *s.cfg.ActiveNetParams.GenesisHash,
Broadcast: s.BroadcastMessage,
ChanSeries: chanSeries,
NotifyWhenOnline: s.NotifyWhenOnline,
NotifyWhenOffline: s.NotifyWhenOffline,
FetchSelfAnnouncement: s.getNodeAnnouncement,
UpdateSelfAnnouncement: func() (lnwire.NodeAnnouncement,
error) {
return s.genNodeAnnouncement(nil)
},
ProofMatureDelta: 0,
TrickleDelay: time.Millisecond * time.Duration(cfg.TrickleDelay),
RetransmitTicker: ticker.New(time.Minute * 30),
RebroadcastInterval: time.Hour * 24,
WaitingProofStore: waitingProofStore,
MessageStore: gossipMessageStore,
AnnSigner: s.nodeSigner,
RotateTicker: ticker.New(discovery.DefaultSyncerRotationInterval),
HistoricalSyncTicker: ticker.New(cfg.HistoricalSyncInterval),
NumActiveSyncers: cfg.NumGraphSyncPeers,
MinimumBatchSize: 10,
SubBatchDelay: cfg.Gossip.SubBatchDelay,
IgnoreHistoricalFilters: cfg.IgnoreHistoricalGossipFilters,
PinnedSyncers: cfg.Gossip.PinnedSyncers,
MaxChannelUpdateBurst: cfg.Gossip.MaxChannelUpdateBurst,
ChannelUpdateInterval: cfg.Gossip.ChannelUpdateInterval,
IsAlias: aliasmgr.IsAlias,
SignAliasUpdate: s.signAliasUpdate,
FindBaseByAlias: s.aliasMgr.FindBaseSCID,
GetAlias: s.aliasMgr.GetPeerAlias,
FindChannel: s.findChannel,
}, nodeKeyDesc)
s.localChanMgr = &localchans.Manager{
ForAllOutgoingChannels: s.chanRouter.ForAllOutgoingChannels,
PropagateChanPolicyUpdate: s.authGossiper.PropagateChanPolicyUpdate,
UpdateForwardingPolicies: s.htlcSwitch.UpdateForwardingPolicies,
FetchChannel: s.chanStateDB.FetchChannel,
}
utxnStore, err := contractcourt.NewNurseryStore(
s.cfg.ActiveNetParams.GenesisHash, dbs.ChanStateDB,
)
if err != nil {
srvrLog.Errorf("unable to create nursery store: %v", err)
return nil, err
}
srvrLog.Debugf("Sweeper batch window duration: %v",
cfg.Sweeper.BatchWindowDuration)
sweeperStore, err := sweep.NewSweeperStore(
dbs.ChanStateDB, s.cfg.ActiveNetParams.GenesisHash,
)
if err != nil {
srvrLog.Errorf("unable to create sweeper store: %v", err)
return nil, err
}
s.sweeper = sweep.New(&sweep.UtxoSweeperConfig{
FeeEstimator: cc.FeeEstimator,
GenSweepScript: newSweepPkScriptGen(cc.Wallet),
Signer: cc.Wallet.Cfg.Signer,
Wallet: newSweeperWallet(cc.Wallet),
NewBatchTimer: func() <-chan time.Time {
return time.NewTimer(cfg.Sweeper.BatchWindowDuration).C
},
Notifier: cc.ChainNotifier,
Store: sweeperStore,
MaxInputsPerTx: sweep.DefaultMaxInputsPerTx,
MaxSweepAttempts: sweep.DefaultMaxSweepAttempts,
NextAttemptDeltaFunc: sweep.DefaultNextAttemptDeltaFunc,
MaxFeeRate: sweep.DefaultMaxFeeRate,
FeeRateBucketSize: sweep.DefaultFeeRateBucketSize,
})
s.utxoNursery = contractcourt.NewUtxoNursery(&contractcourt.NurseryConfig{
ChainIO: cc.ChainIO,
ConfDepth: 1,
FetchClosedChannels: s.chanStateDB.FetchClosedChannels,
FetchClosedChannel: s.chanStateDB.FetchClosedChannel,
Notifier: cc.ChainNotifier,
PublishTransaction: cc.Wallet.PublishTransaction,
Store: utxnStore,
SweepInput: s.sweeper.SweepInput,
})
// Construct a closure that wraps the htlcswitch's CloseLink method.
closeLink := func(chanPoint *wire.OutPoint,
closureType contractcourt.ChannelCloseType) {
// TODO(conner): Properly respect the update and error channels
// returned by CloseLink.
// Instruct the switch to close the channel. Provide no close out
// delivery script or target fee per kw because user input is not
// available when the remote peer closes the channel.
s.htlcSwitch.CloseLink(chanPoint, closureType, 0, 0, nil)
}
// We will use the following channel to reliably hand off contract
// breach events from the ChannelArbitrator to the breachArbiter,
contractBreaches := make(chan *contractcourt.ContractBreachEvent, 1)
s.breachArbiter = contractcourt.NewBreachArbiter(&contractcourt.BreachConfig{
CloseLink: closeLink,
DB: s.chanStateDB,
Estimator: s.cc.FeeEstimator,
GenSweepScript: newSweepPkScriptGen(cc.Wallet),
Notifier: cc.ChainNotifier,
PublishTransaction: cc.Wallet.PublishTransaction,
ContractBreaches: contractBreaches,
Signer: cc.Wallet.Cfg.Signer,
Store: contractcourt.NewRetributionStore(
dbs.ChanStateDB,
),
})
s.chainArb = contractcourt.NewChainArbitrator(contractcourt.ChainArbitratorConfig{
ChainHash: *s.cfg.ActiveNetParams.GenesisHash,
IncomingBroadcastDelta: lncfg.DefaultIncomingBroadcastDelta,
OutgoingBroadcastDelta: lncfg.DefaultOutgoingBroadcastDelta,
NewSweepAddr: newSweepPkScriptGen(cc.Wallet),
PublishTx: cc.Wallet.PublishTransaction,
DeliverResolutionMsg: func(msgs ...contractcourt.ResolutionMsg) error {
for _, msg := range msgs {
err := s.htlcSwitch.ProcessContractResolution(msg)
if err != nil {
return err
}
}
return nil
},
IncubateOutputs: func(chanPoint wire.OutPoint,
outHtlcRes *lnwallet.OutgoingHtlcResolution,
inHtlcRes *lnwallet.IncomingHtlcResolution,
broadcastHeight uint32) error {
var (
inRes []lnwallet.IncomingHtlcResolution
outRes []lnwallet.OutgoingHtlcResolution
)
if inHtlcRes != nil {
inRes = append(inRes, *inHtlcRes)
}
if outHtlcRes != nil {
outRes = append(outRes, *outHtlcRes)
}
return s.utxoNursery.IncubateOutputs(
chanPoint, outRes, inRes,
broadcastHeight,
)
},
PreimageDB: s.witnessBeacon,
Notifier: cc.ChainNotifier,
Mempool: cc.MempoolNotifier,
Signer: cc.Wallet.Cfg.Signer,
FeeEstimator: cc.FeeEstimator,
ChainIO: cc.ChainIO,
MarkLinkInactive: func(chanPoint wire.OutPoint) error {
chanID := lnwire.NewChanIDFromOutPoint(&chanPoint)
s.htlcSwitch.RemoveLink(chanID)
return nil
},
IsOurAddress: cc.Wallet.IsOurAddress,
ContractBreach: func(chanPoint wire.OutPoint,
breachRet *lnwallet.BreachRetribution) error {
// processACK will handle the breachArbiter ACKing the
// event.
finalErr := make(chan error, 1)
processACK := func(brarErr error) {
if brarErr != nil {
finalErr <- brarErr
return
}
// If the breachArbiter successfully handled
// the event, we can signal that the handoff
// was successful.
finalErr <- nil
}
event := &contractcourt.ContractBreachEvent{
ChanPoint: chanPoint,
ProcessACK: processACK,
BreachRetribution: breachRet,
}
// Send the contract breach event to the breachArbiter.
select {
case contractBreaches <- event:
case <-s.quit:
return ErrServerShuttingDown
}
// We'll wait for a final error to be available from
// the breachArbiter.
select {
case err := <-finalErr:
return err
case <-s.quit:
return ErrServerShuttingDown
}
},
DisableChannel: func(chanPoint wire.OutPoint) error {
return s.chanStatusMgr.RequestDisable(chanPoint, false)
},
Sweeper: s.sweeper,
Registry: s.invoices,
NotifyClosedChannel: s.channelNotifier.NotifyClosedChannelEvent,
NotifyFullyResolvedChannel: s.channelNotifier.NotifyFullyResolvedChannelEvent,
OnionProcessor: s.sphinx,
PaymentsExpirationGracePeriod: cfg.PaymentsExpirationGracePeriod,
IsForwardedHTLC: s.htlcSwitch.IsForwardedHTLC,
Clock: clock.NewDefaultClock(),
SubscribeBreachComplete: s.breachArbiter.SubscribeBreachComplete,
PutFinalHtlcOutcome: s.chanStateDB.PutOnchainFinalHtlcOutcome, //nolint: lll
HtlcNotifier: s.htlcNotifier,
}, dbs.ChanStateDB)
// Select the configuration and furnding parameters for Bitcoin or
// Litecoin, depending on the primary registered chain.
primaryChain := cfg.registeredChains.PrimaryChain()
chainCfg := cfg.Bitcoin
minRemoteDelay := funding.MinBtcRemoteDelay
maxRemoteDelay := funding.MaxBtcRemoteDelay
if primaryChain == chainreg.LitecoinChain {
chainCfg = cfg.Litecoin
minRemoteDelay = funding.MinLtcRemoteDelay
maxRemoteDelay = funding.MaxLtcRemoteDelay
}
var chanIDSeed [32]byte
if _, err := rand.Read(chanIDSeed[:]); err != nil {
return nil, err
}
// Wrap the DeleteChannelEdges method so that the funding manager can
// use it without depending on several layers of indirection.
deleteAliasEdge := func(scid lnwire.ShortChannelID) (
*channeldb.ChannelEdgePolicy, error) {
info, e1, e2, err := s.graphDB.FetchChannelEdgesByID(
scid.ToUint64(),
)
if err == channeldb.ErrEdgeNotFound {
// This is unlikely but there is a slim chance of this
// being hit if lnd was killed via SIGKILL and the
// funding manager was stepping through the delete
// alias edge logic.
return nil, nil
} else if err != nil {
return nil, err
}
// Grab our key to find our policy.
var ourKey [33]byte
copy(ourKey[:], nodeKeyDesc.PubKey.SerializeCompressed())
var ourPolicy *channeldb.ChannelEdgePolicy
if info != nil && info.NodeKey1Bytes == ourKey {
ourPolicy = e1
} else {
ourPolicy = e2
}
if ourPolicy == nil {
// Something is wrong, so return an error.
return nil, fmt.Errorf("we don't have an edge")
}
err = s.graphDB.DeleteChannelEdges(
false, false, scid.ToUint64(),
)
return ourPolicy, err
}
// Get the development config for funding manager. If we are not in
// development mode, this would be nil.
var devCfg *funding.DevConfig
if lncfg.IsDevBuild() {
devCfg = &funding.DevConfig{
ProcessChannelReadyWait: cfg.Dev.ChannelReadyWait(),
}
}
//nolint:lll
s.fundingMgr, err = funding.NewFundingManager(funding.Config{
Dev: devCfg,
NoWumboChans: !cfg.ProtocolOptions.Wumbo(),
IDKey: nodeKeyDesc.PubKey,
IDKeyLoc: nodeKeyDesc.KeyLocator,
Wallet: cc.Wallet,
PublishTransaction: cc.Wallet.PublishTransaction,
UpdateLabel: func(hash chainhash.Hash, label string) error {
return cc.Wallet.LabelTransaction(hash, label, true)
},
Notifier: cc.ChainNotifier,
ChannelDB: s.chanStateDB,
FeeEstimator: cc.FeeEstimator,
SignMessage: cc.MsgSigner.SignMessage,
CurrentNodeAnnouncement: func() (lnwire.NodeAnnouncement,
error) {
return s.genNodeAnnouncement(nil)
},
SendAnnouncement: s.authGossiper.ProcessLocalAnnouncement,
NotifyWhenOnline: s.NotifyWhenOnline,
TempChanIDSeed: chanIDSeed,
FindChannel: s.findChannel,
DefaultRoutingPolicy: cc.RoutingPolicy,
DefaultMinHtlcIn: cc.MinHtlcIn,
NumRequiredConfs: func(chanAmt btcutil.Amount,
pushAmt lnwire.MilliSatoshi) uint16 {
// For large channels we increase the number
// of confirmations we require for the
// channel to be considered open. As it is
// always the responder that gets to choose
// value, the pushAmt is value being pushed
// to us. This means we have more to lose
// in the case this gets re-orged out, and
// we will require more confirmations before
// we consider it open.
// TODO(halseth): Use Litecoin params in case
// of LTC channels.
// In case the user has explicitly specified
// a default value for the number of
// confirmations, we use it.
defaultConf := uint16(chainCfg.DefaultNumChanConfs)
if defaultConf != 0 {
return defaultConf
}
minConf := uint64(3)
maxConf := uint64(6)
// If this is a wumbo channel, then we'll require the
// max amount of confirmations.
if chanAmt > MaxFundingAmount {
return uint16(maxConf)
}
// If not we return a value scaled linearly
// between 3 and 6, depending on channel size.
// TODO(halseth): Use 1 as minimum?
maxChannelSize := uint64(
lnwire.NewMSatFromSatoshis(MaxFundingAmount))
stake := lnwire.NewMSatFromSatoshis(chanAmt) + pushAmt
conf := maxConf * uint64(stake) / maxChannelSize
if conf < minConf {
conf = minConf
}
if conf > maxConf {
conf = maxConf
}
return uint16(conf)
},
RequiredRemoteDelay: func(chanAmt btcutil.Amount) uint16 {
// We scale the remote CSV delay (the time the
// remote have to claim funds in case of a unilateral
// close) linearly from minRemoteDelay blocks
// for small channels, to maxRemoteDelay blocks
// for channels of size MaxFundingAmount.
// TODO(halseth): Litecoin parameter for LTC.
// In case the user has explicitly specified
// a default value for the remote delay, we
// use it.
defaultDelay := uint16(chainCfg.DefaultRemoteDelay)
if defaultDelay > 0 {
return defaultDelay
}
// If this is a wumbo channel, then we'll require the
// max value.
if chanAmt > MaxFundingAmount {
return maxRemoteDelay
}
// If not we scale according to channel size.
delay := uint16(btcutil.Amount(maxRemoteDelay) *
chanAmt / MaxFundingAmount)
if delay < minRemoteDelay {
delay = minRemoteDelay
}
if delay > maxRemoteDelay {
delay = maxRemoteDelay
}
return delay
},
WatchNewChannel: func(channel *channeldb.OpenChannel,
peerKey *btcec.PublicKey) error {
// First, we'll mark this new peer as a persistent peer
// for re-connection purposes. If the peer is not yet
// tracked or the user hasn't requested it to be perm,
// we'll set false to prevent the server from continuing
// to connect to this peer even if the number of
// channels with this peer is zero.
s.mu.Lock()
pubStr := string(peerKey.SerializeCompressed())
if _, ok := s.persistentPeers[pubStr]; !ok {
s.persistentPeers[pubStr] = false
}
s.mu.Unlock()
// With that taken care of, we'll send this channel to
// the chain arb so it can react to on-chain events.
return s.chainArb.WatchNewChannel(channel)
},
ReportShortChanID: func(chanPoint wire.OutPoint) error {
cid := lnwire.NewChanIDFromOutPoint(&chanPoint)
return s.htlcSwitch.UpdateShortChanID(cid)
},
RequiredRemoteChanReserve: func(chanAmt,
dustLimit btcutil.Amount) btcutil.Amount {
// By default, we'll require the remote peer to maintain
// at least 1% of the total channel capacity at all
// times. If this value ends up dipping below the dust
// limit, then we'll use the dust limit itself as the
// reserve as required by BOLT #2.
reserve := chanAmt / 100
if reserve < dustLimit {
reserve = dustLimit
}
return reserve
},
RequiredRemoteMaxValue: func(chanAmt btcutil.Amount) lnwire.MilliSatoshi {
// By default, we'll allow the remote peer to fully
// utilize the full bandwidth of the channel, minus our
// required reserve.
reserve := lnwire.NewMSatFromSatoshis(chanAmt / 100)
return lnwire.NewMSatFromSatoshis(chanAmt) - reserve
},
RequiredRemoteMaxHTLCs: func(chanAmt btcutil.Amount) uint16 {
if cfg.DefaultRemoteMaxHtlcs > 0 {
return cfg.DefaultRemoteMaxHtlcs
}
// By default, we'll permit them to utilize the full
// channel bandwidth.
return uint16(input.MaxHTLCNumber / 2)
},
ZombieSweeperInterval: 1 * time.Minute,
ReservationTimeout: 10 * time.Minute,
MinChanSize: btcutil.Amount(cfg.MinChanSize),
MaxChanSize: btcutil.Amount(cfg.MaxChanSize),
MaxPendingChannels: cfg.MaxPendingChannels,
RejectPush: cfg.RejectPush,
MaxLocalCSVDelay: chainCfg.MaxLocalDelay,
NotifyOpenChannelEvent: s.channelNotifier.NotifyOpenChannelEvent,
OpenChannelPredicate: chanPredicate,
NotifyPendingOpenChannelEvent: s.channelNotifier.NotifyPendingOpenChannelEvent,
EnableUpfrontShutdown: cfg.EnableUpfrontShutdown,
RegisteredChains: cfg.registeredChains,
MaxAnchorsCommitFeeRate: chainfee.SatPerKVByte(
s.cfg.MaxCommitFeeRateAnchors * 1000).FeePerKWeight(),
DeleteAliasEdge: deleteAliasEdge,
AliasManager: s.aliasMgr,
})
if err != nil {
return nil, err
}
// Next, we'll assemble the sub-system that will maintain an on-disk
// static backup of the latest channel state.
chanNotifier := &channelNotifier{
chanNotifier: s.channelNotifier,
addrs: dbs.ChanStateDB,
}
backupFile := chanbackup.NewMultiFile(cfg.BackupFilePath)
startingChans, err := chanbackup.FetchStaticChanBackups(
s.chanStateDB, s.addrSource,
)
if err != nil {
return nil, err
}
s.chanSubSwapper, err = chanbackup.NewSubSwapper(
startingChans, chanNotifier, s.cc.KeyRing, backupFile,
)
if err != nil {
return nil, err
}
// Assemble a peer notifier which will provide clients with subscriptions
// to peer online and offline events.
s.peerNotifier = peernotifier.New()
// Create a channel event store which monitors all open channels.
s.chanEventStore = chanfitness.NewChannelEventStore(&chanfitness.Config{
SubscribeChannelEvents: func() (subscribe.Subscription, error) {
return s.channelNotifier.SubscribeChannelEvents()
},
SubscribePeerEvents: func() (subscribe.Subscription, error) {
return s.peerNotifier.SubscribePeerEvents()
},
GetOpenChannels: s.chanStateDB.FetchAllOpenChannels,
Clock: clock.NewDefaultClock(),
ReadFlapCount: s.miscDB.ReadFlapCount,
WriteFlapCount: s.miscDB.WriteFlapCounts,
FlapCountTicker: ticker.New(chanfitness.FlapCountFlushRate),
})
if cfg.WtClient.Active {
policy := wtpolicy.DefaultPolicy()
policy.MaxUpdates = cfg.WtClient.MaxUpdates
// We expose the sweep fee rate in sat/vbyte, but the tower
// protocol operations on sat/kw.
sweepRateSatPerVByte := chainfee.SatPerKVByte(
1000 * cfg.WtClient.SweepFeeRate,
)
policy.SweepFeeRate = sweepRateSatPerVByte.FeePerKWeight()
if err := policy.Validate(); err != nil {
return nil, err
}
// authDial is the wrapper around the btrontide.Dial for the
// watchtower.
authDial := func(localKey keychain.SingleKeyECDH,
netAddr *lnwire.NetAddress,
dialer tor.DialFunc) (wtserver.Peer, error) {
return brontide.Dial(
localKey, netAddr, cfg.ConnectionTimeout, dialer,
)
}
// buildBreachRetribution is a call-back that can be used to
// query the BreachRetribution info and channel type given a
// channel ID and commitment height.
buildBreachRetribution := func(chanID lnwire.ChannelID,
commitHeight uint64) (*lnwallet.BreachRetribution,
channeldb.ChannelType, error) {
channel, err := s.chanStateDB.FetchChannelByID(
nil, chanID,
)
if err != nil {
return nil, 0, err
}
br, err := lnwallet.NewBreachRetribution(
channel, commitHeight, 0, nil,
)
if err != nil {
return nil, 0, err
}
return br, channel.ChanType, nil
}
fetchClosedChannel := s.chanStateDB.FetchClosedChannelForID
s.towerClient, err = wtclient.New(&wtclient.Config{
FetchClosedChannel: fetchClosedChannel,
BuildBreachRetribution: buildBreachRetribution,
SessionCloseRange: cfg.WtClient.SessionCloseRange,
ChainNotifier: s.cc.ChainNotifier,
SubscribeChannelEvents: func() (subscribe.Subscription,
error) {
return s.channelNotifier.
SubscribeChannelEvents()
},
Signer: cc.Wallet.Cfg.Signer,
NewAddress: newSweepPkScriptGen(cc.Wallet),
SecretKeyRing: s.cc.KeyRing,
Dial: cfg.net.Dial,
AuthDial: authDial,
DB: dbs.TowerClientDB,
Policy: policy,
ChainHash: *s.cfg.ActiveNetParams.GenesisHash,
MinBackoff: 10 * time.Second,
MaxBackoff: 5 * time.Minute,
MaxTasksInMemQueue: cfg.WtClient.MaxTasksInMemQueue,
})
if err != nil {
return nil, err
}
// Copy the policy for legacy channels and set the blob flag
// signalling support for anchor channels.
anchorPolicy := policy
anchorPolicy.TxPolicy.BlobType |=
blob.Type(blob.FlagAnchorChannel)
s.anchorTowerClient, err = wtclient.New(&wtclient.Config{
FetchClosedChannel: fetchClosedChannel,
BuildBreachRetribution: buildBreachRetribution,
SessionCloseRange: cfg.WtClient.SessionCloseRange,
ChainNotifier: s.cc.ChainNotifier,
SubscribeChannelEvents: func() (subscribe.Subscription,
error) {
return s.channelNotifier.
SubscribeChannelEvents()
},
Signer: cc.Wallet.Cfg.Signer,
NewAddress: newSweepPkScriptGen(cc.Wallet),
SecretKeyRing: s.cc.KeyRing,
Dial: cfg.net.Dial,
AuthDial: authDial,
DB: dbs.TowerClientDB,
Policy: anchorPolicy,
ChainHash: *s.cfg.ActiveNetParams.GenesisHash,
MinBackoff: 10 * time.Second,
MaxBackoff: 5 * time.Minute,
MaxTasksInMemQueue: cfg.WtClient.MaxTasksInMemQueue,
})
if err != nil {
return nil, err
}
}
if len(cfg.ExternalHosts) != 0 {
advertisedIPs := make(map[string]struct{})
for _, addr := range s.currentNodeAnn.Addresses {
advertisedIPs[addr.String()] = struct{}{}
}
s.hostAnn = netann.NewHostAnnouncer(netann.HostAnnouncerConfig{
Hosts: cfg.ExternalHosts,
RefreshTicker: ticker.New(defaultHostSampleInterval),
LookupHost: func(host string) (net.Addr, error) {
return lncfg.ParseAddressString(
host, strconv.Itoa(defaultPeerPort),
cfg.net.ResolveTCPAddr,
)
},
AdvertisedIPs: advertisedIPs,
AnnounceNewIPs: netann.IPAnnouncer(
func(modifier ...netann.NodeAnnModifier) (
lnwire.NodeAnnouncement, error) {
return s.genNodeAnnouncement(
nil, modifier...,
)
}),
})
}
// Create liveness monitor.
s.createLivenessMonitor(cfg, cc)
// Create the connection manager which will be responsible for
// maintaining persistent outbound connections and also accepting new
// incoming connections
cmgr, err := connmgr.New(&connmgr.Config{
Listeners: listeners,
OnAccept: s.InboundPeerConnected,
RetryDuration: time.Second * 5,
TargetOutbound: 100,
Dial: noiseDial(
nodeKeyECDH, s.cfg.net, s.cfg.ConnectionTimeout,
),
OnConnection: s.OutboundPeerConnected,
})
if err != nil {
return nil, err
}
s.connMgr = cmgr
return s, nil
}
// signAliasUpdate takes a ChannelUpdate and returns the signature. This is
// used for option_scid_alias channels where the ChannelUpdate to be sent back
// may differ from what is on disk.
func (s *server) signAliasUpdate(u *lnwire.ChannelUpdate) (*ecdsa.Signature,
error) {
data, err := u.DataToSign()
if err != nil {
return nil, err
}
return s.cc.MsgSigner.SignMessage(s.identityKeyLoc, data, true)
}
// createLivenessMonitor creates a set of health checks using our configured
// values and uses these checks to create a liveness monitor. Available
// health checks,
// - chainHealthCheck (will be disabled for --nochainbackend mode)
// - diskCheck
// - tlsHealthCheck
// - torController, only created when tor is enabled.
//
// If a health check has been disabled by setting attempts to 0, our monitor
// will not run it.
func (s *server) createLivenessMonitor(cfg *Config, cc *chainreg.ChainControl) {
chainBackendAttempts := cfg.HealthChecks.ChainCheck.Attempts
if cfg.Bitcoin.Node == "nochainbackend" {
srvrLog.Info("Disabling chain backend checks for " +
"nochainbackend mode")
chainBackendAttempts = 0
}
chainHealthCheck := healthcheck.NewObservation(
"chain backend",
cc.HealthCheck,
cfg.HealthChecks.ChainCheck.Interval,
cfg.HealthChecks.ChainCheck.Timeout,
cfg.HealthChecks.ChainCheck.Backoff,
chainBackendAttempts,
)
diskCheck := healthcheck.NewObservation(
"disk space",
func() error {
free, err := healthcheck.AvailableDiskSpaceRatio(
cfg.LndDir,
)
if err != nil {
return err
}
// If we have more free space than we require,
// we return a nil error.
if free > cfg.HealthChecks.DiskCheck.RequiredRemaining {
return nil
}
return fmt.Errorf("require: %v free space, got: %v",
cfg.HealthChecks.DiskCheck.RequiredRemaining,
free)
},
cfg.HealthChecks.DiskCheck.Interval,
cfg.HealthChecks.DiskCheck.Timeout,
cfg.HealthChecks.DiskCheck.Backoff,
cfg.HealthChecks.DiskCheck.Attempts,
)
tlsHealthCheck := healthcheck.NewObservation(
"tls",
func() error {
expired, expTime, err := s.tlsManager.IsCertExpired(
s.cc.KeyRing,
)
if err != nil {
return err
}
if expired {
return fmt.Errorf("TLS certificate is "+
"expired as of %v", expTime)
}
// If the certificate is not outdated, no error needs
// to be returned
return nil
},
cfg.HealthChecks.TLSCheck.Interval,
cfg.HealthChecks.TLSCheck.Timeout,
cfg.HealthChecks.TLSCheck.Backoff,
cfg.HealthChecks.TLSCheck.Attempts,
)
checks := []*healthcheck.Observation{
chainHealthCheck, diskCheck, tlsHealthCheck,
}
// If Tor is enabled, add the healthcheck for tor connection.
if s.torController != nil {
torConnectionCheck := healthcheck.NewObservation(
"tor connection",
func() error {
return healthcheck.CheckTorServiceStatus(
s.torController,
s.createNewHiddenService,
)
},
cfg.HealthChecks.TorConnection.Interval,
cfg.HealthChecks.TorConnection.Timeout,
cfg.HealthChecks.TorConnection.Backoff,
cfg.HealthChecks.TorConnection.Attempts,
)
checks = append(checks, torConnectionCheck)
}
// If remote signing is enabled, add the healthcheck for the remote
// signing RPC interface.
if s.cfg.RemoteSigner != nil && s.cfg.RemoteSigner.Enable {
// Because we have two cascading timeouts here, we need to add
// some slack to the "outer" one of them in case the "inner"
// returns exactly on time.
overhead := time.Millisecond * 10
remoteSignerConnectionCheck := healthcheck.NewObservation(
"remote signer connection",
rpcwallet.HealthCheck(
s.cfg.RemoteSigner,
// For the health check we might to be even
// stricter than the initial/normal connect, so
// we use the health check timeout here.
cfg.HealthChecks.RemoteSigner.Timeout,
),
cfg.HealthChecks.RemoteSigner.Interval,
cfg.HealthChecks.RemoteSigner.Timeout+overhead,
cfg.HealthChecks.RemoteSigner.Backoff,
cfg.HealthChecks.RemoteSigner.Attempts,
)
checks = append(checks, remoteSignerConnectionCheck)
}
// If we have not disabled all of our health checks, we create a
// liveness monitor with our configured checks.
s.livenessMonitor = healthcheck.NewMonitor(
&healthcheck.Config{
Checks: checks,
Shutdown: srvrLog.Criticalf,
},
)
}
// Started returns true if the server has been started, and false otherwise.
// NOTE: This function is safe for concurrent access.
func (s *server) Started() bool {
return atomic.LoadInt32(&s.active) != 0
}
// cleaner is used to aggregate "cleanup" functions during an operation that
// starts several subsystems. In case one of the subsystem fails to start
// and a proper resource cleanup is required, the "run" method achieves this
// by running all these added "cleanup" functions.
type cleaner []func() error
// add is used to add a cleanup function to be called when
// the run function is executed.
func (c cleaner) add(cleanup func() error) cleaner {
return append(c, cleanup)
}
// run is used to run all the previousely added cleanup functions.
func (c cleaner) run() {
for i := len(c) - 1; i >= 0; i-- {
if err := c[i](); err != nil {
srvrLog.Infof("Cleanup failed: %v", err)
}
}
}
// Start starts the main daemon server, all requested listeners, and any helper
// goroutines.
// NOTE: This function is safe for concurrent access.
func (s *server) Start() error {
var startErr error
// If one sub system fails to start, the following code ensures that the
// previous started ones are stopped. It also ensures a proper wallet
// shutdown which is important for releasing its resources (boltdb, etc...)
cleanup := cleaner{}
s.start.Do(func() {
if err := s.customMessageServer.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.customMessageServer.Stop)
if s.hostAnn != nil {
if err := s.hostAnn.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.hostAnn.Stop)
}
if s.livenessMonitor != nil {
if err := s.livenessMonitor.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.livenessMonitor.Stop)
}
// Start the notification server. This is used so channel
// management goroutines can be notified when a funding
// transaction reaches a sufficient number of confirmations, or
// when the input for the funding transaction is spent in an
// attempt at an uncooperative close by the counterparty.
if err := s.sigPool.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.sigPool.Stop)
if err := s.writePool.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.writePool.Stop)
if err := s.readPool.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.readPool.Stop)
if err := s.cc.ChainNotifier.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.cc.ChainNotifier.Stop)
if err := s.channelNotifier.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.channelNotifier.Stop)
if err := s.peerNotifier.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(func() error {
return s.peerNotifier.Stop()
})
if err := s.htlcNotifier.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.htlcNotifier.Stop)
if s.towerClient != nil {
if err := s.towerClient.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.towerClient.Stop)
}
if s.anchorTowerClient != nil {
if err := s.anchorTowerClient.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.anchorTowerClient.Stop)
}
if err := s.sweeper.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.sweeper.Stop)
if err := s.utxoNursery.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.utxoNursery.Stop)
if err := s.breachArbiter.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.breachArbiter.Stop)
if err := s.fundingMgr.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.fundingMgr.Stop)
// htlcSwitch must be started before chainArb since the latter
// relies on htlcSwitch to deliver resolution message upon
// start.
if err := s.htlcSwitch.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.htlcSwitch.Stop)
if err := s.interceptableSwitch.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.interceptableSwitch.Stop)
if err := s.chainArb.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.chainArb.Stop)
if err := s.authGossiper.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.authGossiper.Stop)
if err := s.chanRouter.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.chanRouter.Stop)
if err := s.invoices.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.invoices.Stop)
if err := s.sphinx.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.sphinx.Stop)
if err := s.chanStatusMgr.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.chanStatusMgr.Stop)
if err := s.chanEventStore.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(func() error {
s.chanEventStore.Stop()
return nil
})
s.missionControl.RunStoreTicker()
cleanup.add(func() error {
s.missionControl.StopStoreTicker()
return nil
})
// Before we start the connMgr, we'll check to see if we have
// any backups to recover. We do this now as we want to ensure
// that have all the information we need to handle channel
// recovery _before_ we even accept connections from any peers.
chanRestorer := &chanDBRestorer{
db: s.chanStateDB,
secretKeys: s.cc.KeyRing,
chainArb: s.chainArb,
}
if len(s.chansToRestore.PackedSingleChanBackups) != 0 {
err := chanbackup.UnpackAndRecoverSingles(
s.chansToRestore.PackedSingleChanBackups,
s.cc.KeyRing, chanRestorer, s,
)
if err != nil {
startErr = fmt.Errorf("unable to unpack single "+
"backups: %v", err)
return
}
}
if len(s.chansToRestore.PackedMultiChanBackup) != 0 {
err := chanbackup.UnpackAndRecoverMulti(
s.chansToRestore.PackedMultiChanBackup,
s.cc.KeyRing, chanRestorer, s,
)
if err != nil {
startErr = fmt.Errorf("unable to unpack chan "+
"backup: %v", err)
return
}
}
if err := s.chanSubSwapper.Start(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.chanSubSwapper.Stop)
if s.torController != nil {
if err := s.createNewHiddenService(); err != nil {
startErr = err
return
}
cleanup = cleanup.add(s.torController.Stop)
}
if s.natTraversal != nil {
s.wg.Add(1)
go s.watchExternalIP()
}
// Start connmgr last to prevent connections before init.
s.connMgr.Start()
cleanup = cleanup.add(func() error {
s.connMgr.Stop()
return nil
})
// If peers are specified as a config option, we'll add those
// peers first.
for _, peerAddrCfg := range s.cfg.AddPeers {
parsedPubkey, parsedHost, err := lncfg.ParseLNAddressPubkey(
peerAddrCfg,
)
if err != nil {
startErr = fmt.Errorf("unable to parse peer "+
"pubkey from config: %v", err)
return
}
addr, err := parseAddr(parsedHost, s.cfg.net)
if err != nil {
startErr = fmt.Errorf("unable to parse peer "+
"address provided as a config option: "+
"%v", err)
return
}
peerAddr := &lnwire.NetAddress{
IdentityKey: parsedPubkey,
Address: addr,
ChainNet: s.cfg.ActiveNetParams.Net,
}
err = s.ConnectToPeer(
peerAddr, true,
s.cfg.ConnectionTimeout,
)
if err != nil {
startErr = fmt.Errorf("unable to connect to "+
"peer address provided as a config "+
"option: %v", err)
return
}
}
// Subscribe to NodeAnnouncements that advertise new addresses
// our persistent peers.
if err := s.updatePersistentPeerAddrs(); err != nil {
startErr = err
return
}
// With all the relevant sub-systems started, we'll now attempt
// to establish persistent connections to our direct channel
// collaborators within the network. Before doing so however,
// we'll prune our set of link nodes found within the database
// to ensure we don't reconnect to any nodes we no longer have
// open channels with.
if err := s.chanStateDB.PruneLinkNodes(); err != nil {
startErr = err
return
}
if err := s.establishPersistentConnections(); err != nil {
startErr = err
return
}
// setSeedList is a helper function that turns multiple DNS seed
// server tuples from the command line or config file into the
// data structure we need and does a basic formal sanity check
// in the process.
setSeedList := func(tuples []string, genesisHash chainhash.Hash) {
if len(tuples) == 0 {
return
}
result := make([][2]string, len(tuples))
for idx, tuple := range tuples {
tuple = strings.TrimSpace(tuple)
if len(tuple) == 0 {
return
}
servers := strings.Split(tuple, ",")
if len(servers) > 2 || len(servers) == 0 {
srvrLog.Warnf("Ignoring invalid DNS "+
"seed tuple: %v", servers)
return
}
copy(result[idx][:], servers)
}
chainreg.ChainDNSSeeds[genesisHash] = result
}
// Let users overwrite the DNS seed nodes. We only allow them
// for bitcoin mainnet/testnet and litecoin mainnet, all other
// combinations will just be ignored.
if s.cfg.Bitcoin.Active && s.cfg.Bitcoin.MainNet {
setSeedList(
s.cfg.Bitcoin.DNSSeeds,
chainreg.BitcoinMainnetGenesis,
)
}
if s.cfg.Bitcoin.Active && s.cfg.Bitcoin.TestNet3 {
setSeedList(
s.cfg.Bitcoin.DNSSeeds,
chainreg.BitcoinTestnetGenesis,
)
}
if s.cfg.Bitcoin.Active && s.cfg.Bitcoin.SigNet {
setSeedList(
s.cfg.Bitcoin.DNSSeeds,
chainreg.BitcoinSignetGenesis,
)
}
if s.cfg.Litecoin.Active && s.cfg.Litecoin.MainNet {
setSeedList(
s.cfg.Litecoin.DNSSeeds,
chainreg.LitecoinMainnetGenesis,
)
}
// If network bootstrapping hasn't been disabled, then we'll
// configure the set of active bootstrappers, and launch a
// dedicated goroutine to maintain a set of persistent
// connections.
if shouldPeerBootstrap(s.cfg) {
bootstrappers, err := initNetworkBootstrappers(s)
if err != nil {
startErr = err
return
}
s.wg.Add(1)
go s.peerBootstrapper(defaultMinPeers, bootstrappers)
} else {
srvrLog.Infof("Auto peer bootstrapping is disabled")
}
// Set the active flag now that we've completed the full
// startup.
atomic.StoreInt32(&s.active, 1)
})
if startErr != nil {
cleanup.run()
}
return startErr
}
// Stop gracefully shutsdown the main daemon server. This function will signal
// any active goroutines, or helper objects to exit, then blocks until they've
// all successfully exited. Additionally, any/all listeners are closed.
// NOTE: This function is safe for concurrent access.
func (s *server) Stop() error {
s.stop.Do(func() {
atomic.StoreInt32(&s.stopping, 1)
close(s.quit)
// Shutdown connMgr first to prevent conns during shutdown.
s.connMgr.Stop()
// Shutdown the wallet, funding manager, and the rpc server.
if err := s.chanStatusMgr.Stop(); err != nil {
srvrLog.Warnf("failed to stop chanStatusMgr: %v", err)
}
if err := s.htlcSwitch.Stop(); err != nil {
srvrLog.Warnf("failed to stop htlcSwitch: %v", err)
}
if err := s.sphinx.Stop(); err != nil {
srvrLog.Warnf("failed to stop sphinx: %v", err)
}
if err := s.invoices.Stop(); err != nil {
srvrLog.Warnf("failed to stop invoices: %v", err)
}
if err := s.chanRouter.Stop(); err != nil {
srvrLog.Warnf("failed to stop chanRouter: %v", err)
}
if err := s.chainArb.Stop(); err != nil {
srvrLog.Warnf("failed to stop chainArb: %v", err)
}
if err := s.fundingMgr.Stop(); err != nil {
srvrLog.Warnf("failed to stop fundingMgr: %v", err)
}
if err := s.breachArbiter.Stop(); err != nil {
srvrLog.Warnf("failed to stop breachArbiter: %v", err)
}
if err := s.utxoNursery.Stop(); err != nil {
srvrLog.Warnf("failed to stop utxoNursery: %v", err)
}
if err := s.authGossiper.Stop(); err != nil {
srvrLog.Warnf("failed to stop authGossiper: %v", err)
}
if err := s.sweeper.Stop(); err != nil {
srvrLog.Warnf("failed to stop sweeper: %v", err)
}
if err := s.channelNotifier.Stop(); err != nil {
srvrLog.Warnf("failed to stop channelNotifier: %v", err)
}
if err := s.peerNotifier.Stop(); err != nil {
srvrLog.Warnf("failed to stop peerNotifier: %v", err)
}
if err := s.htlcNotifier.Stop(); err != nil {
srvrLog.Warnf("failed to stop htlcNotifier: %v", err)
}
if err := s.chanSubSwapper.Stop(); err != nil {
srvrLog.Warnf("failed to stop chanSubSwapper: %v", err)
}
if err := s.cc.ChainNotifier.Stop(); err != nil {
srvrLog.Warnf("Unable to stop ChainNotifier: %v", err)
}
s.chanEventStore.Stop()
s.missionControl.StopStoreTicker()
// Disconnect from each active peers to ensure that
// peerTerminationWatchers signal completion to each peer.
for _, peer := range s.Peers() {
err := s.DisconnectPeer(peer.IdentityKey())
if err != nil {
srvrLog.Warnf("could not disconnect peer: %v"+
"received error: %v", peer.IdentityKey(),
err,
)
}
}
// Now that all connections have been torn down, stop the tower
// client which will reliably flush all queued states to the
// tower. If this is halted for any reason, the force quit timer
// will kick in and abort to allow this method to return.
if s.towerClient != nil {
if err := s.towerClient.Stop(); err != nil {
srvrLog.Warnf("Unable to shut down tower "+
"client: %v", err)
}
}
if s.anchorTowerClient != nil {
if err := s.anchorTowerClient.Stop(); err != nil {
srvrLog.Warnf("Unable to shut down anchor "+
"tower client: %v", err)
}
}
if s.hostAnn != nil {
if err := s.hostAnn.Stop(); err != nil {
srvrLog.Warnf("unable to shut down host "+
"annoucner: %v", err)
}
}
if s.livenessMonitor != nil {
if err := s.livenessMonitor.Stop(); err != nil {
srvrLog.Warnf("unable to shutdown liveness "+
"monitor: %v", err)
}
}
// Wait for all lingering goroutines to quit.
s.wg.Wait()
s.sigPool.Stop()
s.writePool.Stop()
s.readPool.Stop()
})
return nil
}
// Stopped returns true if the server has been instructed to shutdown.
// NOTE: This function is safe for concurrent access.
func (s *server) Stopped() bool {
return atomic.LoadInt32(&s.stopping) != 0
}
// configurePortForwarding attempts to set up port forwarding for the different
// ports that the server will be listening on.
//
// NOTE: This should only be used when using some kind of NAT traversal to
// automatically set up forwarding rules.
func (s *server) configurePortForwarding(ports ...uint16) ([]string, error) {
ip, err := s.natTraversal.ExternalIP()
if err != nil {
return nil, err
}
s.lastDetectedIP = ip
externalIPs := make([]string, 0, len(ports))
for _, port := range ports {
if err := s.natTraversal.AddPortMapping(port); err != nil {
srvrLog.Debugf("Unable to forward port %d: %v", port, err)
continue
}
hostIP := fmt.Sprintf("%v:%d", ip, port)
externalIPs = append(externalIPs, hostIP)
}
return externalIPs, nil
}
// removePortForwarding attempts to clear the forwarding rules for the different
// ports the server is currently listening on.
//
// NOTE: This should only be used when using some kind of NAT traversal to
// automatically set up forwarding rules.
func (s *server) removePortForwarding() {
forwardedPorts := s.natTraversal.ForwardedPorts()
for _, port := range forwardedPorts {
if err := s.natTraversal.DeletePortMapping(port); err != nil {
srvrLog.Errorf("Unable to remove forwarding rules for "+
"port %d: %v", port, err)
}
}
}
// watchExternalIP continuously checks for an updated external IP address every
// 15 minutes. Once a new IP address has been detected, it will automatically
// handle port forwarding rules and send updated node announcements to the
// currently connected peers.
//
// NOTE: This MUST be run as a goroutine.
func (s *server) watchExternalIP() {
defer s.wg.Done()
// Before exiting, we'll make sure to remove the forwarding rules set
// up by the server.
defer s.removePortForwarding()
// Keep track of the external IPs set by the user to avoid replacing
// them when detecting a new IP.
ipsSetByUser := make(map[string]struct{})
for _, ip := range s.cfg.ExternalIPs {
ipsSetByUser[ip.String()] = struct{}{}
}
forwardedPorts := s.natTraversal.ForwardedPorts()
ticker := time.NewTicker(15 * time.Minute)
defer ticker.Stop()
out:
for {
select {
case <-ticker.C:
// We'll start off by making sure a new IP address has
// been detected.
ip, err := s.natTraversal.ExternalIP()
if err != nil {
srvrLog.Debugf("Unable to retrieve the "+
"external IP address: %v", err)
continue
}
// Periodically renew the NAT port forwarding.
for _, port := range forwardedPorts {
err := s.natTraversal.AddPortMapping(port)
if err != nil {
srvrLog.Warnf("Unable to automatically "+
"re-create port forwarding using %s: %v",
s.natTraversal.Name(), err)
} else {
srvrLog.Debugf("Automatically re-created "+
"forwarding for port %d using %s to "+
"advertise external IP",
port, s.natTraversal.Name())
}
}
if ip.Equal(s.lastDetectedIP) {
continue
}
srvrLog.Infof("Detected new external IP address %s", ip)
// Next, we'll craft the new addresses that will be
// included in the new node announcement and advertised
// to the network. Each address will consist of the new
// IP detected and one of the currently advertised
// ports.
var newAddrs []net.Addr
for _, port := range forwardedPorts {
hostIP := fmt.Sprintf("%v:%d", ip, port)
addr, err := net.ResolveTCPAddr("tcp", hostIP)
if err != nil {
srvrLog.Debugf("Unable to resolve "+
"host %v: %v", addr, err)
continue
}
newAddrs = append(newAddrs, addr)
}
// Skip the update if we weren't able to resolve any of
// the new addresses.
if len(newAddrs) == 0 {
srvrLog.Debug("Skipping node announcement " +
"update due to not being able to " +
"resolve any new addresses")
continue
}
// Now, we'll need to update the addresses in our node's
// announcement in order to propagate the update
// throughout the network. We'll only include addresses
// that have a different IP from the previous one, as
// the previous IP is no longer valid.
currentNodeAnn := s.getNodeAnnouncement()
for _, addr := range currentNodeAnn.Addresses {
host, _, err := net.SplitHostPort(addr.String())
if err != nil {
srvrLog.Debugf("Unable to determine "+
"host from address %v: %v",
addr, err)
continue
}
// We'll also make sure to include external IPs
// set manually by the user.
_, setByUser := ipsSetByUser[addr.String()]
if setByUser || host != s.lastDetectedIP.String() {
newAddrs = append(newAddrs, addr)
}
}
// Then, we'll generate a new timestamped node
// announcement with the updated addresses and broadcast
// it to our peers.
newNodeAnn, err := s.genNodeAnnouncement(
nil, netann.NodeAnnSetAddrs(newAddrs),
)
if err != nil {
srvrLog.Debugf("Unable to generate new node "+
"announcement: %v", err)
continue
}
err = s.BroadcastMessage(nil, &newNodeAnn)
if err != nil {
srvrLog.Debugf("Unable to broadcast new node "+
"announcement to peers: %v", err)
continue
}
// Finally, update the last IP seen to the current one.
s.lastDetectedIP = ip
case <-s.quit:
break out
}
}
}
// initNetworkBootstrappers initializes a set of network peer bootstrappers
// based on the server, and currently active bootstrap mechanisms as defined
// within the current configuration.
func initNetworkBootstrappers(s *server) ([]discovery.NetworkPeerBootstrapper, error) {
srvrLog.Infof("Initializing peer network bootstrappers!")
var bootStrappers []discovery.NetworkPeerBootstrapper
// First, we'll create an instance of the ChannelGraphBootstrapper as
// this can be used by default if we've already partially seeded the
// network.
chanGraph := autopilot.ChannelGraphFromDatabase(s.graphDB)
graphBootstrapper, err := discovery.NewGraphBootstrapper(chanGraph)
if err != nil {
return nil, err
}
bootStrappers = append(bootStrappers, graphBootstrapper)
// If this isn't simnet mode, then one of our additional bootstrapping
// sources will be the set of running DNS seeds.
if !s.cfg.Bitcoin.SimNet || !s.cfg.Litecoin.SimNet {
dnsSeeds, ok := chainreg.ChainDNSSeeds[*s.cfg.ActiveNetParams.GenesisHash]
// If we have a set of DNS seeds for this chain, then we'll add
// it as an additional bootstrapping source.
if ok {
srvrLog.Infof("Creating DNS peer bootstrapper with "+
"seeds: %v", dnsSeeds)
dnsBootStrapper := discovery.NewDNSSeedBootstrapper(
dnsSeeds, s.cfg.net, s.cfg.ConnectionTimeout,
)
bootStrappers = append(bootStrappers, dnsBootStrapper)
}
}
return bootStrappers, nil
}
// createBootstrapIgnorePeers creates a map of peers that the bootstrap process
// needs to ignore, which is made of three parts,
// - the node itself needs to be skipped as it doesn't make sense to connect
// to itself.
// - the peers that already have connections with, as in s.peersByPub.
// - the peers that we are attempting to connect, as in s.persistentPeers.
func (s *server) createBootstrapIgnorePeers() map[autopilot.NodeID]struct{} {
s.mu.RLock()
defer s.mu.RUnlock()
ignore := make(map[autopilot.NodeID]struct{})
// We should ignore ourselves from bootstrapping.
selfKey := autopilot.NewNodeID(s.identityECDH.PubKey())
ignore[selfKey] = struct{}{}
// Ignore all connected peers.
for _, peer := range s.peersByPub {
nID := autopilot.NewNodeID(peer.IdentityKey())
ignore[nID] = struct{}{}
}
// Ignore all persistent peers as they have a dedicated reconnecting
// process.
for pubKeyStr := range s.persistentPeers {
var nID autopilot.NodeID
copy(nID[:], []byte(pubKeyStr))
ignore[nID] = struct{}{}
}
return ignore
}
// peerBootstrapper is a goroutine which is tasked with attempting to establish
// and maintain a target minimum number of outbound connections. With this
// invariant, we ensure that our node is connected to a diverse set of peers
// and that nodes newly joining the network receive an up to date network view
// as soon as possible.
func (s *server) peerBootstrapper(numTargetPeers uint32,
bootstrappers []discovery.NetworkPeerBootstrapper) {
defer s.wg.Done()
// Before we continue, init the ignore peers map.
ignoreList := s.createBootstrapIgnorePeers()
// We'll start off by aggressively attempting connections to peers in
// order to be a part of the network as soon as possible.
s.initialPeerBootstrap(ignoreList, numTargetPeers, bootstrappers)
// Once done, we'll attempt to maintain our target minimum number of
// peers.
//
// We'll use a 15 second backoff, and double the time every time an
// epoch fails up to a ceiling.
backOff := time.Second * 15
// We'll create a new ticker to wake us up every 15 seconds so we can
// see if we've reached our minimum number of peers.
sampleTicker := time.NewTicker(backOff)
defer sampleTicker.Stop()
// We'll use the number of attempts and errors to determine if we need
// to increase the time between discovery epochs.
var epochErrors uint32 // To be used atomically.
var epochAttempts uint32
for {
select {
// The ticker has just woken us up, so we'll need to check if
// we need to attempt to connect our to any more peers.
case <-sampleTicker.C:
// Obtain the current number of peers, so we can gauge
// if we need to sample more peers or not.
s.mu.RLock()
numActivePeers := uint32(len(s.peersByPub))
s.mu.RUnlock()
// If we have enough peers, then we can loop back
// around to the next round as we're done here.
if numActivePeers >= numTargetPeers {
continue
}
// If all of our attempts failed during this last back
// off period, then will increase our backoff to 5
// minute ceiling to avoid an excessive number of
// queries
//
// TODO(roasbeef): add reverse policy too?
if epochAttempts > 0 &&
atomic.LoadUint32(&epochErrors) >= epochAttempts {
sampleTicker.Stop()
backOff *= 2
if backOff > bootstrapBackOffCeiling {
backOff = bootstrapBackOffCeiling
}
srvrLog.Debugf("Backing off peer bootstrapper to "+
"%v", backOff)
sampleTicker = time.NewTicker(backOff)
continue
}
atomic.StoreUint32(&epochErrors, 0)
epochAttempts = 0
// Since we know need more peers, we'll compute the
// exact number we need to reach our threshold.
numNeeded := numTargetPeers - numActivePeers
srvrLog.Debugf("Attempting to obtain %v more network "+
"peers", numNeeded)
// With the number of peers we need calculated, we'll
// query the network bootstrappers to sample a set of
// random addrs for us.
//
// Before we continue, get a copy of the ignore peers
// map.
ignoreList = s.createBootstrapIgnorePeers()
peerAddrs, err := discovery.MultiSourceBootstrap(
ignoreList, numNeeded*2, bootstrappers...,
)
if err != nil {
srvrLog.Errorf("Unable to retrieve bootstrap "+
"peers: %v", err)
continue
}
// Finally, we'll launch a new goroutine for each
// prospective peer candidates.
for _, addr := range peerAddrs {
epochAttempts++
go func(a *lnwire.NetAddress) {
// TODO(roasbeef): can do AS, subnet,
// country diversity, etc
errChan := make(chan error, 1)
s.connectToPeer(
a, errChan,
s.cfg.ConnectionTimeout,
)
select {
case err := <-errChan:
if err == nil {
return
}
srvrLog.Errorf("Unable to "+
"connect to %v: %v",
a, err)
atomic.AddUint32(&epochErrors, 1)
case <-s.quit:
}
}(addr)
}
case <-s.quit:
return
}
}
}
// bootstrapBackOffCeiling is the maximum amount of time we'll wait between
// failed attempts to locate a set of bootstrap peers. We'll slowly double our
// query back off each time we encounter a failure.
const bootstrapBackOffCeiling = time.Minute * 5
// initialPeerBootstrap attempts to continuously connect to peers on startup
// until the target number of peers has been reached. This ensures that nodes
// receive an up to date network view as soon as possible.
func (s *server) initialPeerBootstrap(ignore map[autopilot.NodeID]struct{},
numTargetPeers uint32,
bootstrappers []discovery.NetworkPeerBootstrapper) {
srvrLog.Debugf("Init bootstrap with targetPeers=%v, bootstrappers=%v, "+
"ignore=%v", numTargetPeers, len(bootstrappers), len(ignore))
// We'll start off by waiting 2 seconds between failed attempts, then
// double each time we fail until we hit the bootstrapBackOffCeiling.
var delaySignal <-chan time.Time
delayTime := time.Second * 2
// As want to be more aggressive, we'll use a lower back off celling
// then the main peer bootstrap logic.
backOffCeiling := bootstrapBackOffCeiling / 5
for attempts := 0; ; attempts++ {
// Check if the server has been requested to shut down in order
// to prevent blocking.
if s.Stopped() {
return
}
// We can exit our aggressive initial peer bootstrapping stage
// if we've reached out target number of peers.
s.mu.RLock()
numActivePeers := uint32(len(s.peersByPub))
s.mu.RUnlock()
if numActivePeers >= numTargetPeers {
return
}
if attempts > 0 {
srvrLog.Debugf("Waiting %v before trying to locate "+
"bootstrap peers (attempt #%v)", delayTime,
attempts)
// We've completed at least one iterating and haven't
// finished, so we'll start to insert a delay period
// between each attempt.
delaySignal = time.After(delayTime)
select {
case <-delaySignal:
case <-s.quit:
return
}
// After our delay, we'll double the time we wait up to
// the max back off period.
delayTime *= 2
if delayTime > backOffCeiling {
delayTime = backOffCeiling
}
}
// Otherwise, we'll request for the remaining number of peers
// in order to reach our target.
peersNeeded := numTargetPeers - numActivePeers
bootstrapAddrs, err := discovery.MultiSourceBootstrap(
ignore, peersNeeded, bootstrappers...,
)
if err != nil {
srvrLog.Errorf("Unable to retrieve initial bootstrap "+
"peers: %v", err)
continue
}
// Then, we'll attempt to establish a connection to the
// different peer addresses retrieved by our bootstrappers.
var wg sync.WaitGroup
for _, bootstrapAddr := range bootstrapAddrs {
wg.Add(1)
go func(addr *lnwire.NetAddress) {
defer wg.Done()
errChan := make(chan error, 1)
go s.connectToPeer(
addr, errChan, s.cfg.ConnectionTimeout,
)
// We'll only allow this connection attempt to
// take up to 3 seconds. This allows us to move
// quickly by discarding peers that are slowing
// us down.
select {
case err := <-errChan:
if err == nil {
return
}
srvrLog.Errorf("Unable to connect to "+
"%v: %v", addr, err)
// TODO: tune timeout? 3 seconds might be *too*
// aggressive but works well.
case <-time.After(3 * time.Second):
srvrLog.Tracef("Skipping peer %v due "+
"to not establishing a "+
"connection within 3 seconds",
addr)
case <-s.quit:
}
}(bootstrapAddr)
}
wg.Wait()
}
}
// createNewHiddenService automatically sets up a v2 or v3 onion service in
// order to listen for inbound connections over Tor.
func (s *server) createNewHiddenService() error {
// Determine the different ports the server is listening on. The onion
// service's virtual port will map to these ports and one will be picked
// at random when the onion service is being accessed.
listenPorts := make([]int, 0, len(s.listenAddrs))
for _, listenAddr := range s.listenAddrs {
port := listenAddr.(*net.TCPAddr).Port
listenPorts = append(listenPorts, port)
}
encrypter, err := lnencrypt.KeyRingEncrypter(s.cc.KeyRing)
if err != nil {
return err
}
// Once the port mapping has been set, we can go ahead and automatically
// create our onion service. The service's private key will be saved to
// disk in order to regain access to this service when restarting `lnd`.
onionCfg := tor.AddOnionConfig{
VirtualPort: defaultPeerPort,
TargetPorts: listenPorts,
Store: tor.NewOnionFile(
s.cfg.Tor.PrivateKeyPath, 0600, s.cfg.Tor.EncryptKey,
encrypter,
),
}
switch {
case s.cfg.Tor.V2:
onionCfg.Type = tor.V2
case s.cfg.Tor.V3:
onionCfg.Type = tor.V3
}
addr, err := s.torController.AddOnion(onionCfg)
if err != nil {
return err
}
// Now that the onion service has been created, we'll add the onion
// address it can be reached at to our list of advertised addresses.
newNodeAnn, err := s.genNodeAnnouncement(
nil, func(currentAnn *lnwire.NodeAnnouncement) {
currentAnn.Addresses = append(currentAnn.Addresses, addr)
},
)
if err != nil {
return fmt.Errorf("unable to generate new node "+
"announcement: %v", err)
}
// Finally, we'll update the on-disk version of our announcement so it
// will eventually propagate to nodes in the network.
selfNode := &channeldb.LightningNode{
HaveNodeAnnouncement: true,
LastUpdate: time.Unix(int64(newNodeAnn.Timestamp), 0),
Addresses: newNodeAnn.Addresses,
Alias: newNodeAnn.Alias.String(),
Features: lnwire.NewFeatureVector(
newNodeAnn.Features, lnwire.Features,
),
Color: newNodeAnn.RGBColor,
AuthSigBytes: newNodeAnn.Signature.ToSignatureBytes(),
}
copy(selfNode.PubKeyBytes[:], s.identityECDH.PubKey().SerializeCompressed())
if err := s.graphDB.SetSourceNode(selfNode); err != nil {
return fmt.Errorf("can't set self node: %v", err)
}
return nil
}
// findChannel finds a channel given a public key and ChannelID. It is an
// optimization that is quicker than seeking for a channel given only the
// ChannelID.
func (s *server) findChannel(node *btcec.PublicKey, chanID lnwire.ChannelID) (
*channeldb.OpenChannel, error) {
nodeChans, err := s.chanStateDB.FetchOpenChannels(node)
if err != nil {
return nil, err
}
for _, channel := range nodeChans {
if chanID.IsChanPoint(&channel.FundingOutpoint) {
return channel, nil
}
}
return nil, fmt.Errorf("unable to find channel")
}
// getNodeAnnouncement fetches the current, fully signed node announcement.
func (s *server) getNodeAnnouncement() lnwire.NodeAnnouncement {
s.mu.Lock()
defer s.mu.Unlock()
return *s.currentNodeAnn
}
// genNodeAnnouncement generates and returns the current fully signed node
// announcement. The time stamp of the announcement will be updated in order
// to ensure it propagates through the network.
func (s *server) genNodeAnnouncement(features *lnwire.RawFeatureVector,
modifiers ...netann.NodeAnnModifier) (lnwire.NodeAnnouncement, error) {
s.mu.Lock()
defer s.mu.Unlock()
// First, try to update our feature manager with the updated set of
// features.
if features != nil {
proposedFeatures := map[feature.Set]*lnwire.RawFeatureVector{
feature.SetNodeAnn: features,
}
err := s.featureMgr.UpdateFeatureSets(proposedFeatures)
if err != nil {
return lnwire.NodeAnnouncement{}, err
}
// If we could successfully update our feature manager, add
// an update modifier to include these new features to our
// set.
modifiers = append(
modifiers, netann.NodeAnnSetFeatures(features),
)
}
// Always update the timestamp when refreshing to ensure the update
// propagates.
modifiers = append(modifiers, netann.NodeAnnSetTimestamp)
// Apply the requested changes to the node announcement.
for _, modifier := range modifiers {
modifier(s.currentNodeAnn)
}
// Sign a new update after applying all of the passed modifiers.
err := netann.SignNodeAnnouncement(
s.nodeSigner, s.identityKeyLoc, s.currentNodeAnn,
)
if err != nil {
return lnwire.NodeAnnouncement{}, err
}
return *s.currentNodeAnn, nil
}
// updateAndBrodcastSelfNode generates a new node announcement
// applying the giving modifiers and updating the time stamp
// to ensure it propagates through the network. Then it brodcasts
// it to the network.
func (s *server) updateAndBrodcastSelfNode(features *lnwire.RawFeatureVector,
modifiers ...netann.NodeAnnModifier) error {
newNodeAnn, err := s.genNodeAnnouncement(features, modifiers...)
if err != nil {
return fmt.Errorf("unable to generate new node "+
"announcement: %v", err)
}
// Update the on-disk version of our announcement.
// Load and modify self node istead of creating anew instance so we
// don't risk overwriting any existing values.
selfNode, err := s.graphDB.SourceNode()
if err != nil {
return fmt.Errorf("unable to get current source node: %v", err)
}
selfNode.HaveNodeAnnouncement = true
selfNode.LastUpdate = time.Unix(int64(newNodeAnn.Timestamp), 0)
selfNode.Addresses = newNodeAnn.Addresses
selfNode.Alias = newNodeAnn.Alias.String()
selfNode.Features = s.featureMgr.Get(feature.SetNodeAnn)
selfNode.Color = newNodeAnn.RGBColor
selfNode.AuthSigBytes = newNodeAnn.Signature.ToSignatureBytes()
copy(selfNode.PubKeyBytes[:], s.identityECDH.PubKey().SerializeCompressed())
if err := s.graphDB.SetSourceNode(selfNode); err != nil {
return fmt.Errorf("can't set self node: %v", err)
}
// Finally, propagate it to the nodes in the network.
err = s.BroadcastMessage(nil, &newNodeAnn)
if err != nil {
rpcsLog.Debugf("Unable to broadcast new node "+
"announcement to peers: %v", err)
return err
}
return nil
}
type nodeAddresses struct {
pubKey *btcec.PublicKey
addresses []net.Addr
}
// establishPersistentConnections attempts to establish persistent connections
// to all our direct channel collaborators. In order to promote liveness of our
// active channels, we instruct the connection manager to attempt to establish
// and maintain persistent connections to all our direct channel counterparties.
func (s *server) establishPersistentConnections() error {
// nodeAddrsMap stores the combination of node public keys and addresses
// that we'll attempt to reconnect to. PubKey strings are used as keys
// since other PubKey forms can't be compared.
nodeAddrsMap := map[string]*nodeAddresses{}
// Iterate through the list of LinkNodes to find addresses we should
// attempt to connect to based on our set of previous connections. Set
// the reconnection port to the default peer port.
linkNodes, err := s.chanStateDB.LinkNodeDB().FetchAllLinkNodes()
if err != nil && err != channeldb.ErrLinkNodesNotFound {
return err
}
for _, node := range linkNodes {
pubStr := string(node.IdentityPub.SerializeCompressed())
nodeAddrs := &nodeAddresses{
pubKey: node.IdentityPub,
addresses: node.Addresses,
}
nodeAddrsMap[pubStr] = nodeAddrs
}
// After checking our previous connections for addresses to connect to,
// iterate through the nodes in our channel graph to find addresses
// that have been added via NodeAnnouncement messages.
sourceNode, err := s.graphDB.SourceNode()
if err != nil {
return err
}
// TODO(roasbeef): instead iterate over link nodes and query graph for
// each of the nodes.
selfPub := s.identityECDH.PubKey().SerializeCompressed()
err = sourceNode.ForEachChannel(nil, func(
tx kvdb.RTx,
chanInfo *channeldb.ChannelEdgeInfo,
policy, _ *channeldb.ChannelEdgePolicy) error {
// If the remote party has announced the channel to us, but we
// haven't yet, then we won't have a policy. However, we don't
// need this to connect to the peer, so we'll log it and move on.
if policy == nil {
srvrLog.Warnf("No channel policy found for "+
"ChannelPoint(%v): ", chanInfo.ChannelPoint)
}
// We'll now fetch the peer opposite from us within this
// channel so we can queue up a direct connection to them.
channelPeer, err := chanInfo.FetchOtherNode(tx, selfPub)
if err != nil {
return fmt.Errorf("unable to fetch channel peer for "+
"ChannelPoint(%v): %v", chanInfo.ChannelPoint,
err)
}
pubStr := string(channelPeer.PubKeyBytes[:])
// Add all unique addresses from channel
// graph/NodeAnnouncements to the list of addresses we'll
// connect to for this peer.
addrSet := make(map[string]net.Addr)
for _, addr := range channelPeer.Addresses {
switch addr.(type) {
case *net.TCPAddr:
addrSet[addr.String()] = addr
// We'll only attempt to connect to Tor addresses if Tor
// outbound support is enabled.
case *tor.OnionAddr:
if s.cfg.Tor.Active {
addrSet[addr.String()] = addr
}
}
}
// If this peer is also recorded as a link node, we'll add any
// additional addresses that have not already been selected.
linkNodeAddrs, ok := nodeAddrsMap[pubStr]
if ok {
for _, lnAddress := range linkNodeAddrs.addresses {
switch lnAddress.(type) {
case *net.TCPAddr:
addrSet[lnAddress.String()] = lnAddress
// We'll only attempt to connect to Tor
// addresses if Tor outbound support is enabled.
case *tor.OnionAddr:
if s.cfg.Tor.Active {
addrSet[lnAddress.String()] = lnAddress
}
}
}
}
// Construct a slice of the deduped addresses.
var addrs []net.Addr
for _, addr := range addrSet {
addrs = append(addrs, addr)
}
n := &nodeAddresses{
addresses: addrs,
}
n.pubKey, err = channelPeer.PubKey()
if err != nil {
return err
}
nodeAddrsMap[pubStr] = n
return nil
})
if err != nil && err != channeldb.ErrGraphNoEdgesFound {
return err
}
srvrLog.Debugf("Establishing %v persistent connections on start",
len(nodeAddrsMap))
// Acquire and hold server lock until all persistent connection requests
// have been recorded and sent to the connection manager.
s.mu.Lock()
defer s.mu.Unlock()
// Iterate through the combined list of addresses from prior links and
// node announcements and attempt to reconnect to each node.
var numOutboundConns int
for pubStr, nodeAddr := range nodeAddrsMap {
// Add this peer to the set of peers we should maintain a
// persistent connection with. We set the value to false to
// indicate that we should not continue to reconnect if the
// number of channels returns to zero, since this peer has not
// been requested as perm by the user.
s.persistentPeers[pubStr] = false
if _, ok := s.persistentPeersBackoff[pubStr]; !ok {
s.persistentPeersBackoff[pubStr] = s.cfg.MinBackoff
}
for _, address := range nodeAddr.addresses {
// Create a wrapper address which couples the IP and
// the pubkey so the brontide authenticated connection
// can be established.
lnAddr := &lnwire.NetAddress{
IdentityKey: nodeAddr.pubKey,
Address: address,
}
s.persistentPeerAddrs[pubStr] = append(
s.persistentPeerAddrs[pubStr], lnAddr)
}
// We'll connect to the first 10 peers immediately, then
// randomly stagger any remaining connections if the
// stagger initial reconnect flag is set. This ensures
// that mobile nodes or nodes with a small number of
// channels obtain connectivity quickly, but larger
// nodes are able to disperse the costs of connecting to
// all peers at once.
if numOutboundConns < numInstantInitReconnect ||
!s.cfg.StaggerInitialReconnect {
go s.connectToPersistentPeer(pubStr)
} else {
go s.delayInitialReconnect(pubStr)
}
numOutboundConns++
}
return nil
}
// delayInitialReconnect will attempt a reconnection to the given peer after
// sampling a value for the delay between 0s and the maxInitReconnectDelay.
//
// NOTE: This method MUST be run as a goroutine.
func (s *server) delayInitialReconnect(pubStr string) {
delay := time.Duration(prand.Intn(maxInitReconnectDelay)) * time.Second
select {
case <-time.After(delay):
s.connectToPersistentPeer(pubStr)
case <-s.quit:
}
}
// prunePersistentPeerConnection removes all internal state related to
// persistent connections to a peer within the server. This is used to avoid
// persistent connection retries to peers we do not have any open channels with.
func (s *server) prunePersistentPeerConnection(compressedPubKey [33]byte) {
pubKeyStr := string(compressedPubKey[:])
s.mu.Lock()
if perm, ok := s.persistentPeers[pubKeyStr]; ok && !perm {
delete(s.persistentPeers, pubKeyStr)
delete(s.persistentPeersBackoff, pubKeyStr)
delete(s.persistentPeerAddrs, pubKeyStr)
s.cancelConnReqs(pubKeyStr, nil)
s.mu.Unlock()
srvrLog.Infof("Pruned peer %x from persistent connections, "+
"peer has no open channels", compressedPubKey)
return
}
s.mu.Unlock()
}
// BroadcastMessage sends a request to the server to broadcast a set of
// messages to all peers other than the one specified by the `skips` parameter.
// All messages sent via BroadcastMessage will be queued for lazy delivery to
// the target peers.
//
// NOTE: This function is safe for concurrent access.
func (s *server) BroadcastMessage(skips map[route.Vertex]struct{},
msgs ...lnwire.Message) error {
// Filter out peers found in the skips map. We synchronize access to
// peersByPub throughout this process to ensure we deliver messages to
// exact set of peers present at the time of invocation.
s.mu.RLock()
peers := make([]*peer.Brontide, 0, len(s.peersByPub))
for pubStr, sPeer := range s.peersByPub {
if skips != nil {
if _, ok := skips[sPeer.PubKey()]; ok {
srvrLog.Tracef("Skipping %x in broadcast with "+
"pubStr=%x", sPeer.PubKey(), pubStr)
continue
}
}
peers = append(peers, sPeer)
}
s.mu.RUnlock()
// Iterate over all known peers, dispatching a go routine to enqueue
// all messages to each of peers.
var wg sync.WaitGroup
for _, sPeer := range peers {
srvrLog.Debugf("Sending %v messages to peer %x", len(msgs),
sPeer.PubKey())
// Dispatch a go routine to enqueue all messages to this peer.
wg.Add(1)
s.wg.Add(1)
go func(p lnpeer.Peer) {
defer s.wg.Done()
defer wg.Done()
p.SendMessageLazy(false, msgs...)
}(sPeer)
}
// Wait for all messages to have been dispatched before returning to
// caller.
wg.Wait()
return nil
}
// NotifyWhenOnline can be called by other subsystems to get notified when a
// particular peer comes online. The peer itself is sent across the peerChan.
//
// NOTE: This function is safe for concurrent access.
func (s *server) NotifyWhenOnline(peerKey [33]byte,
peerChan chan<- lnpeer.Peer) {
s.mu.Lock()
// Compute the target peer's identifier.
pubStr := string(peerKey[:])
// Check if peer is connected.
peer, ok := s.peersByPub[pubStr]
if ok {
// Unlock here so that the mutex isn't held while we are
// waiting for the peer to become active.
s.mu.Unlock()
// Wait until the peer signals that it is actually active
// rather than only in the server's maps.
select {
case <-peer.ActiveSignal():
case <-peer.QuitSignal():
// The peer quit, so we'll add the channel to the slice
// and return.
s.mu.Lock()
s.peerConnectedListeners[pubStr] = append(
s.peerConnectedListeners[pubStr], peerChan,
)
s.mu.Unlock()
return
}
// Connected, can return early.
srvrLog.Debugf("Notifying that peer %x is online", peerKey)
select {
case peerChan <- peer:
case <-s.quit:
}
return
}
// Not connected, store this listener such that it can be notified when
// the peer comes online.
s.peerConnectedListeners[pubStr] = append(
s.peerConnectedListeners[pubStr], peerChan,
)
s.mu.Unlock()
}
// NotifyWhenOffline delivers a notification to the caller of when the peer with
// the given public key has been disconnected. The notification is signaled by
// closing the channel returned.
func (s *server) NotifyWhenOffline(peerPubKey [33]byte) <-chan struct{} {
s.mu.Lock()
defer s.mu.Unlock()
c := make(chan struct{})
// If the peer is already offline, we can immediately trigger the
// notification.
peerPubKeyStr := string(peerPubKey[:])
if _, ok := s.peersByPub[peerPubKeyStr]; !ok {
srvrLog.Debugf("Notifying that peer %x is offline", peerPubKey)
close(c)
return c
}
// Otherwise, the peer is online, so we'll keep track of the channel to
// trigger the notification once the server detects the peer
// disconnects.
s.peerDisconnectedListeners[peerPubKeyStr] = append(
s.peerDisconnectedListeners[peerPubKeyStr], c,
)
return c
}
// FindPeer will return the peer that corresponds to the passed in public key.
// This function is used by the funding manager, allowing it to update the
// daemon's local representation of the remote peer.
//
// NOTE: This function is safe for concurrent access.
func (s *server) FindPeer(peerKey *btcec.PublicKey) (*peer.Brontide, error) {
s.mu.RLock()
defer s.mu.RUnlock()
pubStr := string(peerKey.SerializeCompressed())
return s.findPeerByPubStr(pubStr)
}
// FindPeerByPubStr will return the peer that corresponds to the passed peerID,
// which should be a string representation of the peer's serialized, compressed
// public key.
//
// NOTE: This function is safe for concurrent access.
func (s *server) FindPeerByPubStr(pubStr string) (*peer.Brontide, error) {
s.mu.RLock()
defer s.mu.RUnlock()
return s.findPeerByPubStr(pubStr)
}
// findPeerByPubStr is an internal method that retrieves the specified peer from
// the server's internal state using.
func (s *server) findPeerByPubStr(pubStr string) (*peer.Brontide, error) {
peer, ok := s.peersByPub[pubStr]
if !ok {
return nil, ErrPeerNotConnected
}
return peer, nil
}
// nextPeerBackoff computes the next backoff duration for a peer's pubkey using
// exponential backoff. If no previous backoff was known, the default is
// returned.
func (s *server) nextPeerBackoff(pubStr string,
startTime time.Time) time.Duration {
// Now, determine the appropriate backoff to use for the retry.
backoff, ok := s.persistentPeersBackoff[pubStr]
if !ok {
// If an existing backoff was unknown, use the default.
return s.cfg.MinBackoff
}
// If the peer failed to start properly, we'll just use the previous
// backoff to compute the subsequent randomized exponential backoff
// duration. This will roughly double on average.
if startTime.IsZero() {
return computeNextBackoff(backoff, s.cfg.MaxBackoff)
}
// The peer succeeded in starting. If the connection didn't last long
// enough to be considered stable, we'll continue to back off retries
// with this peer.
connDuration := time.Since(startTime)
if connDuration < defaultStableConnDuration {
return computeNextBackoff(backoff, s.cfg.MaxBackoff)
}
// The peer succeed in starting and this was stable peer, so we'll
// reduce the timeout duration by the length of the connection after
// applying randomized exponential backoff. We'll only apply this in the
// case that:
// reb(curBackoff) - connDuration > cfg.MinBackoff
relaxedBackoff := computeNextBackoff(backoff, s.cfg.MaxBackoff) - connDuration
if relaxedBackoff > s.cfg.MinBackoff {
return relaxedBackoff
}
// Lastly, if reb(currBackoff) - connDuration <= cfg.MinBackoff, meaning
// the stable connection lasted much longer than our previous backoff.
// To reward such good behavior, we'll reconnect after the default
// timeout.
return s.cfg.MinBackoff
}
// shouldDropConnection determines if our local connection to a remote peer
// should be dropped in the case of concurrent connection establishment. In
// order to deterministically decide which connection should be dropped, we'll
// utilize the ordering of the local and remote public key. If we didn't use
// such a tie breaker, then we risk _both_ connections erroneously being
// dropped.
func shouldDropLocalConnection(local, remote *btcec.PublicKey) bool {
localPubBytes := local.SerializeCompressed()
remotePubPbytes := remote.SerializeCompressed()
// The connection that comes from the node with a "smaller" pubkey
// should be kept. Therefore, if our pubkey is "greater" than theirs, we
// should drop our established connection.
return bytes.Compare(localPubBytes, remotePubPbytes) > 0
}
// InboundPeerConnected initializes a new peer in response to a new inbound
// connection.
//
// NOTE: This function is safe for concurrent access.
func (s *server) InboundPeerConnected(conn net.Conn) {
// Exit early if we have already been instructed to shutdown, this
// prevents any delayed callbacks from accidentally registering peers.
if s.Stopped() {
return
}
nodePub := conn.(*brontide.Conn).RemotePub()
pubStr := string(nodePub.SerializeCompressed())
s.mu.Lock()
defer s.mu.Unlock()
// If we already have an outbound connection to this peer, then ignore
// this new connection.
if p, ok := s.outboundPeers[pubStr]; ok {
srvrLog.Debugf("Already have outbound connection for %v, "+
"ignoring inbound connection from local=%v, remote=%v",
p, conn.LocalAddr(), conn.RemoteAddr())
conn.Close()
return
}
// If we already have a valid connection that is scheduled to take
// precedence once the prior peer has finished disconnecting, we'll
// ignore this connection.
if p, ok := s.scheduledPeerConnection[pubStr]; ok {
srvrLog.Debugf("Ignoring connection from %v, peer %v already "+
"scheduled", conn.RemoteAddr(), p)
conn.Close()
return
}
srvrLog.Infof("New inbound connection from %v", conn.RemoteAddr())
// Check to see if we already have a connection with this peer. If so,
// we may need to drop our existing connection. This prevents us from
// having duplicate connections to the same peer. We forgo adding a
// default case as we expect these to be the only error values returned
// from findPeerByPubStr.
connectedPeer, err := s.findPeerByPubStr(pubStr)
switch err {
case ErrPeerNotConnected:
// We were unable to locate an existing connection with the
// target peer, proceed to connect.
s.cancelConnReqs(pubStr, nil)
s.peerConnected(conn, nil, true)
case nil:
// We already have a connection with the incoming peer. If the
// connection we've already established should be kept and is
// not of the same type of the new connection (inbound), then
// we'll close out the new connection s.t there's only a single
// connection between us.
localPub := s.identityECDH.PubKey()
if !connectedPeer.Inbound() &&
!shouldDropLocalConnection(localPub, nodePub) {
srvrLog.Warnf("Received inbound connection from "+
"peer %v, but already have outbound "+
"connection, dropping conn", connectedPeer)
conn.Close()
return
}
// Otherwise, if we should drop the connection, then we'll
// disconnect our already connected peer.
srvrLog.Debugf("Disconnecting stale connection to %v",
connectedPeer)
s.cancelConnReqs(pubStr, nil)
// Remove the current peer from the server's internal state and
// signal that the peer termination watcher does not need to
// execute for this peer.
s.removePeer(connectedPeer)
s.ignorePeerTermination[connectedPeer] = struct{}{}
s.scheduledPeerConnection[pubStr] = func() {
s.peerConnected(conn, nil, true)
}
}
}
// OutboundPeerConnected initializes a new peer in response to a new outbound
// connection.
// NOTE: This function is safe for concurrent access.
func (s *server) OutboundPeerConnected(connReq *connmgr.ConnReq, conn net.Conn) {
// Exit early if we have already been instructed to shutdown, this
// prevents any delayed callbacks from accidentally registering peers.
if s.Stopped() {
return
}
nodePub := conn.(*brontide.Conn).RemotePub()
pubStr := string(nodePub.SerializeCompressed())
s.mu.Lock()
defer s.mu.Unlock()
// If we already have an inbound connection to this peer, then ignore
// this new connection.
if p, ok := s.inboundPeers[pubStr]; ok {
srvrLog.Debugf("Already have inbound connection for %v, "+
"ignoring outbound connection from local=%v, remote=%v",
p, conn.LocalAddr(), conn.RemoteAddr())
if connReq != nil {
s.connMgr.Remove(connReq.ID())
}
conn.Close()
return
}
if _, ok := s.persistentConnReqs[pubStr]; !ok && connReq != nil {
srvrLog.Debugf("Ignoring canceled outbound connection")
s.connMgr.Remove(connReq.ID())
conn.Close()
return
}
// If we already have a valid connection that is scheduled to take
// precedence once the prior peer has finished disconnecting, we'll
// ignore this connection.
if _, ok := s.scheduledPeerConnection[pubStr]; ok {
srvrLog.Debugf("Ignoring connection, peer already scheduled")
if connReq != nil {
s.connMgr.Remove(connReq.ID())
}
conn.Close()
return
}
srvrLog.Infof("Established connection to: %x@%v", pubStr,
conn.RemoteAddr())
if connReq != nil {
// A successful connection was returned by the connmgr.
// Immediately cancel all pending requests, excluding the
// outbound connection we just established.
ignore := connReq.ID()
s.cancelConnReqs(pubStr, &ignore)
} else {
// This was a successful connection made by some other
// subsystem. Remove all requests being managed by the connmgr.
s.cancelConnReqs(pubStr, nil)
}
// If we already have a connection with this peer, decide whether or not
// we need to drop the stale connection. We forgo adding a default case
// as we expect these to be the only error values returned from
// findPeerByPubStr.
connectedPeer, err := s.findPeerByPubStr(pubStr)
switch err {
case ErrPeerNotConnected:
// We were unable to locate an existing connection with the
// target peer, proceed to connect.
s.peerConnected(conn, connReq, false)
case nil:
// We already have a connection with the incoming peer. If the
// connection we've already established should be kept and is
// not of the same type of the new connection (outbound), then
// we'll close out the new connection s.t there's only a single
// connection between us.
localPub := s.identityECDH.PubKey()
if connectedPeer.Inbound() &&
shouldDropLocalConnection(localPub, nodePub) {
srvrLog.Warnf("Established outbound connection to "+
"peer %v, but already have inbound "+
"connection, dropping conn", connectedPeer)
if connReq != nil {
s.connMgr.Remove(connReq.ID())
}
conn.Close()
return
}
// Otherwise, _their_ connection should be dropped. So we'll
// disconnect the peer and send the now obsolete peer to the
// server for garbage collection.
srvrLog.Debugf("Disconnecting stale connection to %v",
connectedPeer)
// Remove the current peer from the server's internal state and
// signal that the peer termination watcher does not need to
// execute for this peer.
s.removePeer(connectedPeer)
s.ignorePeerTermination[connectedPeer] = struct{}{}
s.scheduledPeerConnection[pubStr] = func() {
s.peerConnected(conn, connReq, false)
}
}
}
// UnassignedConnID is the default connection ID that a request can have before
// it actually is submitted to the connmgr.
// TODO(conner): move into connmgr package, or better, add connmgr method for
// generating atomic IDs
const UnassignedConnID uint64 = 0
// cancelConnReqs stops all persistent connection requests for a given pubkey.
// Any attempts initiated by the peerTerminationWatcher are canceled first.
// Afterwards, each connection request removed from the connmgr. The caller can
// optionally specify a connection ID to ignore, which prevents us from
// canceling a successful request. All persistent connreqs for the provided
// pubkey are discarded after the operationjw.
func (s *server) cancelConnReqs(pubStr string, skip *uint64) {
// First, cancel any lingering persistent retry attempts, which will
// prevent retries for any with backoffs that are still maturing.
if cancelChan, ok := s.persistentRetryCancels[pubStr]; ok {
close(cancelChan)
delete(s.persistentRetryCancels, pubStr)
}
// Next, check to see if we have any outstanding persistent connection
// requests to this peer. If so, then we'll remove all of these
// connection requests, and also delete the entry from the map.
connReqs, ok := s.persistentConnReqs[pubStr]
if !ok {
return
}
for _, connReq := range connReqs {
srvrLog.Tracef("Canceling %s:", connReqs)
// Atomically capture the current request identifier.
connID := connReq.ID()
// Skip any zero IDs, this indicates the request has not
// yet been schedule.
if connID == UnassignedConnID {
continue
}
// Skip a particular connection ID if instructed.
if skip != nil && connID == *skip {
continue
}
s.connMgr.Remove(connID)
}
delete(s.persistentConnReqs, pubStr)
}
// handleCustomMessage dispatches an incoming custom peers message to
// subscribers.
func (s *server) handleCustomMessage(peer [33]byte, msg *lnwire.Custom) error {
srvrLog.Debugf("Custom message received: peer=%x, type=%d",
peer, msg.Type)
return s.customMessageServer.SendUpdate(&CustomMessage{
Peer: peer,
Msg: msg,
})
}
// SubscribeCustomMessages subscribes to a stream of incoming custom peer
// messages.
func (s *server) SubscribeCustomMessages() (*subscribe.Client, error) {
return s.customMessageServer.Subscribe()
}
// peerConnected is a function that handles initialization a newly connected
// peer by adding it to the server's global list of all active peers, and
// starting all the goroutines the peer needs to function properly. The inbound
// boolean should be true if the peer initiated the connection to us.
func (s *server) peerConnected(conn net.Conn, connReq *connmgr.ConnReq,
inbound bool) {
brontideConn := conn.(*brontide.Conn)
addr := conn.RemoteAddr()
pubKey := brontideConn.RemotePub()
srvrLog.Infof("Finalizing connection to %x@%s, inbound=%v",
pubKey.SerializeCompressed(), addr, inbound)
peerAddr := &lnwire.NetAddress{
IdentityKey: pubKey,
Address: addr,
ChainNet: s.cfg.ActiveNetParams.Net,
}
// With the brontide connection established, we'll now craft the feature
// vectors to advertise to the remote node.
initFeatures := s.featureMgr.Get(feature.SetInit)
legacyFeatures := s.featureMgr.Get(feature.SetLegacyGlobal)
// Lookup past error caches for the peer in the server. If no buffer is
// found, create a fresh buffer.
pkStr := string(peerAddr.IdentityKey.SerializeCompressed())
errBuffer, ok := s.peerErrors[pkStr]
if !ok {
var err error
errBuffer, err = queue.NewCircularBuffer(peer.ErrorBufferSize)
if err != nil {
srvrLog.Errorf("unable to create peer %v", err)
return
}
}
// Now that we've established a connection, create a peer, and it to the
// set of currently active peers. Configure the peer with the incoming
// and outgoing broadcast deltas to prevent htlcs from being accepted or
// offered that would trigger channel closure. In case of outgoing
// htlcs, an extra block is added to prevent the channel from being
// closed when the htlc is outstanding and a new block comes in.
pCfg := peer.Config{
Conn: brontideConn,
ConnReq: connReq,
Addr: peerAddr,
Inbound: inbound,
Features: initFeatures,
LegacyFeatures: legacyFeatures,
OutgoingCltvRejectDelta: lncfg.DefaultOutgoingCltvRejectDelta,
ChanActiveTimeout: s.cfg.ChanEnableTimeout,
ErrorBuffer: errBuffer,
WritePool: s.writePool,
ReadPool: s.readPool,
Switch: s.htlcSwitch,
InterceptSwitch: s.interceptableSwitch,
ChannelDB: s.chanStateDB,
ChannelGraph: s.graphDB,
ChainArb: s.chainArb,
AuthGossiper: s.authGossiper,
ChanStatusMgr: s.chanStatusMgr,
ChainIO: s.cc.ChainIO,
FeeEstimator: s.cc.FeeEstimator,
Signer: s.cc.Wallet.Cfg.Signer,
SigPool: s.sigPool,
Wallet: s.cc.Wallet,
ChainNotifier: s.cc.ChainNotifier,
RoutingPolicy: s.cc.RoutingPolicy,
Sphinx: s.sphinx,
WitnessBeacon: s.witnessBeacon,
Invoices: s.invoices,
ChannelNotifier: s.channelNotifier,
HtlcNotifier: s.htlcNotifier,
TowerClient: s.towerClient,
AnchorTowerClient: s.anchorTowerClient,
DisconnectPeer: s.DisconnectPeer,
GenNodeAnnouncement: func(...netann.NodeAnnModifier) (
lnwire.NodeAnnouncement, error) {
return s.genNodeAnnouncement(nil)
},
PongBuf: s.pongBuf,
PrunePersistentPeerConnection: s.prunePersistentPeerConnection,
FetchLastChanUpdate: s.fetchLastChanUpdate(),
FundingManager: s.fundingMgr,
Hodl: s.cfg.Hodl,
UnsafeReplay: s.cfg.UnsafeReplay,
MaxOutgoingCltvExpiry: s.cfg.MaxOutgoingCltvExpiry,
MaxChannelFeeAllocation: s.cfg.MaxChannelFeeAllocation,
CoopCloseTargetConfs: s.cfg.CoopCloseTargetConfs,
MaxAnchorsCommitFeeRate: chainfee.SatPerKVByte(
s.cfg.MaxCommitFeeRateAnchors * 1000).FeePerKWeight(),
ChannelCommitInterval: s.cfg.ChannelCommitInterval,
PendingCommitInterval: s.cfg.PendingCommitInterval,
ChannelCommitBatchSize: s.cfg.ChannelCommitBatchSize,
HandleCustomMessage: s.handleCustomMessage,
GetAliases: s.aliasMgr.GetAliases,
RequestAlias: s.aliasMgr.RequestAlias,
AddLocalAlias: s.aliasMgr.AddLocalAlias,
Quit: s.quit,
}
copy(pCfg.PubKeyBytes[:], peerAddr.IdentityKey.SerializeCompressed())
copy(pCfg.ServerPubKey[:], s.identityECDH.PubKey().SerializeCompressed())
p := peer.NewBrontide(pCfg)
// TODO(roasbeef): update IP address for link-node
// * also mark last-seen, do it one single transaction?
s.addPeer(p)
// Once we have successfully added the peer to the server, we can
// delete the previous error buffer from the server's map of error
// buffers.
delete(s.peerErrors, pkStr)
// Dispatch a goroutine to asynchronously start the peer. This process
// includes sending and receiving Init messages, which would be a DOS
// vector if we held the server's mutex throughout the procedure.
s.wg.Add(1)
go s.peerInitializer(p)
}
// addPeer adds the passed peer to the server's global state of all active
// peers.
func (s *server) addPeer(p *peer.Brontide) {
if p == nil {
return
}
// Ignore new peers if we're shutting down.
if s.Stopped() {
p.Disconnect(ErrServerShuttingDown)
return
}
// Track the new peer in our indexes so we can quickly look it up either
// according to its public key, or its peer ID.
// TODO(roasbeef): pipe all requests through to the
// queryHandler/peerManager
pubSer := p.IdentityKey().SerializeCompressed()
pubStr := string(pubSer)
s.peersByPub[pubStr] = p
if p.Inbound() {
s.inboundPeers[pubStr] = p
} else {
s.outboundPeers[pubStr] = p
}
// Inform the peer notifier of a peer online event so that it can be reported
// to clients listening for peer events.
var pubKey [33]byte
copy(pubKey[:], pubSer)
s.peerNotifier.NotifyPeerOnline(pubKey)
}
// peerInitializer asynchronously starts a newly connected peer after it has
// been added to the server's peer map. This method sets up a
// peerTerminationWatcher for the given peer, and ensures that it executes even
// if the peer failed to start. In the event of a successful connection, this
// method reads the negotiated, local feature-bits and spawns the appropriate
// graph synchronization method. Any registered clients of NotifyWhenOnline will
// be signaled of the new peer once the method returns.
//
// NOTE: This MUST be launched as a goroutine.
func (s *server) peerInitializer(p *peer.Brontide) {
defer s.wg.Done()
// Avoid initializing peers while the server is exiting.
if s.Stopped() {
return
}
// Create a channel that will be used to signal a successful start of
// the link. This prevents the peer termination watcher from beginning
// its duty too early.
ready := make(chan struct{})
// Before starting the peer, launch a goroutine to watch for the
// unexpected termination of this peer, which will ensure all resources
// are properly cleaned up, and re-establish persistent connections when
// necessary. The peer termination watcher will be short circuited if
// the peer is ever added to the ignorePeerTermination map, indicating
// that the server has already handled the removal of this peer.
s.wg.Add(1)
go s.peerTerminationWatcher(p, ready)
// Start the peer! If an error occurs, we Disconnect the peer, which
// will unblock the peerTerminationWatcher.
if err := p.Start(); err != nil {
p.Disconnect(fmt.Errorf("unable to start peer: %v", err))
return
}
// Otherwise, signal to the peerTerminationWatcher that the peer startup
// was successful, and to begin watching the peer's wait group.
close(ready)
pubStr := string(p.IdentityKey().SerializeCompressed())
s.mu.Lock()
defer s.mu.Unlock()
// Check if there are listeners waiting for this peer to come online.
srvrLog.Debugf("Notifying that peer %v is online", p)
for _, peerChan := range s.peerConnectedListeners[pubStr] {
select {
case peerChan <- p:
case <-s.quit:
return
}
}
delete(s.peerConnectedListeners, pubStr)
}
// peerTerminationWatcher waits until a peer has been disconnected unexpectedly,
// and then cleans up all resources allocated to the peer, notifies relevant
// sub-systems of its demise, and finally handles re-connecting to the peer if
// it's persistent. If the server intentionally disconnects a peer, it should
// have a corresponding entry in the ignorePeerTermination map which will cause
// the cleanup routine to exit early. The passed `ready` chan is used to
// synchronize when WaitForDisconnect should begin watching on the peer's
// waitgroup. The ready chan should only be signaled if the peer starts
// successfully, otherwise the peer should be disconnected instead.
//
// NOTE: This MUST be launched as a goroutine.
func (s *server) peerTerminationWatcher(p *peer.Brontide, ready chan struct{}) {
defer s.wg.Done()
p.WaitForDisconnect(ready)
srvrLog.Debugf("Peer %v has been disconnected", p)
// If the server is exiting then we can bail out early ourselves as all
// the other sub-systems will already be shutting down.
if s.Stopped() {
srvrLog.Debugf("Server quitting, exit early for peer %v", p)
return
}
// Next, we'll cancel all pending funding reservations with this node.
// If we tried to initiate any funding flows that haven't yet finished,
// then we need to unlock those committed outputs so they're still
// available for use.
s.fundingMgr.CancelPeerReservations(p.PubKey())
pubKey := p.IdentityKey()
// We'll also inform the gossiper that this peer is no longer active,
// so we don't need to maintain sync state for it any longer.
s.authGossiper.PruneSyncState(p.PubKey())
// Tell the switch to remove all links associated with this peer.
// Passing nil as the target link indicates that all links associated
// with this interface should be closed.
//
// TODO(roasbeef): instead add a PurgeInterfaceLinks function?
links, err := s.htlcSwitch.GetLinksByInterface(p.PubKey())
if err != nil && err != htlcswitch.ErrNoLinksFound {
srvrLog.Errorf("Unable to get channel links for %v: %v", p, err)
}
for _, link := range links {
s.htlcSwitch.RemoveLink(link.ChanID())
}
s.mu.Lock()
defer s.mu.Unlock()
// If there were any notification requests for when this peer
// disconnected, we can trigger them now.
srvrLog.Debugf("Notifying that peer %v is offline", p)
pubStr := string(pubKey.SerializeCompressed())
for _, offlineChan := range s.peerDisconnectedListeners[pubStr] {
close(offlineChan)
}
delete(s.peerDisconnectedListeners, pubStr)
// If the server has already removed this peer, we can short circuit the
// peer termination watcher and skip cleanup.
if _, ok := s.ignorePeerTermination[p]; ok {
delete(s.ignorePeerTermination, p)
pubKey := p.PubKey()
pubStr := string(pubKey[:])
// If a connection callback is present, we'll go ahead and
// execute it now that previous peer has fully disconnected. If
// the callback is not present, this likely implies the peer was
// purposefully disconnected via RPC, and that no reconnect
// should be attempted.
connCallback, ok := s.scheduledPeerConnection[pubStr]
if ok {
delete(s.scheduledPeerConnection, pubStr)
connCallback()
}
return
}
// First, cleanup any remaining state the server has regarding the peer
// in question.
s.removePeer(p)
// Next, check to see if this is a persistent peer or not.
if _, ok := s.persistentPeers[pubStr]; !ok {
return
}
// Get the last address that we used to connect to the peer.
addrs := []net.Addr{
p.NetAddress().Address,
}
// We'll ensure that we locate all the peers advertised addresses for
// reconnection purposes.
advertisedAddrs, err := s.fetchNodeAdvertisedAddrs(pubKey)
switch {
// We found advertised addresses, so use them.
case err == nil:
addrs = advertisedAddrs
// The peer doesn't have an advertised address.
case err == errNoAdvertisedAddr:
// If it is an outbound peer then we fall back to the existing
// peer address.
if !p.Inbound() {
break
}
// Fall back to the existing peer address if
// we're not accepting connections over Tor.
if s.torController == nil {
break
}
// If we are, the peer's address won't be known
// to us (we'll see a private address, which is
// the address used by our onion service to dial
// to lnd), so we don't have enough information
// to attempt a reconnect.
srvrLog.Debugf("Ignoring reconnection attempt "+
"to inbound peer %v without "+
"advertised address", p)
return
// We came across an error retrieving an advertised
// address, log it, and fall back to the existing peer
// address.
default:
srvrLog.Errorf("Unable to retrieve advertised "+
"address for node %x: %v", p.PubKey(),
err)
}
// Make an easy lookup map so that we can check if an address
// is already in the address list that we have stored for this peer.
existingAddrs := make(map[string]bool)
for _, addr := range s.persistentPeerAddrs[pubStr] {
existingAddrs[addr.String()] = true
}
// Add any missing addresses for this peer to persistentPeerAddr.
for _, addr := range addrs {
if existingAddrs[addr.String()] {
continue
}
s.persistentPeerAddrs[pubStr] = append(
s.persistentPeerAddrs[pubStr],
&lnwire.NetAddress{
IdentityKey: p.IdentityKey(),
Address: addr,
ChainNet: p.NetAddress().ChainNet,
},
)
}
// Record the computed backoff in the backoff map.
backoff := s.nextPeerBackoff(pubStr, p.StartTime())
s.persistentPeersBackoff[pubStr] = backoff
// Initialize a retry canceller for this peer if one does not
// exist.
cancelChan, ok := s.persistentRetryCancels[pubStr]
if !ok {
cancelChan = make(chan struct{})
s.persistentRetryCancels[pubStr] = cancelChan
}
// We choose not to wait group this go routine since the Connect
// call can stall for arbitrarily long if we shutdown while an
// outbound connection attempt is being made.
go func() {
srvrLog.Debugf("Scheduling connection re-establishment to "+
"persistent peer %x in %s",
p.IdentityKey().SerializeCompressed(), backoff)
select {
case <-time.After(backoff):
case <-cancelChan:
return
case <-s.quit:
return
}
srvrLog.Debugf("Attempting to re-establish persistent "+
"connection to peer %x",
p.IdentityKey().SerializeCompressed())
s.connectToPersistentPeer(pubStr)
}()
}
// connectToPersistentPeer uses all the stored addresses for a peer to attempt
// to connect to the peer. It creates connection requests if there are
// currently none for a given address and it removes old connection requests
// if the associated address is no longer in the latest address list for the
// peer.
func (s *server) connectToPersistentPeer(pubKeyStr string) {
s.mu.Lock()
defer s.mu.Unlock()
// Create an easy lookup map of the addresses we have stored for the
// peer. We will remove entries from this map if we have existing
// connection requests for the associated address and then any leftover
// entries will indicate which addresses we should create new
// connection requests for.
addrMap := make(map[string]*lnwire.NetAddress)
for _, addr := range s.persistentPeerAddrs[pubKeyStr] {
addrMap[addr.String()] = addr
}
// Go through each of the existing connection requests and
// check if they correspond to the latest set of addresses. If
// there is a connection requests that does not use one of the latest
// advertised addresses then remove that connection request.
var updatedConnReqs []*connmgr.ConnReq
for _, connReq := range s.persistentConnReqs[pubKeyStr] {
lnAddr := connReq.Addr.(*lnwire.NetAddress).Address.String()
switch _, ok := addrMap[lnAddr]; ok {
// If the existing connection request is using one of the
// latest advertised addresses for the peer then we add it to
// updatedConnReqs and remove the associated address from
// addrMap so that we don't recreate this connReq later on.
case true:
updatedConnReqs = append(
updatedConnReqs, connReq,
)
delete(addrMap, lnAddr)
// If the existing connection request is using an address that
// is not one of the latest advertised addresses for the peer
// then we remove the connecting request from the connection
// manager.
case false:
srvrLog.Info(
"Removing conn req:", connReq.Addr.String(),
)
s.connMgr.Remove(connReq.ID())
}
}
s.persistentConnReqs[pubKeyStr] = updatedConnReqs
cancelChan, ok := s.persistentRetryCancels[pubKeyStr]
if !ok {
cancelChan = make(chan struct{})
s.persistentRetryCancels[pubKeyStr] = cancelChan
}
// Any addresses left in addrMap are new ones that we have not made
// connection requests for. So create new connection requests for those.
// If there is more than one address in the address map, stagger the
// creation of the connection requests for those.
go func() {
ticker := time.NewTicker(multiAddrConnectionStagger)
defer ticker.Stop()
for _, addr := range addrMap {
// Send the persistent connection request to the
// connection manager, saving the request itself so we
// can cancel/restart the process as needed.
connReq := &connmgr.ConnReq{
Addr: addr,
Permanent: true,
}
s.mu.Lock()
s.persistentConnReqs[pubKeyStr] = append(
s.persistentConnReqs[pubKeyStr], connReq,
)
s.mu.Unlock()
srvrLog.Debugf("Attempting persistent connection to "+
"channel peer %v", addr)
go s.connMgr.Connect(connReq)
select {
case <-s.quit:
return
case <-cancelChan:
return
case <-ticker.C:
}
}
}()
}
// removePeer removes the passed peer from the server's state of all active
// peers.
func (s *server) removePeer(p *peer.Brontide) {
if p == nil {
return
}
srvrLog.Debugf("removing peer %v", p)
// As the peer is now finished, ensure that the TCP connection is
// closed and all of its related goroutines have exited.
p.Disconnect(fmt.Errorf("server: disconnecting peer %v", p))
// If this peer had an active persistent connection request, remove it.
if p.ConnReq() != nil {
s.connMgr.Remove(p.ConnReq().ID())
}
// Ignore deleting peers if we're shutting down.
if s.Stopped() {
return
}
pKey := p.PubKey()
pubSer := pKey[:]
pubStr := string(pubSer)
delete(s.peersByPub, pubStr)
if p.Inbound() {
delete(s.inboundPeers, pubStr)
} else {
delete(s.outboundPeers, pubStr)
}
// Copy the peer's error buffer across to the server if it has any items
// in it so that we can restore peer errors across connections.
if p.ErrorBuffer().Total() > 0 {
s.peerErrors[pubStr] = p.ErrorBuffer()
}
// Inform the peer notifier of a peer offline event so that it can be
// reported to clients listening for peer events.
var pubKey [33]byte
copy(pubKey[:], pubSer)
s.peerNotifier.NotifyPeerOffline(pubKey)
}
// ConnectToPeer requests that the server connect to a Lightning Network peer
// at the specified address. This function will *block* until either a
// connection is established, or the initial handshake process fails.
//
// NOTE: This function is safe for concurrent access.
func (s *server) ConnectToPeer(addr *lnwire.NetAddress,
perm bool, timeout time.Duration) error {
targetPub := string(addr.IdentityKey.SerializeCompressed())
// Acquire mutex, but use explicit unlocking instead of defer for
// better granularity. In certain conditions, this method requires
// making an outbound connection to a remote peer, which requires the
// lock to be released, and subsequently reacquired.
s.mu.Lock()
// Ensure we're not already connected to this peer.
peer, err := s.findPeerByPubStr(targetPub)
if err == nil {
s.mu.Unlock()
return &errPeerAlreadyConnected{peer: peer}
}
// Peer was not found, continue to pursue connection with peer.
// If there's already a pending connection request for this pubkey,
// then we ignore this request to ensure we don't create a redundant
// connection.
if reqs, ok := s.persistentConnReqs[targetPub]; ok {
srvrLog.Warnf("Already have %d persistent connection "+
"requests for %v, connecting anyway.", len(reqs), addr)
}
// If there's not already a pending or active connection to this node,
// then instruct the connection manager to attempt to establish a
// persistent connection to the peer.
srvrLog.Debugf("Connecting to %v", addr)
if perm {
connReq := &connmgr.ConnReq{
Addr: addr,
Permanent: true,
}
// Since the user requested a permanent connection, we'll set
// the entry to true which will tell the server to continue
// reconnecting even if the number of channels with this peer is
// zero.
s.persistentPeers[targetPub] = true
if _, ok := s.persistentPeersBackoff[targetPub]; !ok {
s.persistentPeersBackoff[targetPub] = s.cfg.MinBackoff
}
s.persistentConnReqs[targetPub] = append(
s.persistentConnReqs[targetPub], connReq,
)
s.mu.Unlock()
go s.connMgr.Connect(connReq)
return nil
}
s.mu.Unlock()
// If we're not making a persistent connection, then we'll attempt to
// connect to the target peer. If the we can't make the connection, or
// the crypto negotiation breaks down, then return an error to the
// caller.
errChan := make(chan error, 1)
s.connectToPeer(addr, errChan, timeout)
select {
case err := <-errChan:
return err
case <-s.quit:
return ErrServerShuttingDown
}
}
// connectToPeer establishes a connection to a remote peer. errChan is used to
// notify the caller if the connection attempt has failed. Otherwise, it will be
// closed.
func (s *server) connectToPeer(addr *lnwire.NetAddress,
errChan chan<- error, timeout time.Duration) {
conn, err := brontide.Dial(
s.identityECDH, addr, timeout, s.cfg.net.Dial,
)
if err != nil {
srvrLog.Errorf("Unable to connect to %v: %v", addr, err)
select {
case errChan <- err:
case <-s.quit:
}
return
}
close(errChan)
srvrLog.Tracef("Brontide dialer made local=%v, remote=%v",
conn.LocalAddr(), conn.RemoteAddr())
s.OutboundPeerConnected(nil, conn)
}
// DisconnectPeer sends the request to server to close the connection with peer
// identified by public key.
//
// NOTE: This function is safe for concurrent access.
func (s *server) DisconnectPeer(pubKey *btcec.PublicKey) error {
pubBytes := pubKey.SerializeCompressed()
pubStr := string(pubBytes)
s.mu.Lock()
defer s.mu.Unlock()
// Check that were actually connected to this peer. If not, then we'll
// exit in an error as we can't disconnect from a peer that we're not
// currently connected to.
peer, err := s.findPeerByPubStr(pubStr)
if err == ErrPeerNotConnected {
return fmt.Errorf("peer %x is not connected", pubBytes)
}
srvrLog.Infof("Disconnecting from %v", peer)
s.cancelConnReqs(pubStr, nil)
// If this peer was formerly a persistent connection, then we'll remove
// them from this map so we don't attempt to re-connect after we
// disconnect.
delete(s.persistentPeers, pubStr)
delete(s.persistentPeersBackoff, pubStr)
// Remove the peer by calling Disconnect. Previously this was done with
// removePeer, which bypassed the peerTerminationWatcher.
peer.Disconnect(fmt.Errorf("server: DisconnectPeer called"))
return nil
}
// OpenChannel sends a request to the server to open a channel to the specified
// peer identified by nodeKey with the passed channel funding parameters.
//
// NOTE: This function is safe for concurrent access.
func (s *server) OpenChannel(
req *funding.InitFundingMsg) (chan *lnrpc.OpenStatusUpdate, chan error) {
// The updateChan will have a buffer of 2, since we expect a ChanPending
// + a ChanOpen update, and we want to make sure the funding process is
// not blocked if the caller is not reading the updates.
req.Updates = make(chan *lnrpc.OpenStatusUpdate, 2)
req.Err = make(chan error, 1)
// First attempt to locate the target peer to open a channel with, if
// we're unable to locate the peer then this request will fail.
pubKeyBytes := req.TargetPubkey.SerializeCompressed()
s.mu.RLock()
peer, ok := s.peersByPub[string(pubKeyBytes)]
if !ok {
s.mu.RUnlock()
req.Err <- fmt.Errorf("peer %x is not online", pubKeyBytes)
return req.Updates, req.Err
}
req.Peer = peer
s.mu.RUnlock()
// We'll wait until the peer is active before beginning the channel
// opening process.
select {
case <-peer.ActiveSignal():
case <-peer.QuitSignal():
req.Err <- fmt.Errorf("peer %x disconnected", pubKeyBytes)
return req.Updates, req.Err
case <-s.quit:
req.Err <- ErrServerShuttingDown
return req.Updates, req.Err
}
// If the fee rate wasn't specified, then we'll use a default
// confirmation target.
if req.FundingFeePerKw == 0 {
estimator := s.cc.FeeEstimator
feeRate, err := estimator.EstimateFeePerKW(6)
if err != nil {
req.Err <- err
return req.Updates, req.Err
}
req.FundingFeePerKw = feeRate
}
// Spawn a goroutine to send the funding workflow request to the funding
// manager. This allows the server to continue handling queries instead
// of blocking on this request which is exported as a synchronous
// request to the outside world.
go s.fundingMgr.InitFundingWorkflow(req)
return req.Updates, req.Err
}
// Peers returns a slice of all active peers.
//
// NOTE: This function is safe for concurrent access.
func (s *server) Peers() []*peer.Brontide {
s.mu.RLock()
defer s.mu.RUnlock()
peers := make([]*peer.Brontide, 0, len(s.peersByPub))
for _, peer := range s.peersByPub {
peers = append(peers, peer)
}
return peers
}
// computeNextBackoff uses a truncated exponential backoff to compute the next
// backoff using the value of the exiting backoff. The returned duration is
// randomized in either direction by 1/20 to prevent tight loops from
// stabilizing.
func computeNextBackoff(currBackoff, maxBackoff time.Duration) time.Duration {
// Double the current backoff, truncating if it exceeds our maximum.
nextBackoff := 2 * currBackoff
if nextBackoff > maxBackoff {
nextBackoff = maxBackoff
}
// Using 1/10 of our duration as a margin, compute a random offset to
// avoid the nodes entering connection cycles.
margin := nextBackoff / 10
var wiggle big.Int
wiggle.SetUint64(uint64(margin))
if _, err := rand.Int(rand.Reader, &wiggle); err != nil {
// Randomizing is not mission critical, so we'll just return the
// current backoff.
return nextBackoff
}
// Otherwise add in our wiggle, but subtract out half of the margin so
// that the backoff can tweaked by 1/20 in either direction.
return nextBackoff + (time.Duration(wiggle.Uint64()) - margin/2)
}
// errNoAdvertisedAddr is an error returned when we attempt to retrieve the
// advertised address of a node, but they don't have one.
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) {
vertex, err := route.NewVertexFromBytes(pub.SerializeCompressed())
if err != nil {
return nil, err
}
node, err := s.graphDB.FetchLightningNode(vertex)
if err != nil {
return nil, err
}
if len(node.Addresses) == 0 {
return nil, errNoAdvertisedAddr
}
return node.Addresses, nil
}
// fetchLastChanUpdate returns a function which is able to retrieve our latest
// channel update for a target channel.
func (s *server) fetchLastChanUpdate() func(lnwire.ShortChannelID) (
*lnwire.ChannelUpdate, error) {
ourPubKey := s.identityECDH.PubKey().SerializeCompressed()
return func(cid lnwire.ShortChannelID) (*lnwire.ChannelUpdate, error) {
info, edge1, edge2, err := s.chanRouter.GetChannelByID(cid)
if err != nil {
return nil, err
}
return netann.ExtractChannelUpdate(
ourPubKey[:], info, edge1, edge2,
)
}
}
// applyChannelUpdate applies the channel update to the different sub-systems of
// the server. The useAlias boolean denotes whether or not to send an alias in
// place of the real SCID.
func (s *server) applyChannelUpdate(update *lnwire.ChannelUpdate,
op *wire.OutPoint, useAlias bool) error {
var (
peerAlias *lnwire.ShortChannelID
defaultAlias lnwire.ShortChannelID
)
chanID := lnwire.NewChanIDFromOutPoint(op)
// Fetch the peer's alias from the lnwire.ChannelID so it can be used
// in the ChannelUpdate if it hasn't been announced yet.
if useAlias {
foundAlias, _ := s.aliasMgr.GetPeerAlias(chanID)
if foundAlias != defaultAlias {
peerAlias = &foundAlias
}
}
errChan := s.authGossiper.ProcessLocalAnnouncement(
update, discovery.RemoteAlias(peerAlias),
)
select {
case err := <-errChan:
return err
case <-s.quit:
return ErrServerShuttingDown
}
}
// SendCustomMessage sends a custom message to the peer with the specified
// pubkey.
func (s *server) SendCustomMessage(peerPub [33]byte, msgType lnwire.MessageType,
data []byte) error {
peer, err := s.FindPeerByPubStr(string(peerPub[:]))
if err != nil {
return err
}
// We'll wait until the peer is active.
select {
case <-peer.ActiveSignal():
case <-peer.QuitSignal():
return fmt.Errorf("peer %x disconnected", peerPub)
case <-s.quit:
return ErrServerShuttingDown
}
msg, err := lnwire.NewCustom(msgType, data)
if err != nil {
return err
}
// Send the message as low-priority. For now we assume that all
// application-defined message are low priority.
return peer.SendMessageLazy(true, msg)
}
// newSweepPkScriptGen creates closure that generates a new public key script
// which should be used to sweep any funds into the on-chain wallet.
// Specifically, the script generated is a version 0, pay-to-witness-pubkey-hash
// (p2wkh) output.
func newSweepPkScriptGen(
wallet lnwallet.WalletController) func() ([]byte, error) {
return func() ([]byte, error) {
sweepAddr, err := wallet.NewAddress(
lnwallet.TaprootPubkey, false,
lnwallet.DefaultAccountName,
)
if err != nil {
return nil, err
}
return txscript.PayToAddrScript(sweepAddr)
}
}
// shouldPeerBootstrap returns true if we should attempt to perform peer
// bootstrapping to actively seek our peers using the set of active network
// bootstrappers.
func shouldPeerBootstrap(cfg *Config) bool {
isSimnet := (cfg.Bitcoin.SimNet || cfg.Litecoin.SimNet)
isSignet := (cfg.Bitcoin.SigNet || cfg.Litecoin.SigNet)
isRegtest := (cfg.Bitcoin.RegTest || cfg.Litecoin.RegTest)
isDevNetwork := isSimnet || isSignet || isRegtest
// TODO(yy): remove the check on simnet/regtest such that the itest is
// covering the bootstrapping process.
return !cfg.NoNetBootstrap && !isDevNetwork
}