13593508312

Committed 28 Feb 2025 05:41PM UTC coverage: 58.287% (-10.4%) from 68.65%

Build # 13593508312

Build Type

Pull #9458

github

Committed by

web-flow

Commit Message

Merge d40067c0c into f1182e433

Pull Request Pull Request #9458: multi+server.go: add initial permissions for some peers

Run Details

346 of 548 new or added lines in 10 files covered. (63.14%)

27412 existing lines in 442 files now uncovered.

94709 of 162488 relevant lines covered (58.29%)

1.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.03

/server.go

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"
        "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/chainio"
        "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/cluster"
        "github.com/lightningnetwork/lnd/contractcourt"
        "github.com/lightningnetwork/lnd/discovery"
        "github.com/lightningnetwork/lnd/feature"
        "github.com/lightningnetwork/lnd/fn/v2"
        "github.com/lightningnetwork/lnd/funding"
        "github.com/lightningnetwork/lnd/graph"
        graphdb "github.com/lightningnetwork/lnd/graph/db"
        "github.com/lightningnetwork/lnd/graph/db/models"
        "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/chanfunding"
        "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

        // EndorsementExperimentEnd is the time after which nodes should stop
        // propagating experimental endorsement signals.
        //
        // Per blip04: January 1, 2026 12:00:00 AM UTC in unix seconds.
        EndorsementExperimentEnd = time.Unix(1767225600, 0)

        // ErrGossiperBan is one of the errors that can be returned when we
        // attempt to finalize a connection to a remote peer.
        ErrGossiperBan = errors.New("gossiper has banned remote's key")

        // ErrNoMoreRestrictedAccessSlots is one of the errors that can be
        // returned when we attempt to finalize a connection. It means that
        // this peer has no pending-open, open, or closed channels with us and
        // are already at our connection ceiling for a peer with this access
        // status.
        ErrNoMoreRestrictedAccessSlots = errors.New("no more restricted slots")

        // ErrNoPeerScore is returned when we expect to find a score in
        // peerScores, but one does not exist.
        ErrNoPeerScore = errors.New("peer score not found")

        // ErrNoPendingPeerInfo is returned when we couldn't find any pending
        // peer info.
        ErrNoPendingPeerInfo = errors.New("no pending peer info")
)

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

// peerAccessStatus denotes the p2p access status of a given peer. This will be
// used to assign peer ban scores that determine an action the server will
// take.
type peerAccessStatus int

const (
        // peerStatusRestricted indicates that the peer only has access to the
        // limited number of "free" reserved slots.
        peerStatusRestricted peerAccessStatus = iota

        // peerStatusTemporary indicates that the peer only has temporary p2p
        // access to the server.
        peerStatusTemporary

        // peerStatusProtected indicates that the peer has been granted
        // permanent p2p access to the server. The peer can still have its
        // access revoked.
        peerStatusProtected
)

