lnd.xprv/server.go
Olaoluwa Osuntokun e1152148b7
server: only attempt to reconnect to peer if connection not pending
This commit fixes a bug that would possibly result in tens of goroutine
beaching launched in an attempt to persistently connect to a peer. This
bug has been fixed by ensuring that we’ll only launch a new pending
connection attempt if we don’t already have one pending.
2017-05-04 17:39:08 -07:00

1242 lines
35 KiB
Go

package main
import (
"bytes"
"crypto/rand"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"net"
"strconv"
"sync"
"sync/atomic"
"time"
"github.com/boltdb/bolt"
"github.com/lightningnetwork/lightning-onion"
"github.com/lightningnetwork/lnd/brontide"
"github.com/lightningnetwork/lnd/chainntnfs"
"github.com/lightningnetwork/lnd/channeldb"
"github.com/lightningnetwork/lnd/discovery"
"github.com/lightningnetwork/lnd/lnrpc"
"github.com/lightningnetwork/lnd/lnwallet"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/routing"
"github.com/roasbeef/btcd/btcec"
"github.com/roasbeef/btcd/chaincfg/chainhash"
"github.com/roasbeef/btcd/connmgr"
"github.com/roasbeef/btcutil"
)
// 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 {
started int32 // atomic
shutdown int32 // atomic
// identityPriv is the private key used to authenticate any incoming
// connections.
identityPriv *btcec.PrivateKey
// nodeSigner is an implementation of the MessageSigner implementation
// that's backed by the identituy private key of the running lnd node.
nodeSigner *nodeSigner
// lightningID is the sha256 of the public key corresponding to our
// long-term identity private key.
lightningID [32]byte
peersMtx sync.RWMutex
peersByID map[int32]*peer
peersByPub map[string]*peer
persistentPeers map[string]struct{}
inboundPeers map[string]*peer
outboundPeers map[string]*peer
rpcServer *rpcServer
chainNotifier chainntnfs.ChainNotifier
bio lnwallet.BlockChainIO
lnwallet *lnwallet.LightningWallet
fundingMgr *fundingManager
chanDB *channeldb.DB
htlcSwitch *htlcSwitch
invoices *invoiceRegistry
breachArbiter *breachArbiter
chanRouter *routing.ChannelRouter
discoverSrv *discovery.AuthenticatedGossiper
utxoNursery *utxoNursery
sphinx *sphinx.Router
connMgr *connmgr.ConnManager
pendingConnMtx sync.RWMutex
persistentConnReqs map[string][]*connmgr.ConnReq
broadcastRequests chan *broadcastReq
sendRequests chan *sendReq
newPeers chan *peer
donePeers chan *peer
queries chan interface{}
// globalFeatures feature vector which affects HTLCs and thus are also
// advertised to other nodes.
globalFeatures *lnwire.FeatureVector
// localFeatures is an feature vector which represent the features which
// only affect the protocol between these two nodes.
localFeatures *lnwire.FeatureVector
wg sync.WaitGroup
quit chan struct{}
}
// newServer creates a new instance of the server which is to listen using the
// passed listener address.
func newServer(listenAddrs []string, notifier chainntnfs.ChainNotifier,
bio lnwallet.BlockChainIO, fundingSigner lnwallet.MessageSigner,
wallet *lnwallet.LightningWallet, chanDB *channeldb.DB) (*server, error) {
privKey, err := wallet.GetIdentitykey()
if err != nil {
return nil, err
}
privKey.Curve = btcec.S256()
listeners := make([]net.Listener, len(listenAddrs))
for i, addr := range listenAddrs {
listeners[i], err = brontide.NewListener(privKey, addr)
if err != nil {
return nil, err
}
}
serializedPubKey := privKey.PubKey().SerializeCompressed()
s := &server{
lnwallet: wallet,
bio: bio,
chainNotifier: notifier,
chanDB: chanDB,
invoices: newInvoiceRegistry(chanDB),
utxoNursery: newUtxoNursery(chanDB, notifier, wallet),
htlcSwitch: newHtlcSwitch(),
identityPriv: privKey,
nodeSigner: newNodeSigner(privKey),
// TODO(roasbeef): derive proper onion key based on rotation
// schedule
sphinx: sphinx.NewRouter(privKey, activeNetParams.Params),
lightningID: sha256.Sum256(serializedPubKey),
persistentPeers: make(map[string]struct{}),
persistentConnReqs: make(map[string][]*connmgr.ConnReq),
peersByID: make(map[int32]*peer),
peersByPub: make(map[string]*peer),
inboundPeers: make(map[string]*peer),
outboundPeers: make(map[string]*peer),
newPeers: make(chan *peer, 10),
donePeers: make(chan *peer, 10),
broadcastRequests: make(chan *broadcastReq),
sendRequests: make(chan *sendReq),
globalFeatures: globalFeatures,
localFeatures: localFeatures,
queries: make(chan interface{}),
quit: make(chan struct{}),
}
// If the debug HTLC flag is on, then we invoice a "master debug"
// invoice which all outgoing payments will be sent and all incoming
// HTLCs with the debug R-Hash immediately settled.
if cfg.DebugHTLC {
kiloCoin := btcutil.Amount(btcutil.SatoshiPerBitcoin * 1000)
s.invoices.AddDebugInvoice(kiloCoin, *debugPre)
srvrLog.Debugf("Debug HTLC invoice inserted, preimage=%x, hash=%x",
debugPre[:], debugHash[:])
}
// If external IP addresses have been specified, add those to the list
// of this server's addresses.
selfAddrs := make([]net.Addr, 0, len(cfg.ExternalIPs))
for _, ip := range cfg.ExternalIPs {
var addr string
_, _, err = net.SplitHostPort(ip)
if err != nil {
addr = net.JoinHostPort(ip, strconv.Itoa(defaultPeerPort))
} else {
addr = ip
}
lnAddr, err := net.ResolveTCPAddr("tcp", addr)
if err != nil {
return nil, err
}
selfAddrs = append(selfAddrs, lnAddr)
}
chanGraph := chanDB.ChannelGraph()
// TODO(roasbeef): make alias configurable
alias := lnwire.NewAlias(hex.EncodeToString(serializedPubKey[:10]))
self := &channeldb.LightningNode{
LastUpdate: time.Now(),
Addresses: selfAddrs,
PubKey: privKey.PubKey(),
Alias: alias.String(),
Features: globalFeatures,
}
// If our information has changed since our last boot, then we'll
// re-sign our node announcement so a fresh authenticated version of it
// can be propagated throughout the network upon startup.
// TODO(roasbeef): don't always set timestamp above to _now.
self.AuthSig, err = discovery.SignAnnouncement(s.nodeSigner,
s.identityPriv.PubKey(),
&lnwire.NodeAnnouncement{
Timestamp: uint32(self.LastUpdate.Unix()),
Addresses: self.Addresses,
NodeID: self.PubKey,
Alias: alias,
Features: self.Features,
},
)
if err != nil {
return nil, fmt.Errorf("unable to generate signature for "+
"self node announcement: %v", err)
}
if err := chanGraph.SetSourceNode(self); err != nil {
return nil, fmt.Errorf("can't set self node: %v", err)
}
s.chanRouter, err = routing.New(routing.Config{
Graph: chanGraph,
Chain: bio,
Notifier: notifier,
SendToSwitch: func(firstHop *btcec.PublicKey,
htlcAdd *lnwire.UpdateAddHTLC) ([32]byte, error) {
firstHopPub := firstHop.SerializeCompressed()
destInterface := chainhash.Hash(sha256.Sum256(firstHopPub))
return s.htlcSwitch.SendHTLC(&htlcPacket{
dest: destInterface,
msg: htlcAdd,
})
},
})
if err != nil {
return nil, fmt.Errorf("can't create router: %v", err)
}
s.discoverSrv, err = discovery.New(discovery.Config{
Broadcast: s.broadcastMessage,
Notifier: s.chainNotifier,
Router: s.chanRouter,
SendToPeer: s.sendToPeer,
TrickleDelay: time.Millisecond * 300,
ProofMatureDelta: 0,
})
if err != nil {
return nil, err
}
s.rpcServer = newRPCServer(s)
s.breachArbiter = newBreachArbiter(wallet, chanDB, notifier, s.htlcSwitch)
var chanIDSeed [32]byte
if _, err := rand.Read(chanIDSeed[:]); err != nil {
return nil, err
}
s.fundingMgr, err = newFundingManager(fundingConfig{
IDKey: s.identityPriv.PubKey(),
Wallet: wallet,
Notifier: s.chainNotifier,
SignMessage: func(pubKey *btcec.PublicKey, msg []byte) (*btcec.Signature, error) {
if pubKey.IsEqual(s.identityPriv.PubKey()) {
return s.nodeSigner.SignMessage(pubKey, msg)
}
return fundingSigner.SignMessage(pubKey, msg)
},
SendAnnouncement: func(msg lnwire.Message) error {
s.discoverSrv.ProcessLocalAnnouncement(msg,
s.identityPriv.PubKey())
return nil
},
ArbiterChan: s.breachArbiter.newContracts,
SendToPeer: s.sendToPeer,
FindPeer: s.findPeer,
TempChanIDSeed: chanIDSeed,
FindChannel: func(chanID lnwire.ChannelID) (*lnwallet.LightningChannel, error) {
dbChannels, err := chanDB.FetchAllChannels()
if err != nil {
return nil, err
}
for _, channel := range dbChannels {
if chanID.IsChanPoint(channel.ChanID) {
return lnwallet.NewLightningChannel(wallet.Signer,
notifier, channel)
}
}
return nil, fmt.Errorf("unable to find channel")
},
})
if err != nil {
return nil, err
}
// TODO(roasbeef): introduce closure and config system to decouple the
// initialization above ^
// 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,
GetNewAddress: nil,
Dial: noiseDial(s.identityPriv),
OnConnection: s.outboundPeerConnected,
})
if err != nil {
return nil, err
}
s.connMgr = cmgr
return s, nil
}
// Start starts the main daemon server, all requested listeners, and any helper
// goroutines.
func (s *server) Start() error {
// Already running?
if atomic.AddInt32(&s.started, 1) != 1 {
return nil
}
// 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.chainNotifier.Start(); err != nil {
return err
}
if err := s.rpcServer.Start(); err != nil {
return err
}
if err := s.fundingMgr.Start(); err != nil {
return err
}
if err := s.htlcSwitch.Start(); err != nil {
return err
}
if err := s.utxoNursery.Start(); err != nil {
return err
}
if err := s.breachArbiter.Start(); err != nil {
return err
}
if err := s.discoverSrv.Start(); err != nil {
return err
}
if err := s.chanRouter.Start(); err != nil {
return err
}
s.wg.Add(1)
go s.queryHandler()
// With all the relevant sub-systems started, we'll now atetmpt to
// stasblish persistent connections to our direct channel collaborators
// within the network.
if err := s.establishPersistentConnections(); err != nil {
return err
}
return nil
}
// 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.
func (s *server) Stop() error {
// Bail if we're already shutting down.
if atomic.AddInt32(&s.shutdown, 1) != 1 {
return nil
}
// Shutdown the wallet, funding manager, and the rpc server.
s.chainNotifier.Stop()
s.rpcServer.Stop()
s.fundingMgr.Stop()
s.chanRouter.Stop()
s.htlcSwitch.Stop()
s.utxoNursery.Stop()
s.breachArbiter.Stop()
s.discoverSrv.Stop()
s.lnwallet.Shutdown()
// Signal all the lingering goroutines to quit.
close(s.quit)
s.wg.Wait()
return nil
}
// 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.chanDB.FetchAllLinkNodes()
if err != nil && err != channeldb.ErrLinkNodesNotFound {
return err
}
for _, node := range linkNodes {
for _, address := range node.Addresses {
if address.Port == 0 {
address.Port = defaultPeerPort
}
}
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.
chanGraph := s.chanDB.ChannelGraph()
sourceNode, err := chanGraph.SourceNode()
if err != nil {
return err
}
// TODO(roasbeef): instead iterate over link nodes and query graph for
// each of the nodes.
err = sourceNode.ForEachChannel(nil, func(_ *bolt.Tx,
_ *channeldb.ChannelEdgeInfo, policy *channeldb.ChannelEdgePolicy) error {
pubStr := string(policy.Node.PubKey.SerializeCompressed())
// Add addresses from channel graph/NodeAnnouncements to the
// list of addresses we'll connect to. If there are duplicates
// that have different ports specified, the port from the
// channel graph should supersede the port from the link node.
var addrs []*net.TCPAddr
linkNodeAddrs, ok := nodeAddrsMap[pubStr]
if ok {
for _, lnAddress := range linkNodeAddrs.addresses {
var addrMatched bool
for _, polAddress := range policy.Node.Addresses {
polTCPAddr, ok := polAddress.(*net.TCPAddr)
if ok && polTCPAddr.IP.Equal(lnAddress.IP) {
addrMatched = true
addrs = append(addrs, polTCPAddr)
}
}
if !addrMatched {
addrs = append(addrs, lnAddress)
}
}
} else {
for _, addr := range policy.Node.Addresses {
polTCPAddr, ok := addr.(*net.TCPAddr)
if ok {
addrs = append(addrs, polTCPAddr)
}
}
}
nodeAddrsMap[pubStr] = &nodeAddresses{
pubKey: policy.Node.PubKey,
addresses: addrs,
}
return nil
})
if err != nil && err != channeldb.ErrGraphNoEdgesFound {
return err
}
// Iterate through the combined list of addresses from prior links and
// node announcements and attempt to reconnect to each node.
for pubStr, nodeAddr := range nodeAddrsMap {
// Add this peer to the set of peers we should maintain a
// persistent connection with.
s.persistentPeers[pubStr] = struct{}{}
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,
}
srvrLog.Debugf("Attempting persistent connection to "+
"channel peer %v", lnAddr)
// 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: lnAddr,
Permanent: true,
}
s.pendingConnMtx.Lock()
s.persistentConnReqs[pubStr] = append(s.persistentConnReqs[pubStr],
connReq)
s.pendingConnMtx.Unlock()
go s.connMgr.Connect(connReq)
}
}
return nil
}
// WaitForShutdown blocks all goroutines have been stopped.
func (s *server) WaitForShutdown() {
s.wg.Wait()
}
// broadcastReq is a message sent to the server by a related subsystem when it
// wishes to broadcast one or more messages to all connected peers. Thi
type broadcastReq struct {
ignore *btcec.PublicKey
msgs []lnwire.Message
errChan chan error // MUST be buffered.
}
// broadcastMessage sends a request to the server to broadcast a set of
// messages to all peers other than the one specified by the `skip` parameter.
func (s *server) broadcastMessage(skip *btcec.PublicKey, msgs ...lnwire.Message) error {
errChan := make(chan error, 1)
msgsToSend := make([]lnwire.Message, 0, len(msgs))
msgsToSend = append(msgsToSend, msgs...)
broadcastReq := &broadcastReq{
ignore: skip,
msgs: msgsToSend,
errChan: errChan,
}
select {
case s.broadcastRequests <- broadcastReq:
case <-s.quit:
return errors.New("server shutting down")
}
select {
case err := <-errChan:
return err
case <-s.quit:
return errors.New("server shutting down")
}
}
// sendReq is message sent to the server by a related subsystem which it
// wishes to send a set of messages to a specified peer.
type sendReq struct {
target *btcec.PublicKey
msgs []lnwire.Message
errChan chan error
}
type nodeAddresses struct {
pubKey *btcec.PublicKey
addresses []*net.TCPAddr
}
// sendToPeer send a message to the server telling it to send the specific set
// of message to a particular peer. If the peer connect be found, then this
// method will return a non-nil error.
func (s *server) sendToPeer(target *btcec.PublicKey, msgs ...lnwire.Message) error {
errChan := make(chan error, 1)
msgsToSend := make([]lnwire.Message, 0, len(msgs))
msgsToSend = append(msgsToSend, msgs...)
sMsg := &sendReq{
target: target,
msgs: msgsToSend,
errChan: errChan,
}
select {
case s.sendRequests <- sMsg:
case <-s.quit:
return errors.New("server shutting down")
}
select {
case err := <-errChan:
return err
case <-s.quit:
return errors.New("server shutting down")
}
}
// 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.
func (s *server) findPeer(peerKey *btcec.PublicKey) (*peer, error) {
serializedIDKey := string(peerKey.SerializeCompressed())
s.peersMtx.RLock()
peer := s.peersByPub[serializedIDKey]
s.peersMtx.RUnlock()
if peer == nil {
return nil, errors.New("Peer not found. Pubkey: " +
string(peerKey.SerializeCompressed()))
}
return peer, nil
}
// peerTerminationWatcher waits until a peer has been disconnected, 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.
//
// NOTE: This MUST be launched as a goroutine.
func (s *server) peerTerminationWatcher(p *peer) {
p.WaitForDisconnect()
srvrLog.Debugf("Peer %v has been disconnected", p)
// Tell the switch to unregister all links associated with this peer.
// Passing nil as the target link indicates that all
// links associated with this interface should be closed.
p.server.htlcSwitch.UnregisterLink(p.addr.IdentityKey, nil)
// Send the peer to be garbage collected by the server.
p.server.donePeers <- p
// If this peer had an active persistent connection request, then we
// can remove this as we manually decide below if we should attempt to
// re-connect.
if p.connReq != nil {
s.connMgr.Remove(p.connReq.ID())
}
// Next, check to see if this is a persistent peer or not.
pubStr := string(p.addr.IdentityKey.SerializeCompressed())
if _, ok := s.persistentPeers[pubStr]; ok {
srvrLog.Debugf("Attempting to re-establish persistent "+
"connection to peer %v", p)
// If so, then we'll attempt to re-establish a persistent
// connection to the peer.
// TODO(roasbeef): look up latest info for peer in database
connReq := &connmgr.ConnReq{
Addr: p.addr,
Permanent: true,
}
s.pendingConnMtx.Lock()
// We'll only need to re-launch a connection requests if one
// isn't already currently pending.
if _, ok := s.persistentConnReqs[pubStr]; ok {
return
}
// Otherwise, we'll launch a new connection requests in order
// to attempt to maintain a persistent connection with this
// peer.
s.persistentConnReqs[pubStr] = append(s.persistentConnReqs[pubStr],
connReq)
s.pendingConnMtx.Unlock()
go s.connMgr.Connect(connReq)
}
}
// 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.
func (s *server) peerConnected(conn net.Conn, connReq *connmgr.ConnReq, inbound bool) {
brontideConn := conn.(*brontide.Conn)
peerAddr := &lnwire.NetAddress{
IdentityKey: brontideConn.RemotePub(),
Address: conn.RemoteAddr().(*net.TCPAddr),
ChainNet: activeNetParams.Net,
}
// Now that we've established a connection, create a peer, and
// it to the set of currently active peers.
p, err := newPeer(conn, connReq, s, peerAddr, inbound)
if err != nil {
srvrLog.Errorf("unable to create peer %v", err)
return
}
// TODO(roasbeef): update IP address for link-node
// * also mark last-seen, do it one single transaction?
// Attempt to start the peer, if we're unable to do so, then disconnect
// this peer.
if err := p.Start(); err != nil {
srvrLog.Errorf("unable to start peer: %v", err)
p.Disconnect()
return
}
s.newPeers <- p
}
// 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.
func (s *server) inboundPeerConnected(conn net.Conn) {
s.peersMtx.Lock()
defer s.peersMtx.Unlock()
srvrLog.Infof("New inbound connection from %v", conn.RemoteAddr())
localPub := s.identityPriv.PubKey()
nodePub := conn.(*brontide.Conn).RemotePub()
// Check to see if we should drop our connection, if not, then we'll
// close out this connection with the remote peer. This
// prevents us from having duplicate connections, or none.
pubStr := string(nodePub.SerializeCompressed())
if connectedPeer, ok := s.peersByPub[pubStr]; ok {
// If the connection we've already established should be kept,
// then we'll close out this connection s.t there's only a
// single connection between us.
if !shouldDropLocalConnection(localPub, nodePub) {
srvrLog.Warnf("Received inbound connection from "+
"peer %x, but already connected, dropping conn",
nodePub.SerializeCompressed())
conn.Close()
return
}
// Otherwise, if we should drop the connection, then we'll
// disconnect our already connected peer, and also send the
// peer to the peer garbage collection goroutine.
srvrLog.Debugf("Disconnecting stale connection to %v",
connectedPeer)
connectedPeer.Disconnect()
s.donePeers <- connectedPeer
}
// 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.
s.pendingConnMtx.Lock()
if connReqs, ok := s.persistentConnReqs[pubStr]; ok {
for _, connReq := range connReqs {
s.connMgr.Remove(connReq.ID())
}
delete(s.persistentConnReqs, pubStr)
}
s.pendingConnMtx.Unlock()
go s.peerConnected(conn, nil, false)
}
// outboundPeerConnected initializes a new peer in response to a new outbound
// connection.
func (s *server) outboundPeerConnected(connReq *connmgr.ConnReq, conn net.Conn) {
s.peersMtx.Lock()
defer s.peersMtx.Unlock()
localPub := s.identityPriv.PubKey()
nodePub := conn.(*brontide.Conn).RemotePub()
pubStr := string(nodePub.SerializeCompressed())
// If we already have an outbound connection to this peer, then ignore
// this new connection.
if _, ok := s.outboundPeers[pubStr]; ok {
srvrLog.Debugf("Ignoring duplicate outbound connection")
conn.Close()
return
}
if _, ok := s.persistentConnReqs[pubStr]; !ok && connReq != nil {
srvrLog.Debugf("Ignoring cancelled outbound connection")
conn.Close()
return
}
srvrLog.Infof("Established connection to: %v", conn.RemoteAddr())
// As we've just established an outbound connection to this peer, we'll
// cancel all other persistent connection requests and eliminate the
// entry for this peer from the map.
s.pendingConnMtx.Lock()
if connReqs, ok := s.persistentConnReqs[pubStr]; ok {
for _, pConnReq := range connReqs {
if pConnReq.ID() != connReq.ID() {
s.connMgr.Remove(pConnReq.ID())
}
}
delete(s.persistentConnReqs, pubStr)
}
s.pendingConnMtx.Unlock()
// If we already have an inbound connection from this peer, then we'll
// check to see _which_ of our connections should be dropped.
if connectedPeer, ok := s.peersByPub[pubStr]; ok {
// If our (this) connection should be dropped, then we'll do
// so, in order to ensure we don't have any duplicate
// connections.
if shouldDropLocalConnection(localPub, nodePub) {
srvrLog.Warnf("Established outbound connection to "+
"peer %x, but already connected, dropping conn",
nodePub.SerializeCompressed())
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)
connectedPeer.Disconnect()
s.donePeers <- connectedPeer
}
go s.peerConnected(conn, connReq, true)
}
// addPeer adds the passed peer to the server's global state of all active
// peers.
func (s *server) addPeer(p *peer) {
if p == nil {
return
}
// Ignore new peers if we're shutting down.
if atomic.LoadInt32(&s.shutdown) != 0 {
p.Disconnect()
return
}
// Track the new peer in our indexes so we can quickly look it up either
// according to its public key, or it's peer ID.
// TODO(roasbeef): pipe all requests through to the
// queryHandler/peerManager
s.peersMtx.Lock()
pubStr := string(p.addr.IdentityKey.SerializeCompressed())
s.peersByID[p.id] = p
s.peersByPub[pubStr] = p
if p.inbound {
s.inboundPeers[pubStr] = p
} else {
s.outboundPeers[pubStr] = p
}
s.peersMtx.Unlock()
// Launch a goroutine to watch for the termination of this peer so we
// can ensure all resources are properly cleaned up and if need be
// connections are re-established.
go s.peerTerminationWatcher(p)
// Once the peer has been added to our indexes, send a message to the
// channel router so we can synchronize our view of the channel graph
// with this new peer.
go s.discoverSrv.SynchronizeNode(p.addr.IdentityKey)
}
// removePeer removes the passed peer from the server's state of all active
// peers.
func (s *server) removePeer(p *peer) {
s.peersMtx.Lock()
defer s.peersMtx.Unlock()
srvrLog.Debugf("removing peer %v", p)
if p == nil {
return
}
// As the peer is now finished, ensure that the TCP connection is
// closed and all of its related goroutines have exited.
p.Disconnect()
// Ignore deleting peers if we're shutting down.
if atomic.LoadInt32(&s.shutdown) != 0 {
return
}
pubStr := string(p.addr.IdentityKey.SerializeCompressed())
delete(s.peersByID, p.id)
delete(s.peersByPub, pubStr)
if p.inbound {
delete(s.inboundPeers, pubStr)
} else {
delete(s.outboundPeers, pubStr)
}
}
// connectPeerMsg is a message requesting the server to open a connection to a
// particular peer. This message also houses an error channel which will be
// used to report success/failure.
type connectPeerMsg struct {
addr *lnwire.NetAddress
persistent bool
err chan error
}
// listPeersMsg is a message sent to the server in order to obtain a listing
// of all currently active channels.
type listPeersMsg struct {
resp chan []*peer
}
// openChanReq is a message sent to the server in order to request the
// initiation of a channel funding workflow to the peer with either the specified
// relative peer ID, or a global lightning ID.
type openChanReq struct {
targetPeerID int32
targetPubkey *btcec.PublicKey
// TODO(roasbeef): make enums in lnwire
channelType uint8
coinType uint64
localFundingAmt btcutil.Amount
remoteFundingAmt btcutil.Amount
pushAmt btcutil.Amount
numConfs uint32
updates chan *lnrpc.OpenStatusUpdate
err chan error
}
// queryHandler handles any requests to modify the server's internal state of
// all active peers, or query/mutate the server's global state. Additionally,
// any queries directed at peers will be handled by this goroutine.
//
// NOTE: This MUST be run as a goroutine.
func (s *server) queryHandler() {
go s.connMgr.Start()
out:
for {
select {
// New peers.
case p := <-s.newPeers:
s.addPeer(p)
// Finished peers.
case p := <-s.donePeers:
s.removePeer(p)
case bMsg := <-s.broadcastRequests:
ignore := bMsg.ignore
srvrLog.Debugf("Broadcasting %v messages", len(bMsg.msgs))
// Launch a new goroutine to handle the broadcast
// request, this allows us process this request
// asynchronously without blocking subsequent broadcast
// requests.
go func() {
s.peersMtx.RLock()
for _, sPeer := range s.peersByPub {
if ignore != nil &&
sPeer.addr.IdentityKey.IsEqual(ignore) {
srvrLog.Debugf("Skipping %v in broadcast",
ignore.SerializeCompressed())
continue
}
go func(p *peer) {
for _, msg := range bMsg.msgs {
p.queueMsg(msg, nil)
}
}(sPeer)
}
s.peersMtx.RUnlock()
bMsg.errChan <- nil
}()
case sMsg := <-s.sendRequests:
// TODO(roasbeef): use [33]byte everywhere instead
// * eliminate usage of mutexes, funnel all peer
// mutation to this goroutine
target := sMsg.target.SerializeCompressed()
srvrLog.Debugf("Attempting to send msgs %v to: %x",
len(sMsg.msgs), target)
// Launch a new goroutine to handle this send request,
// this allows us process this request asynchronously
// without blocking future send requests.
go func() {
s.peersMtx.RLock()
targetPeer, ok := s.peersByPub[string(target)]
if !ok {
s.peersMtx.RUnlock()
srvrLog.Errorf("unable to send message to %x, "+
"peer not found", target)
sMsg.errChan <- errors.New("peer not found")
return
}
s.peersMtx.RUnlock()
sMsg.errChan <- nil
for _, msg := range sMsg.msgs {
targetPeer.queueMsg(msg, nil)
}
}()
case query := <-s.queries:
switch msg := query.(type) {
case *connectPeerMsg:
s.handleConnectPeer(msg)
case *listPeersMsg:
s.handleListPeers(msg)
case *openChanReq:
s.handleOpenChanReq(msg)
}
case <-s.quit:
break out
}
}
s.connMgr.Stop()
s.wg.Done()
}
// handleListPeers sends a lice of all currently active peers to the original
// caller.
func (s *server) handleListPeers(msg *listPeersMsg) {
s.peersMtx.RLock()
peers := make([]*peer, 0, len(s.peersByID))
for _, peer := range s.peersByID {
peers = append(peers, peer)
}
s.peersMtx.RUnlock()
msg.resp <- peers
}
// handleConnectPeer attempts to establish a connection to the address enclosed
// within the passed connectPeerMsg. This function is *async*, a goroutine will
// be spawned in order to finish the request, and respond to the caller.
func (s *server) handleConnectPeer(msg *connectPeerMsg) {
addr := msg.addr
targetPub := string(msg.addr.IdentityKey.SerializeCompressed())
// Ensure we're not already connected to this
// peer.
s.peersMtx.RLock()
peer, ok := s.peersByPub[targetPub]
if ok {
s.peersMtx.RUnlock()
msg.err <- fmt.Errorf("already connected to peer: %v", peer)
return
}
s.peersMtx.RUnlock()
// 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.
s.pendingConnMtx.RLock()
if _, ok := s.persistentConnReqs[targetPub]; ok {
s.pendingConnMtx.RUnlock()
msg.err <- fmt.Errorf("connection attempt to %v is pending",
addr)
return
}
s.pendingConnMtx.RUnlock()
// 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 msg.persistent {
go s.connMgr.Connect(&connmgr.ConnReq{
Addr: addr,
Permanent: true,
})
msg.err <- nil
} else {
// If we're not making a persistent connection, then we'll
// attempt to connect o the target peer, returning an error
// which indicates success of failure.
go func() {
// Attempt to connect to the remote node. If the we
// can't make the connection, or the crypto negotiation
// breaks down, then return an error to the caller.
conn, err := brontide.Dial(s.identityPriv, addr)
if err != nil {
msg.err <- err
return
}
s.outboundPeerConnected(nil, conn)
msg.err <- nil
}()
}
}
// handleOpenChanReq first locates the target peer, and if found hands off the
// request to the funding manager allowing it to initiate the channel funding
// workflow.
func (s *server) handleOpenChanReq(req *openChanReq) {
var (
targetPeer *peer
pubKeyBytes []byte
)
// If the user is targeting the peer by public key, then we'll need to
// convert that into a string for our map. Otherwise, we expect them to
// target by peer ID instead.
if req.targetPubkey != nil {
pubKeyBytes = req.targetPubkey.SerializeCompressed()
}
// 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.
s.peersMtx.RLock()
if peer, ok := s.peersByID[req.targetPeerID]; ok {
targetPeer = peer
} else if peer, ok := s.peersByPub[string(pubKeyBytes)]; ok {
targetPeer = peer
}
s.peersMtx.RUnlock()
if targetPeer == nil {
req.err <- fmt.Errorf("unable to find peer nodeID(%x), "+
"peerID(%v)", pubKeyBytes, req.targetPeerID)
return
}
// 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.
// TODO(roasbeef): server semaphore to restrict num goroutines
go s.fundingMgr.initFundingWorkflow(targetPeer.addr, req)
}
// 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.
func (s *server) ConnectToPeer(addr *lnwire.NetAddress,
perm bool) error {
errChan := make(chan error, 1)
s.queries <- &connectPeerMsg{
addr: addr,
persistent: perm,
err: errChan,
}
return <-errChan
}
// OpenChannel sends a request to the server to open a channel to the specified
// peer identified by ID with the passed channel funding paramters.
func (s *server) OpenChannel(peerID int32, nodeKey *btcec.PublicKey,
localAmt, pushAmt btcutil.Amount,
numConfs uint32) (chan *lnrpc.OpenStatusUpdate, chan error) {
errChan := make(chan error, 1)
updateChan := make(chan *lnrpc.OpenStatusUpdate, 1)
req := &openChanReq{
targetPeerID: peerID,
targetPubkey: nodeKey,
localFundingAmt: localAmt,
pushAmt: pushAmt,
numConfs: numConfs,
updates: updateChan,
err: errChan,
}
s.queries <- req
return updateChan, errChan
}
// Peers returns a slice of all active peers.
func (s *server) Peers() []*peer {
resp := make(chan []*peer, 1)
s.queries <- &listPeersMsg{resp}
return <-resp
}