// peerSlotStatus determines whether a peer gets access to one of our free
// slots or gets to bypass this safety mechanism.
type peerSlotStatus struct {
        // state determines which privileges the peer has with our server.
        state peerAccessStatus
}

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

        implCfg *ImplementationCfg

        // 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 is a map of the active peers.
        //
        // NOTE: The key used here is the raw bytes of the peer's public key to
        // string conversion, which means it cannot be printed using `%s` as it
        // will just print the binary.
        //
        // TODO(yy): Use the hex string instead.
        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 *graphdb.ChannelGraph

        chanStateDB *channeldb.ChannelStateDB

        addrSource channeldb.AddrSource

        // miscDB is the DB that contains all "other" databases within the main
        // channel DB that haven't been separated out yet.
        miscDB *channeldb.DB

        invoicesDB invoices.InvoiceDB

        aliasMgr *aliasmgr.Manager

        htlcSwitch *htlcswitch.Switch

        interceptableSwitch *htlcswitch.InterceptableSwitch

        invoices *invoices.InvoiceRegistry

        invoiceHtlcModifier *invoices.HtlcModificationInterceptor

        channelNotifier *channelnotifier.ChannelNotifier

        peerNotifier *peernotifier.PeerNotifier

        htlcNotifier *htlcswitch.HtlcNotifier

        witnessBeacon contractcourt.WitnessBeacon

        breachArbitrator *contractcourt.BreachArbitrator

        missionController *routing.MissionController
        defaultMC         *routing.MissionControl

        graphBuilder *graph.Builder

        chanRouter *routing.ChannelRouter

        controlTower routing.ControlTower

        authGossiper *discovery.AuthenticatedGossiper

        localChanMgr *localchans.Manager

        utxoNursery *contractcourt.UtxoNursery

        sweeper *sweep.UtxoSweeper

        chainArb *contractcourt.ChainArbitrator

        sphinx *hop.OnionProcessor

        towerClientMgr *wtclient.Manager

        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

        // txPublisher is a publisher with fee-bumping capability.
        txPublisher *sweep.TxPublisher

        // blockbeatDispatcher is a block dispatcher that notifies subscribers
        // of new blocks.
        blockbeatDispatcher *chainio.BlockbeatDispatcher

        // peerAccessMan implements peer access controls.
        peerAccessMan *accessMan

        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.graphBuilder.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,
        leaderElector cluster.LeaderElector,
        implCfg *ImplementationCfg) (*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,
                )
        )

        var serializedPubKey [33]byte
        copy(serializedPubKey[:], nodeKeyDesc.PubKey.SerializeCompressed())

        netParams := cfg.ActiveNetParams.Params

        // Initialize the sphinx router.
        replayLog := htlcswitch.NewDecayedLog(
                dbs.DecayedLogDB, cc.ChainNotifier,
        )
        sphinxRouter := sphinx.NewRouter(nodeKeyECDH, 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,
        )

        // If the taproot overlay flag is set, but we don't have an aux funding
        // controller, then we'll exit as this is incompatible.
        if cfg.ProtocolOptions.TaprootOverlayChans &&
                implCfg.AuxFundingController.IsNone() {

                return nil, fmt.Errorf("taproot overlay flag set, but not " +
                        "aux controllers")
        }

        //nolint:ll
        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.CustomFeatures(),
                NoTaprootChans:            !cfg.ProtocolOptions.TaprootChans,
                NoTaprootOverlay:          !cfg.ProtocolOptions.TaprootOverlayChans,
                NoRouteBlinding:           cfg.ProtocolOptions.NoRouteBlinding(),
                NoExperimentalEndorsement: cfg.ProtocolOptions.NoExperimentalEndorsement(),
                NoQuiescence:              cfg.ProtocolOptions.NoQuiescence(),
        })
        if err != nil {
                return nil, err
        }

        invoiceHtlcModifier := invoices.NewHtlcModificationInterceptor()
        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,
                HtlcInterceptor:             invoiceHtlcModifier,
        }

        addrSource := channeldb.NewMultiAddrSource(dbs.ChanStateDB, dbs.GraphDB)

        s := &server{
                cfg:            cfg,
                implCfg:        implCfg,
                graphDB:        dbs.GraphDB,
                chanStateDB:    dbs.ChanStateDB.ChannelStateDB(),
                addrSource:     addrSource,
                miscDB:         dbs.ChanStateDB,
                invoicesDB:     dbs.InvoiceDB,
                cc:             cc,
                sigPool:        lnwallet.NewSigPool(cfg.Workers.Sig, cc.Signer),
                writePool:      writePool,
                readPool:       readPool,
                chansToRestore: chansToRestore,

                blockbeatDispatcher: chainio.NewBlockbeatDispatcher(
                        cc.ChainNotifier,
                ),
                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{}),

                invoiceHtlcModifier: invoiceHtlcModifier,

                customMessageServer: subscribe.NewServer(),

                tlsManager: tlsManager,

                featureMgr: featureMgr,
                quit:       make(chan struct{}),
        }

        // Start the low-level services once they are initialized.
        //
        // TODO(yy): break the server startup into four steps,
        // 1. init the low-level services.
        // 2. start the low-level services.
        // 3. init the high-level services.
        // 4. start the high-level services.
        if err := s.startLowLevelServices(); err != nil {
                return nil, err
        }

        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, &registryConfig,
        )

        s.htlcNotifier = htlcswitch.NewHtlcNotifier(time.Now)

        thresholdSats := btcutil.Amount(cfg.MaxFeeExposure)
        thresholdMSats := lnwire.NewMSatFromSatoshis(thresholdSats)

        linkUpdater := func(shortID lnwire.ShortChannelID) error {
                link, err := s.htlcSwitch.GetLinkByShortID(shortID)
                if err != nil {
                        return err
                }

                s.htlcSwitch.UpdateLinkAliases(link)

                return nil
        }

        s.aliasMgr, err = aliasmgr.NewManager(dbs.ChanStateDB, linkUpdater)
        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,
                MaxFeeExposure:         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,
        }

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

        // 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 := &models.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: %w", 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 := dbs.GraphDB.SetSourceNode(selfNode); err != nil {
                return nil, fmt.Errorf("can't set self node: %w", 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{
                OnConfigUpdate:          fn.Some(s.UpdateRoutingConfig),
                Estimator:               estimator,
                MaxMcHistory:            routingConfig.MaxMcHistory,
                McFlushInterval:         routingConfig.McFlushInterval,
                MinFailureRelaxInterval: routing.DefaultMinFailureRelaxInterval,
        }

        s.missionController, err = routing.NewMissionController(
                dbs.ChanStateDB, selfNode.PubKeyBytes, mcCfg,
        )
        if err != nil {
                return nil, fmt.Errorf("can't create mission control "+
                        "manager: %w", err)
        }
        s.defaultMC, err = s.missionController.GetNamespacedStore(
                routing.DefaultMissionControlNamespace,
        )
        if err != nil {
                return nil, fmt.Errorf("can't create mission control in the "+
                        "default namespace: %w", 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 := dbs.GraphDB.SourceNode()
        if err != nil {
                return nil, fmt.Errorf("error getting source node: %w", err)
        }
        paymentSessionSource := &routing.SessionSource{
                GraphSessionFactory: dbs.GraphDB,
                SourceNode:          sourceNode,
                MissionControl:      s.defaultMC,
                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.graphBuilder, err = graph.NewBuilder(&graph.Config{
                SelfNode:            selfNode.PubKeyBytes,
                Graph:               dbs.GraphDB,
                Chain:               cc.ChainIO,
                ChainView:           cc.ChainView,
                Notifier:            cc.ChainNotifier,
                ChannelPruneExpiry:  graph.DefaultChannelPruneExpiry,
                GraphPruneInterval:  time.Hour,
                FirstTimePruneDelay: graph.DefaultFirstTimePruneDelay,
                AssumeChannelValid:  cfg.Routing.AssumeChannelValid,
                StrictZombiePruning: strictPruning,
                IsAlias:             aliasmgr.IsAlias,
        })
        if err != nil {
                return nil, fmt.Errorf("can't create graph builder: %w", err)
        }

        s.chanRouter, err = routing.New(routing.Config{
                SelfNode:           selfNode.PubKeyBytes,
                RoutingGraph:       dbs.GraphDB,
                Chain:              cc.ChainIO,
                Payer:              s.htlcSwitch,
                Control:            s.controlTower,
                MissionControl:     s.defaultMC,
                SessionSource:      paymentSessionSource,
                GetLink:            s.htlcSwitch.GetLinkByShortID,
                NextPaymentID:      sequencer.NextID,
                PathFindingConfig:  pathFindingConfig,
                Clock:              clock.NewDefaultClock(),
                ApplyChannelUpdate: s.graphBuilder.ApplyChannelUpdate,
                ClosedSCIDs:        s.fetchClosedChannelSCIDs(),
                TrafficShaper:      implCfg.TrafficShaper,
        })
        if err != nil {
                return nil, fmt.Errorf("can't create router: %w", 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
        }

        scidCloserMan := discovery.NewScidCloserMan(s.graphDB, s.chanStateDB)

        s.authGossiper = discovery.New(discovery.Config{
                Graph:                 s.graphBuilder,
                ChainIO:               s.cc.ChainIO,
                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:        cfg.Gossip.AnnouncementConf,
                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,
                NoTimestampQueries:      cfg.ProtocolOptions.NoTimestampQueryOption, //nolint:ll
                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,
                IsStillZombieChannel:    s.graphBuilder.IsZombieChannel,
                ScidCloser:              scidCloserMan,
                AssumeChannelValid:      cfg.Routing.AssumeChannelValid,
        }, nodeKeyDesc)

        accessCfg := &accessManConfig{
                initAccessPerms: func() (map[string]channeldb.ChanCount,
                        error) {

                        genesisHash := *s.cfg.ActiveNetParams.GenesisHash
                        return s.chanStateDB.FetchPermAndTempPeers(
                                genesisHash[:],
                        )
                },
                shouldDisconnect:   s.authGossiper.ShouldDisconnect,
                maxRestrictedSlots: s.cfg.NumRestrictedSlots,
        }

        peerAccessMan, err := newAccessMan(accessCfg)
        if err != nil {
                return nil, err
        }

        s.peerAccessMan = peerAccessMan

        selfVertex := route.Vertex(nodeKeyDesc.PubKey.SerializeCompressed())
        //nolint:ll
        s.localChanMgr = &localchans.Manager{
                SelfPub:              nodeKeyDesc.PubKey,
                DefaultRoutingPolicy: cc.RoutingPolicy,
                ForAllOutgoingChannels: func(cb func(*models.ChannelEdgeInfo,
                        *models.ChannelEdgePolicy) error) error {

                        return s.graphDB.ForEachNodeChannel(selfVertex,
                                func(_ kvdb.RTx, c *models.ChannelEdgeInfo,
                                        e *models.ChannelEdgePolicy,
                                        _ *models.ChannelEdgePolicy) error {

                                        // NOTE: The invoked callback here may
                                        // receive a nil channel policy.
                                        return cb(c, e)
                                },
                        )
                },
                PropagateChanPolicyUpdate: s.authGossiper.PropagateChanPolicyUpdate,
                UpdateForwardingPolicies:  s.htlcSwitch.UpdateForwardingPolicies,
                FetchChannel:              s.chanStateDB.FetchChannel,
                AddEdge: func(edge *models.ChannelEdgeInfo) error {
                        return s.graphBuilder.AddEdge(edge)
                },
        }

        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
        }

        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
        }

        aggregator := sweep.NewBudgetAggregator(
                cc.FeeEstimator, sweep.DefaultMaxInputsPerTx,
                s.implCfg.AuxSweeper,
        )

        s.txPublisher = sweep.NewTxPublisher(sweep.TxPublisherConfig{
                Signer:     cc.Wallet.Cfg.Signer,
                Wallet:     cc.Wallet,
                Estimator:  cc.FeeEstimator,
                Notifier:   cc.ChainNotifier,
                AuxSweeper: s.implCfg.AuxSweeper,
        })

        s.sweeper = sweep.New(&sweep.UtxoSweeperConfig{
                FeeEstimator: cc.FeeEstimator,
                GenSweepScript: newSweepPkScriptGen(
                        cc.Wallet, s.cfg.ActiveNetParams.Params,
                ),
                Signer:               cc.Wallet.Cfg.Signer,
                Wallet:               newSweeperWallet(cc.Wallet),
                Mempool:              cc.MempoolNotifier,
                Notifier:             cc.ChainNotifier,
                Store:                sweeperStore,
                MaxInputsPerTx:       sweep.DefaultMaxInputsPerTx,
                MaxFeeRate:           cfg.Sweeper.MaxFeeRate,
                Aggregator:           aggregator,
                Publisher:            s.txPublisher,
                NoDeadlineConfTarget: cfg.Sweeper.NoDeadlineConfTarget,
        })

        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,
                Budget:              s.cfg.Sweeper.Budget,
        })

        // 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 BreachArbitrator,
        contractBreaches := make(chan *contractcourt.ContractBreachEvent, 1)

        s.breachArbitrator = contractcourt.NewBreachArbitrator(
                &contractcourt.BreachConfig{
                        CloseLink: closeLink,
                        DB:        s.chanStateDB,
                        Estimator: s.cc.FeeEstimator,
                        GenSweepScript: newSweepPkScriptGen(
                                cc.Wallet, s.cfg.ActiveNetParams.Params,
                        ),
                        Notifier:           cc.ChainNotifier,
                        PublishTransaction: cc.Wallet.PublishTransaction,
                        ContractBreaches:   contractBreaches,
                        Signer:             cc.Wallet.Cfg.Signer,
                        Store: contractcourt.NewRetributionStore(
                                dbs.ChanStateDB,
                        ),
                        AuxSweeper: s.implCfg.AuxSweeper,
                },
        )

        //nolint:ll
        s.chainArb = contractcourt.NewChainArbitrator(contractcourt.ChainArbitratorConfig{
                ChainHash:              *s.cfg.ActiveNetParams.GenesisHash,
                IncomingBroadcastDelta: lncfg.DefaultIncomingBroadcastDelta,
                OutgoingBroadcastDelta: lncfg.DefaultOutgoingBroadcastDelta,
                NewSweepAddr: func() ([]byte, error) {
                        addr, err := newSweepPkScriptGen(
                                cc.Wallet, netParams,
                        )().Unpack()
                        if err != nil {
                                return nil, err
                        }

                        return addr.DeliveryAddress, nil
                },
                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 fn.Option[lnwallet.OutgoingHtlcResolution],
                        inHtlcRes fn.Option[lnwallet.IncomingHtlcResolution],
                        broadcastHeight uint32,
                        deadlineHeight fn.Option[int32]) error {

                        return s.utxoNursery.IncubateOutputs(
                                chanPoint, outHtlcRes, inHtlcRes,
                                broadcastHeight, deadlineHeight,
                        )
                },
                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 BreachArbitrator ACKing
                        // the event.
                        finalErr := make(chan error, 1)
                        processACK := func(brarErr error) {
                                if brarErr != nil {
                                        finalErr <- brarErr
                                        return
                                }

                                // If the BreachArbitrator 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
                        // BreachArbitrator.
                        select {
                        case contractBreaches <- event:
                        case <-s.quit:
                                return ErrServerShuttingDown
                        }

                        // We'll wait for a final error to be available from
                        // the BreachArbitrator.
                        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.breachArbitrator.SubscribeBreachComplete,
                PutFinalHtlcOutcome:           s.chanStateDB.PutOnchainFinalHtlcOutcome,
                HtlcNotifier:                  s.htlcNotifier,
                Budget:                        *s.cfg.Sweeper.Budget,

                // TODO(yy): remove this hack once PaymentCircuit is interfaced.
                QueryIncomingCircuit: func(
                        circuit models.CircuitKey) *models.CircuitKey {

                        // Get the circuit map.
                        circuits := s.htlcSwitch.CircuitLookup()

                        // Lookup the outgoing circuit.
                        pc := circuits.LookupOpenCircuit(circuit)
                        if pc == nil {
                                return nil
                        }

                        return &pc.Incoming
                },
                AuxLeafStore: implCfg.AuxLeafStore,
                AuxSigner:    implCfg.AuxSigner,
                AuxResolver:  implCfg.AuxContractResolver,
        }, dbs.ChanStateDB)

        // Select the configuration and funding parameters for Bitcoin.
        chainCfg := cfg.Bitcoin
        minRemoteDelay := funding.MinBtcRemoteDelay
        maxRemoteDelay := funding.MaxBtcRemoteDelay

        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) (
                *models.ChannelEdgePolicy, error) {

                info, e1, e2, err := s.graphDB.FetchChannelEdgesByID(
                        scid.ToUint64(),
                )
                if errors.Is(err, graphdb.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 *models.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
        }

        // For the reservationTimeout and the zombieSweeperInterval different
        // values are set in case we are in a dev environment so enhance test
        // capacilities.
        reservationTimeout := chanfunding.DefaultReservationTimeout
        zombieSweeperInterval := lncfg.DefaultZombieSweeperInterval

        // 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(),
                }

                reservationTimeout = cfg.Dev.GetReservationTimeout()
                zombieSweeperInterval = cfg.Dev.GetZombieSweeperInterval()

                srvrLog.Debugf("Using the dev config for the fundingMgr: %v, "+
                        "reservationTimeout=%v, zombieSweeperInterval=%v",
                        devCfg, reservationTimeout, zombieSweeperInterval)
        }

        //nolint:ll
        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.

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

                        // 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:         zombieSweeperInterval,
                ReservationTimeout:            reservationTimeout,
                MinChanSize:                   btcutil.Amount(cfg.MinChanSize),
                MaxChanSize:                   btcutil.Amount(cfg.MaxChanSize),
                MaxPendingChannels:            cfg.MaxPendingChannels,
                RejectPush:                    cfg.RejectPush,
                MaxLocalCSVDelay:              chainCfg.MaxLocalDelay,
                NotifyOpenChannelEvent:        s.notifyOpenChannelPeerEvent,
                OpenChannelPredicate:          chanPredicate,
                NotifyPendingOpenChannelEvent: s.notifyPendingOpenChannelPeerEvent,
                NotifyFundingTimeout:          s.notifyFundingTimeoutPeerEvent,
                EnableUpfrontShutdown:         cfg.EnableUpfrontShutdown,
                MaxAnchorsCommitFeeRate: chainfee.SatPerKVByte(
                        s.cfg.MaxCommitFeeRateAnchors * 1000).FeePerKWeight(),
                DeleteAliasEdge:      deleteAliasEdge,
                AliasManager:         s.aliasMgr,
                IsSweeperOutpoint:    s.sweeper.IsSweeperOutpoint,
                AuxFundingController: implCfg.AuxFundingController,
                AuxSigner:            implCfg.AuxSigner,
                AuxResolver:          implCfg.AuxContractResolver,
        })
        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:        s.addrSource,
        }
        backupFile := chanbackup.NewMultiFile(
                cfg.BackupFilePath, cfg.NoBackupArchive,
        )
        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,
                                implCfg.AuxLeafStore,
                                implCfg.AuxContractResolver,
                        )
                        if err != nil {
                                return nil, 0, err
                        }

                        return br, channel.ChanType, nil
                }

                fetchClosedChannel := s.chanStateDB.FetchClosedChannelForID

                // Copy the policy for legacy channels and set the blob flag
                // signalling support for anchor channels.
                anchorPolicy := policy
                anchorPolicy.BlobType |= blob.Type(blob.FlagAnchorChannel)

                // Copy the policy for legacy channels and set the blob flag
                // signalling support for taproot channels.
                taprootPolicy := policy
                taprootPolicy.TxPolicy.BlobType |= blob.Type(
                        blob.FlagTaprootChannel,
                )

                s.towerClientMgr, err = wtclient.NewManager(&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: func() ([]byte, error) {
                                addr, err := newSweepPkScriptGen(
                                        cc.Wallet, netParams,
                                )().Unpack()
                                if err != nil {
                                        return nil, err
                                }

                                return addr.DeliveryAddress, nil
                        },
                        SecretKeyRing:      s.cc.KeyRing,
                        Dial:               cfg.net.Dial,
                        AuthDial:           authDial,
                        DB:                 dbs.TowerClientDB,
                        ChainHash:          *s.cfg.ActiveNetParams.GenesisHash,
                        MinBackoff:         10 * time.Second,
                        MaxBackoff:         5 * time.Minute,
                        MaxTasksInMemQueue: cfg.WtClient.MaxTasksInMemQueue,
                }, policy, anchorPolicy, taprootPolicy)
                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, leaderElector)

        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.

                // RESOLVE: We are actually partially accepting inbound
                // connection requests when we call NewListener.
                listeners[i], err = brontide.NewListener(
                        nodeKeyECDH, listenAddr.String(),
                        s.peerAccessMan.checkIncomingConnBanScore,
                )
                if err != nil {
                        return nil, err
                }
        }

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

        // Finally, register the subsystems in blockbeat.
        s.registerBlockConsumers()

        return s, nil
}

// UpdateRoutingConfig is a callback function to update the routing config
// values in the main cfg.
func (s *server) UpdateRoutingConfig(cfg *routing.MissionControlConfig) {
        routerCfg := s.cfg.SubRPCServers.RouterRPC

        switch c := cfg.Estimator.Config().(type) {
        case routing.AprioriConfig:
                routerCfg.ProbabilityEstimatorType =
                        routing.AprioriEstimatorName

                targetCfg := routerCfg.AprioriConfig
                targetCfg.PenaltyHalfLife = c.PenaltyHalfLife
                targetCfg.Weight = c.AprioriWeight
                targetCfg.CapacityFraction = c.CapacityFraction
                targetCfg.HopProbability = c.AprioriHopProbability

        case routing.BimodalConfig:
                routerCfg.ProbabilityEstimatorType =
                        routing.BimodalEstimatorName

                targetCfg := routerCfg.BimodalConfig
                targetCfg.Scale = int64(c.BimodalScaleMsat)
                targetCfg.NodeWeight = c.BimodalNodeWeight
                targetCfg.DecayTime = c.BimodalDecayTime
        }

        routerCfg.MaxMcHistory = cfg.MaxMcHistory
}

// registerBlockConsumers registers the subsystems that consume block events.
// By calling `RegisterQueue`, a list of subsystems are registered in the
// blockbeat for block notifications. When a new block arrives, the subsystems
// in the same queue are notified sequentially, and different queues are
// notified concurrently.
//
// NOTE: To put a subsystem in a different queue, create a slice and pass it to
// a new `RegisterQueue` call.
func (s *server) registerBlockConsumers() {
        // In this queue, when a new block arrives, it will be received and
        // processed in this order: chainArb -> sweeper -> txPublisher.
        consumers := []chainio.Consumer{
                s.chainArb,
                s.sweeper,
                s.txPublisher,
        }
        s.blockbeatDispatcher.RegisterQueue(consumers)
}

// 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.ChannelUpdate1) (*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,
        leaderElector cluster.LeaderElector) {

        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 a leader elector, we add a health check to ensure we are
        // still the leader. During normal operation, we should always be the
        // leader, but there are circumstances where this may change, such as
        // when we lose network connectivity for long enough expiring out lease.
        if leaderElector != nil {
                leaderCheck := healthcheck.NewObservation(
                        "leader status",
                        func() error {
                                // Check if we are still the leader. Note that
                                // we don't need to use a timeout context here
                                // as the healthcheck observer will handle the
                                // timeout case for us.
                                timeoutCtx, cancel := context.WithTimeout(
                                        context.Background(),
                                        cfg.HealthChecks.LeaderCheck.Timeout,
                                )
                                defer cancel()

                                leader, err := leaderElector.IsLeader(
                                        timeoutCtx,
                                )
                                if err != nil {
                                        return fmt.Errorf("unable to check if "+
                                                "still leader: %v", err)
                                }

                                if !leader {
                                        srvrLog.Debug("Not the current leader")
                                        return fmt.Errorf("not the current " +
                                                "leader")
                                }

                                return nil
                        },
                        cfg.HealthChecks.LeaderCheck.Interval,
                        cfg.HealthChecks.LeaderCheck.Timeout,
                        cfg.HealthChecks.LeaderCheck.Backoff,
                        cfg.HealthChecks.LeaderCheck.Attempts,
                )

                checks = append(checks, leaderCheck)
        }

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

// startLowLevelServices starts the low-level services of the server. These
// services must be started successfully before running the main server. The
// services are,
// 1. the chain notifier.
//
// TODO(yy): identify and add more low-level services here.
func (s *server) startLowLevelServices() error {
        var startErr error

        cleanup := cleaner{}

        cleanup = cleanup.add(s.cc.ChainNotifier.Stop)
        if err := s.cc.ChainNotifier.Start(); err != nil {
                startErr = err
        }

        if startErr != nil {
                cleanup.run()
        }

        return startErr
}

// Start starts the main daemon server, all requested listeners, and any helper
// goroutines.
// NOTE: This function is safe for concurrent access.
//
//nolint:funlen
func (s *server) Start() error {
        // Get the current blockbeat.
        beat, err := s.getStartingBeat()
        if err != nil {
                return err
        }

        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() {
                cleanup = cleanup.add(s.customMessageServer.Stop)
                if err := s.customMessageServer.Start(); err != nil {
                        startErr = err
                        return
                }

                if s.hostAnn != nil {
                        cleanup = cleanup.add(s.hostAnn.Stop)
                        if err := s.hostAnn.Start(); err != nil {
                                startErr = err
                                return
                        }
                }

                if s.livenessMonitor != nil {
                        cleanup = cleanup.add(s.livenessMonitor.Stop)
                        if err := s.livenessMonitor.Start(); err != nil {
                                startErr = err
                                return
                        }
                }

                // 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.
                cleanup = cleanup.add(s.sigPool.Stop)
                if err := s.sigPool.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.writePool.Stop)
                if err := s.writePool.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.readPool.Stop)
                if err := s.readPool.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.cc.BestBlockTracker.Stop)
                if err := s.cc.BestBlockTracker.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.channelNotifier.Stop)
                if err := s.channelNotifier.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(func() error {
                        return s.peerNotifier.Stop()
                })
                if err := s.peerNotifier.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.htlcNotifier.Stop)
                if err := s.htlcNotifier.Start(); err != nil {
                        startErr = err
                        return
                }

                if s.towerClientMgr != nil {
                        cleanup = cleanup.add(s.towerClientMgr.Stop)
                        if err := s.towerClientMgr.Start(); err != nil {
                                startErr = err
                                return
                        }
                }

                cleanup = cleanup.add(s.txPublisher.Stop)
                if err := s.txPublisher.Start(beat); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.sweeper.Stop)
                if err := s.sweeper.Start(beat); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.utxoNursery.Stop)
                if err := s.utxoNursery.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.breachArbitrator.Stop)
                if err := s.breachArbitrator.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.fundingMgr.Stop)
                if err := s.fundingMgr.Start(); err != nil {
                        startErr = err
                        return
                }

                // htlcSwitch must be started before chainArb since the latter
                // relies on htlcSwitch to deliver resolution message upon
                // start.
                cleanup = cleanup.add(s.htlcSwitch.Stop)
                if err := s.htlcSwitch.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.interceptableSwitch.Stop)
                if err := s.interceptableSwitch.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.invoiceHtlcModifier.Stop)
                if err := s.invoiceHtlcModifier.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.chainArb.Stop)
                if err := s.chainArb.Start(beat); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.graphBuilder.Stop)
                if err := s.graphBuilder.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.chanRouter.Stop)
                if err := s.chanRouter.Start(); err != nil {
                        startErr = err
                        return
                }
                // The authGossiper depends on the chanRouter and therefore
                // should be started after it.
                cleanup = cleanup.add(s.authGossiper.Stop)
                if err := s.authGossiper.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.invoices.Stop)
                if err := s.invoices.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.sphinx.Stop)
                if err := s.sphinx.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.chanStatusMgr.Stop)
                if err := s.chanStatusMgr.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup = cleanup.add(s.chanEventStore.Stop)
                if err := s.chanEventStore.Start(); err != nil {
                        startErr = err
                        return
                }

                cleanup.add(func() error {
                        s.missionController.StopStoreTickers()
                        return nil
                })
                s.missionController.RunStoreTickers()

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

                // chanSubSwapper must be started after the `channelNotifier`
                // because it depends on channel events as a synchronization
                // point.
                cleanup = cleanup.add(s.chanSubSwapper.Stop)
                if err := s.chanSubSwapper.Start(); err != nil {
                        startErr = err
                        return
                }

                if s.torController != nil {
                        cleanup = cleanup.add(s.torController.Stop)
                        if err := s.createNewHiddenService(); err != nil {
                                startErr = err
                                return
                        }
                }

                if s.natTraversal != nil {
                        s.wg.Add(1)
                        go s.watchExternalIP()
                }

                // Start connmgr last to prevent connections before init.
                cleanup = cleanup.add(func() error {
                        s.connMgr.Stop()
                        return nil
                })

                // RESOLVE: s.connMgr.Start() is called here, but
                // brontide.NewListener() is called in newServer. This means
                // that we are actually listening and partially accepting
                // inbound connections even before the connMgr starts.
                s.connMgr.Start()

                // 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/signet.
                if s.cfg.Bitcoin.MainNet {
                        setSeedList(
                                s.cfg.Bitcoin.DNSSeeds,
                                chainreg.BitcoinMainnetGenesis,
                        )
                }
                if s.cfg.Bitcoin.TestNet3 {
                        setSeedList(
                                s.cfg.Bitcoin.DNSSeeds,
                                chainreg.BitcoinTestnetGenesis,
                        )
                }
                if s.cfg.Bitcoin.SigNet {
                        setSeedList(
                                s.cfg.Bitcoin.DNSSeeds,
                                chainreg.BitcoinSignetGenesis,
                        )
                }

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

                // Start the blockbeat after all other subsystems have been
                // started so they are ready to receive new blocks.
                cleanup = cleanup.add(func() error {
                        s.blockbeatDispatcher.Stop()
                        return nil
                })
                if err := s.blockbeatDispatcher.Start(); err != nil {
                        startErr = err
                        return
                }

                // 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()

                // Stop dispatching blocks to other systems immediately.
                s.blockbeatDispatcher.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.interceptableSwitch.Stop(); err != nil {
                        srvrLog.Warnf("failed to stop interceptable "+
                                "switch: %v", err)
                }
                if err := s.invoiceHtlcModifier.Stop(); err != nil {
                        srvrLog.Warnf("failed to stop htlc invoices "+
                                "modifier: %v", err)
                }
                if err := s.chanRouter.Stop(); err != nil {
                        srvrLog.Warnf("failed to stop chanRouter: %v", err)
                }
                if err := s.graphBuilder.Stop(); err != nil {
                        srvrLog.Warnf("failed to stop graphBuilder %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.breachArbitrator.Stop(); err != nil {
                        srvrLog.Warnf("failed to stop breachArbitrator: %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.txPublisher.Stop(); err != nil {
                        srvrLog.Warnf("failed to stop txPublisher: %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)
                }

                // Update channel.backup file. Make sure to do it before
                // stopping chanSubSwapper.
                singles, err := chanbackup.FetchStaticChanBackups(
                        s.chanStateDB, s.addrSource,
                )
                if err != nil {
                        srvrLog.Warnf("failed to fetch channel states: %v",
                                err)
                } else {
                        err := s.chanSubSwapper.ManualUpdate(singles)
                        if err != nil {
                                srvrLog.Warnf("Manual update of channel "+
                                        "backup failed: %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)
                }
                if err := s.cc.BestBlockTracker.Stop(); err != nil {
                        srvrLog.Warnf("Unable to stop BestBlockTracker: %v",
                                err)
                }
                if err := s.chanEventStore.Stop(); err != nil {
                        srvrLog.Warnf("Unable to stop ChannelEventStore: %v",
                                err)
                }
                s.missionController.StopStoreTickers()

                // 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.towerClientMgr != nil {
                        if err := s.towerClientMgr.Stop(); err != nil {
                                srvrLog.Warnf("Unable to shut down tower "+
                                        "client manager: %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.
                srvrLog.Debug("Waiting for server to shutdown...")
                s.wg.Wait()

                srvrLog.Debug("Stopping buffer pools...")
                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 {
                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 := &models.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: %w", 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
}

// updateAndBroadcastSelfNode generates a new node announcement
// applying the giving modifiers and updating the time stamp
// to ensure it propagates through the network. Then it broadcasts
// it to the network.
func (s *server) updateAndBroadcastSelfNode(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: %w", 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: %w", 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 = s.graphDB.ForEachNodeChannel(sourceNode.PubKeyBytes, func(
                tx kvdb.RTx,
                chanInfo *models.ChannelEdgeInfo,
                policy, _ *models.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 := s.graphDB.FetchOtherNode(
                        tx, chanInfo, 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 && !errors.Is(err, graphdb.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()
}

// bannedPersistentPeerConnection does not actually "ban" a persistent peer. It
// is instead used to remove persistent peer state for a peer that has been
// disconnected for good cause by the server. Currently, a gossip ban from
// sending garbage and the server running out of restricted-access
// (i.e. "free") connection slots are the only way this logic gets hit. In the
// future, this function may expand when more ban criteria is added.
//
// NOTE: The server's write lock MUST be held when this is called.
func (s *server) bannedPersistentPeerConnection(remotePub string) {
        if perm, ok := s.persistentPeers[remotePub]; ok && !perm {
                delete(s.persistentPeers, remotePub)
                delete(s.persistentPeersBackoff, remotePub)
                delete(s.persistentPeerAddrs, remotePub)
                s.cancelConnReqs(remotePub, nil)
        }
}

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

// shouldDropLocalConnection 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()
        pubSer := nodePub.SerializeCompressed()
        pubStr := string(pubSer)

        var pubBytes [33]byte
        copy(pubBytes[:], pubSer)

        s.mu.Lock()
        defer s.mu.Unlock()

        // If the remote node's public key is banned, drop the connection.
        access, err := s.peerAccessMan.assignPeerPerms(nodePub)
        if err != nil {
                // Clean up the persistent peer maps if we're dropping this
                // connection.
                s.bannedPersistentPeerConnection(pubStr)

                srvrLog.Debugf("Dropping connection for %x since we are out "+
                        "of restricted-access connection slots: %v.", pubSer,
                        err)

                conn.Close()

                return
        }

        // 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, access)

        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, access)
                }
        }
}

// 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()
        pubSer := nodePub.SerializeCompressed()
        pubStr := string(pubSer)

        var pubBytes [33]byte
        copy(pubBytes[:], pubSer)

        s.mu.Lock()
        defer s.mu.Unlock()

        access, err := s.peerAccessMan.assignPeerPerms(nodePub)
        if err != nil {
                // Clean up the persistent peer maps if we're dropping this
                // connection.
                s.bannedPersistentPeerConnection(pubStr)

                srvrLog.Debugf("Dropping connection for %x since we are out "+
                        "of restricted-access connection slots: %v.", pubSer,
                        err)

                if connReq != nil {
                        s.connMgr.Remove(connReq.ID())
                }

                conn.Close()

                return
        }

        // 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, access)

        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, access)
                }
        }
}

// 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()
}

// notifyOpenChannelPeerEvent updates the access manager's maps and then calls
// the channelNotifier's NotifyOpenChannelEvent.
func (s *server) notifyOpenChannelPeerEvent(op wire.OutPoint,
        remotePub *btcec.PublicKey) error {

        // Call newOpenChan to update the access manager's maps for this peer.
        if err := s.peerAccessMan.newOpenChan(remotePub); err != nil {
                return err
        }

        // Notify subscribers about this open channel event.
        s.channelNotifier.NotifyOpenChannelEvent(op)

        return nil
}

// notifyPendingOpenChannelPeerEvent updates the access manager's maps and then
// calls the channelNotifier's NotifyPendingOpenChannelEvent.
func (s *server) notifyPendingOpenChannelPeerEvent(op wire.OutPoint,
        pendingChan *channeldb.OpenChannel, remotePub *btcec.PublicKey) error {

        // Call newPendingOpenChan to update the access manager's maps for this
        // peer.
        if err := s.peerAccessMan.newPendingOpenChan(remotePub); err != nil {
                return err
        }

        // Notify subscribers about this event.
        s.channelNotifier.NotifyPendingOpenChannelEvent(op, pendingChan)

        return nil
}

// notifyFundingTimeoutPeerEvent updates the access manager's maps and then
// calls the channelNotifier's NotifyFundingTimeout.
func (s *server) notifyFundingTimeoutPeerEvent(op wire.OutPoint,
        remotePub *btcec.PublicKey) error {

        // Call newPendingCloseChan to potentially demote the peer.
        err := s.peerAccessMan.newPendingCloseChan(remotePub)
        if err == ErrNoMoreRestrictedAccessSlots {
                s.DisconnectPeer(remotePub)
        }

        // Notify subscribers about this event.
        s.channelNotifier.NotifyFundingTimeout(op)

        return nil
}

// 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, access peerAccessStatus,
) {
        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
                }
        }

        // If we directly set the peer.Config TowerClient member to the
        // s.towerClientMgr then in the case that the s.towerClientMgr is nil,
        // the peer.Config's TowerClient member will not evaluate to nil even
        // though the underlying value is nil. To avoid this gotcha which can
        // cause a panic, we need to explicitly pass nil to the peer.Config's
        // TowerClient if needed.
        var towerClient wtclient.ClientManager
        if s.towerClientMgr != nil {
                towerClient = s.towerClientMgr
        }

        thresholdSats := btcutil.Amount(s.cfg.MaxFeeExposure)
        thresholdMSats := lnwire.NewMSatFromSatoshis(thresholdSats)

        // 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,
                BestBlockView:           s.cc.BestBlockTracker,
                RoutingPolicy:           s.cc.RoutingPolicy,
                Sphinx:                  s.sphinx,
                WitnessBeacon:           s.witnessBeacon,
                Invoices:                s.invoices,
                ChannelNotifier:         s.channelNotifier,
                HtlcNotifier:            s.htlcNotifier,
                TowerClient:             towerClient,
                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,
                DisallowRouteBlinding:  s.cfg.ProtocolOptions.NoRouteBlinding(),
                DisallowQuiescence:     s.cfg.ProtocolOptions.NoQuiescence(),
                MaxFeeExposure:         thresholdMSats,
                Quit:                   s.quit,
                AuxLeafStore:           s.implCfg.AuxLeafStore,
                AuxSigner:              s.implCfg.AuxSigner,
                MsgRouter:              s.implCfg.MsgRouter,
                AuxChanCloser:          s.implCfg.AuxChanCloser,
                AuxResolver:            s.implCfg.AuxContractResolver,
                AuxTrafficShaper:       s.implCfg.TrafficShaper,
                ShouldFwdExpEndorsement: func() bool {
                        if s.cfg.ProtocolOptions.NoExperimentalEndorsement() {
                                return false
                        }

                        return clock.NewDefaultClock().Now().Before(
                                EndorsementExperimentEnd,
                        )
                },
        }

        copy(pCfg.PubKeyBytes[:], peerAddr.IdentityKey.SerializeCompressed())
        copy(pCfg.ServerPubKey[:], s.identityECDH.PubKey().SerializeCompressed())

        p := peer.NewBrontide(pCfg)

        // Update the access manager with the access permission for this peer.
        s.peerAccessMan.addPeerAccess(pubKey, access)

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

        pubBytes := p.IdentityKey().SerializeCompressed()

        // Ignore new peers if we're shutting down.
        if s.Stopped() {
                srvrLog.Infof("Server stopped, skipped adding peer=%x",
                        pubBytes)
                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

        // NOTE: This pubStr is a raw bytes to string conversion and will NOT
        // be human-readable.
        pubStr := string(pubBytes)

        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[:], pubBytes)

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

        pubBytes := p.IdentityKey().SerializeCompressed()

        // Avoid initializing peers while the server is exiting.
        if s.Stopped() {
                srvrLog.Infof("Server stopped, skipped initializing peer=%x",
                        pubBytes)
                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 {
                srvrLog.Warnf("Starting peer=%x got error: %v", pubBytes, err)

                p.Disconnect(fmt.Errorf("unable to start peer: %w", err))
                return
        }

        // Otherwise, signal to the peerTerminationWatcher that the peer startup
        // was successful, and to begin watching the peer's wait group.
        close(ready)

        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)

        // TODO(guggero): Do a proper conversion to a string everywhere, or use
        // route.Vertex as the key type of peerConnectedListeners.
        pubStr := string(pubBytes)
        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)
        }

        // Remove the peer's access permission from the access manager.
        s.peerAccessMan.removePeerAccess(p.IdentityKey())

        // 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 at this point we fail the funding
        // because of the missing fee rate information. The caller of the
        // `OpenChannel` method needs to make sure that default values for the
        // fee rate are set beforehand.
        if req.FundingFeePerKw == 0 {
                req.Err <- fmt.Errorf("no FundingFeePerKw specified for " +
                        "the channel opening transaction")

                return req.Updates, req.Err
        }

        // 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.ChannelUpdate1, error) {

        ourPubKey := s.identityECDH.PubKey().SerializeCompressed()
        return func(cid lnwire.ShortChannelID) (*lnwire.ChannelUpdate1, error) {
                info, edge1, edge2, err := s.graphBuilder.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.ChannelUpdate1,
        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,
        netParams *chaincfg.Params) func() fn.Result[lnwallet.AddrWithKey] {

        return func() fn.Result[lnwallet.AddrWithKey] {
                sweepAddr, err := wallet.NewAddress(
                        lnwallet.TaprootPubkey, false,
                        lnwallet.DefaultAccountName,
                )
                if err != nil {
                        return fn.Err[lnwallet.AddrWithKey](err)
                }

                addr, err := txscript.PayToAddrScript(sweepAddr)
                if err != nil {
                        return fn.Err[lnwallet.AddrWithKey](err)
                }

                internalKeyDesc, err := lnwallet.InternalKeyForAddr(
                        wallet, netParams, addr,
                )
                if err != nil {
                        return fn.Err[lnwallet.AddrWithKey](err)
                }

                return fn.Ok(lnwallet.AddrWithKey{
                        DeliveryAddress: addr,
                        InternalKey:     internalKeyDesc,
                })
        }
}

// 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
        isSignet := cfg.Bitcoin.SigNet
        isRegtest := cfg.Bitcoin.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
}

// fetchClosedChannelSCIDs returns a set of SCIDs that have their force closing
// finished.
func (s *server) fetchClosedChannelSCIDs() map[lnwire.ShortChannelID]struct{} {
        // Get a list of closed channels.
        channels, err := s.chanStateDB.FetchClosedChannels(false)
        if err != nil {
                srvrLog.Errorf("Failed to fetch closed channels: %v", err)
                return nil
        }

        // Save the SCIDs in a map.
        closedSCIDs := make(map[lnwire.ShortChannelID]struct{}, len(channels))
        for _, c := range channels {
                // If the channel is not pending, its FC has been finalized.
                if !c.IsPending {
                        closedSCIDs[c.ShortChanID] = struct{}{}
                }
        }

        // Double check whether the reported closed channel has indeed finished
        // closing.
        //
        // NOTE: There are misalignments regarding when a channel's FC is
        // marked as finalized. We double check the pending channels to make
        // sure the returned SCIDs are indeed terminated.
        //
        // TODO(yy): fix the misalignments in `FetchClosedChannels`.
        pendings, err := s.chanStateDB.FetchPendingChannels()
        if err != nil {
                srvrLog.Errorf("Failed to fetch pending channels: %v", err)
                return nil
        }

        for _, c := range pendings {
                if _, ok := closedSCIDs[c.ShortChannelID]; !ok {
                        continue
                }

                // If the channel is still reported as pending, remove it from
                // the map.
                delete(closedSCIDs, c.ShortChannelID)

                srvrLog.Warnf("Channel=%v is prematurely marked as finalized",
                        c.ShortChannelID)
        }

        return closedSCIDs
}

// getStartingBeat returns the current beat. This is used during the startup to
// initialize blockbeat consumers.
func (s *server) getStartingBeat() (*chainio.Beat, error) {
        // beat is the current blockbeat.
        var beat *chainio.Beat

        // We should get a notification with the current best block immediately
        // by passing a nil block.
        blockEpochs, err := s.cc.ChainNotifier.RegisterBlockEpochNtfn(nil)
        if err != nil {
                return beat, fmt.Errorf("register block epoch ntfn: %w", err)
        }
        defer blockEpochs.Cancel()

        // We registered for the block epochs with a nil request. The notifier
        // should send us the current best block immediately. So we need to
        // wait for it here because we need to know the current best height.
        select {
        case bestBlock := <-blockEpochs.Epochs:
                srvrLog.Infof("Received initial block %v at height %d",
                        bestBlock.Hash, bestBlock.Height)

                // Update the current blockbeat.
                beat = chainio.NewBeat(*bestBlock)

        case <-s.quit:
                srvrLog.Debug("LND shutting down")
        }

        return beat, nil
}

lightningnetwork / lnd / 13593508312

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous