<commit_before>package main

import (
	"encoding/hex"
	"errors"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/btcsuite/fastsha256"
	"github.com/lightningnetwork/lightning-onion"
	"github.com/lightningnetwork/lnd/brontide"
	"github.com/lightningnetwork/lnd/chainntnfs"
	"github.com/lightningnetwork/lnd/channeldb"
	"github.com/lightningnetwork/lnd/lnrpc"
	"github.com/lightningnetwork/lnd/lnwallet"
	"github.com/lightningnetwork/lnd/lnwire"
	"github.com/roasbeef/btcd/btcec"
	"github.com/roasbeef/btcd/connmgr"
	"github.com/roasbeef/btcutil"

	"github.com/lightningnetwork/lnd/routing"
)

// server is the main server of the Lightning Network Daemon. The server houses
// global state pertaining to the wallet, database, and the rpcserver.
// Additionally, the server is also used as a central messaging bus to interact
// with any of its companion objects.
type server struct {
	started  int32 // atomic
	shutdown int32 // atomic

	// identityPriv is the private key used to authenticate any incoming
	// connections.
	identityPriv *btcec.PrivateKey

	// lightningID is the sha256 of the public key corresponding to our
	// long-term identity private key.
	lightningID [32]byte

	peersMtx   sync.RWMutex
	peersByID  map[int32]*peer
	peersByPub map[string]*peer

	rpcServer *rpcServer

	chainNotifier chainntnfs.ChainNotifier

	bio      lnwallet.BlockChainIO
	lnwallet *lnwallet.LightningWallet

	fundingMgr *fundingManager
	chanDB     *channeldb.DB

	htlcSwitch    *htlcSwitch
	invoices      *invoiceRegistry
	breachArbiter *breachArbiter

	chanRouter *routing.ChannelRouter

	utxoNursery *utxoNursery

	sphinx *sphinx.Router

	connMgr *connmgr.ConnManager

	pendingConnMtx     sync.RWMutex
	persistentConnReqs map[string]*connmgr.ConnReq

	broadcastRequests chan *broadcastReq
	sendRequests      chan *sendReq

	newPeers  chan *peer
	donePeers chan *peer
	queries   chan interface{}

	wg   sync.WaitGroup
	quit chan struct{}
}

// newServer creates a new instance of the server which is to listen using the
// passed listener address.
func newServer(listenAddrs []string, notifier chainntnfs.ChainNotifier,
	bio lnwallet.BlockChainIO, wallet *lnwallet.LightningWallet,
	chanDB *channeldb.DB) (*server, error) {

	privKey, err := wallet.GetIdentitykey()
	if err != nil {
		return nil, err
	}
	privKey.Curve = btcec.S256()

	listeners := make([]net.Listener, len(listenAddrs))
	for i, addr := range listenAddrs {
		listeners[i], err = brontide.NewListener(privKey, addr)
		if err != nil {
			return nil, err
		}
	}

	serializedPubKey := privKey.PubKey().SerializeCompressed()
	s := &server{
		lnwallet:      wallet,
		bio:           bio,
		chainNotifier: notifier,
		chanDB:        chanDB,

		invoices:    newInvoiceRegistry(chanDB),
		utxoNursery: newUtxoNursery(chanDB, notifier, wallet),
		htlcSwitch:  newHtlcSwitch(),

		identityPriv: privKey,

		// TODO(roasbeef): derive proper onion key based on rotation
		// schedule
		sphinx:      sphinx.NewRouter(privKey, activeNetParams.Params),
		lightningID: fastsha256.Sum256(serializedPubKey),

		persistentConnReqs: make(map[string]*connmgr.ConnReq),

		peersByID:  make(map[int32]*peer),
		peersByPub: make(map[string]*peer),

		newPeers:  make(chan *peer, 10),
		donePeers: make(chan *peer, 10),

		broadcastRequests: make(chan *broadcastReq),
		sendRequests:      make(chan *sendReq),

		queries: make(chan interface{}),
		quit:    make(chan struct{}),
	}

	// If the debug HTLC flag is on, then we invoice a "master debug"
	// invoice which all outgoing payments will be sent and all incoming
	// HTLCs with the debug R-Hash immediately settled.
	if cfg.DebugHTLC {
		kiloCoin := btcutil.Amount(btcutil.SatoshiPerBitcoin * 1000)
		s.invoices.AddDebugInvoice(kiloCoin, *debugPre)
		srvrLog.Debugf("Debug HTLC invoice inserted, preimage=%x, hash=%x",
			debugPre[:], debugHash[:])
	}

	// TODO(roasbeef): add --externalip flag?
	selfAddr, ok := listeners[0].Addr().(*net.TCPAddr)
	if !ok {
		return nil, fmt.Errorf("default listener must be TCP")
	}

	chanGraph := chanDB.ChannelGraph()
	self := &channeldb.LightningNode{
		LastUpdate: time.Now(),
		Address:    selfAddr,
		PubKey:     privKey.PubKey(),
		// TODO(roasbeef): make alias configurable
		Alias: hex.EncodeToString(serializedPubKey[:10]),
	}
	if err := chanGraph.SetSourceNode(self); err != nil {
		return nil, err
	}

	s.chanRouter, err = routing.New(routing.Config{
		Graph:        chanGraph,
		Chain:        bio,
		Notifier:     notifier,
		Broadcast:    s.broadcastMessage,
		SendMessages: s.sendToPeer,
	})
	if err != nil {
		return nil, err
	}

	s.rpcServer = newRpcServer(s)
	s.breachArbiter = newBreachArbiter(wallet, chanDB, notifier, s.htlcSwitch)
	s.fundingMgr = newFundingManager(wallet, s.breachArbiter)

	// TODO(roasbeef): introduce closure and config system to decouple the
	// initialization above ^

	// Create the connection manager which will be responsible for
	// maintaining persistent outbound connections and also accepting new
	// incoming connections
	cmgr, err := connmgr.New(&connmgr.Config{
		Listeners:      listeners,
		OnAccept:       s.inboundPeerConnected,
		RetryDuration:  time.Second * 5,
		TargetOutbound: 100,
		GetNewAddress:  nil,
		Dial:           noiseDial(s.identityPriv),
		OnConnection:   s.outboundPeerConnected,
	})
	if err != nil {
		return nil, err
	}
	s.connMgr = cmgr

	// In order to promote liveness of our active channels, instruct the
	// connection manager to attempt to establish and maintain persistent
	// connections to all our direct channel counterparties.
	linkNodes, err := s.chanDB.FetchAllLinkNodes()
	if err != nil && err != channeldb.ErrLinkNodesNotFound {
		return nil, err
	}
	for _, node := range linkNodes {
		// Create a wrapper address which couples the IP and the pubkey
		// so the brontide authenticated connection can be established.
		node.Addresses[0].Port = defaultPeerPort
		lnAddr := &lnwire.NetAddress{
			IdentityKey: node.IdentityPub,
			Address:     node.Addresses[0],
		}
		pubStr := string(node.IdentityPub.SerializeCompressed())
		srvrLog.Debugf("Attempting persistent connection to channel "+
			"peer %v", lnAddr)

		// Send the persistent connection request to the connection
		// manager, saving the request itself so we can cancel/restart
		// the process as needed.
		connReq := &connmgr.ConnReq{
			Addr:      lnAddr,
			Permanent: true,
		}
		s.persistentConnReqs[pubStr] = connReq
		go s.connMgr.Connect(connReq)
	}

	return s, nil
}

// Start starts the main daemon server, all requested listeners, and any helper
// goroutines.
func (s *server) Start() error {
	// Already running?
	if atomic.AddInt32(&s.started, 1) != 1 {
		return nil
	}
<commit_msg>lnd: revert usage of default port in connmgr persistent connections<commit_after>package main

import (
	"encoding/hex"
	"errors"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/btcsuite/fastsha256"
	"github.com/lightningnetwork/lightning-onion"
	"github.com/lightningnetwork/lnd/brontide"
	"github.com/lightningnetwork/lnd/chainntnfs"
	"github.com/lightningnetwork/lnd/channeldb"
	"github.com/lightningnetwork/lnd/lnrpc"
	"github.com/lightningnetwork/lnd/lnwallet"
	"github.com/lightningnetwork/lnd/lnwire"
	"github.com/roasbeef/btcd/btcec"
	"github.com/roasbeef/btcd/connmgr"
	"github.com/roasbeef/btcutil"

	"github.com/lightningnetwork/lnd/routing"
)

// server is the main server of the Lightning Network Daemon. The server houses
// global state pertaining to the wallet, database, and the rpcserver.
// Additionally, the server is also used as a central messaging bus to interact
// with any of its companion objects.
type server struct {
	started  int32 // atomic
	shutdown int32 // atomic

	// identityPriv is the private key used to authenticate any incoming
	// connections.
	identityPriv *btcec.PrivateKey

	// lightningID is the sha256 of the public key corresponding to our
	// long-term identity private key.
	lightningID [32]byte

	peersMtx   sync.RWMutex
	peersByID  map[int32]*peer
	peersByPub map[string]*peer

	rpcServer *rpcServer

	chainNotifier chainntnfs.ChainNotifier

	bio      lnwallet.BlockChainIO
	lnwallet *lnwallet.LightningWallet

	fundingMgr *fundingManager
	chanDB     *channeldb.DB

	htlcSwitch    *htlcSwitch
	invoices      *invoiceRegistry
	breachArbiter *breachArbiter

	chanRouter *routing.ChannelRouter

	utxoNursery *utxoNursery

	sphinx *sphinx.Router

	connMgr *connmgr.ConnManager

	pendingConnMtx     sync.RWMutex
	persistentConnReqs map[string]*connmgr.ConnReq

	broadcastRequests chan *broadcastReq
	sendRequests      chan *sendReq

	newPeers  chan *peer
	donePeers chan *peer
	queries   chan interface{}

	wg   sync.WaitGroup
	quit chan struct{}
}

// newServer creates a new instance of the server which is to listen using the
// passed listener address.
func newServer(listenAddrs []string, notifier chainntnfs.ChainNotifier,
	bio lnwallet.BlockChainIO, wallet *lnwallet.LightningWallet,
	chanDB *channeldb.DB) (*server, error) {

	privKey, err := wallet.GetIdentitykey()
	if err != nil {
		return nil, err
	}
	privKey.Curve = btcec.S256()

	listeners := make([]net.Listener, len(listenAddrs))
	for i, addr := range listenAddrs {
		listeners[i], err = brontide.NewListener(privKey, addr)
		if err != nil {
			return nil, err
		}
	}

	serializedPubKey := privKey.PubKey().SerializeCompressed()
	s := &server{
		lnwallet:      wallet,
		bio:           bio,
		chainNotifier: notifier,
		chanDB:        chanDB,

		invoices:    newInvoiceRegistry(chanDB),
		utxoNursery: newUtxoNursery(chanDB, notifier, wallet),
		htlcSwitch:  newHtlcSwitch(),

		identityPriv: privKey,

		// TODO(roasbeef): derive proper onion key based on rotation
		// schedule
		sphinx:      sphinx.NewRouter(privKey, activeNetParams.Params),
		lightningID: fastsha256.Sum256(serializedPubKey),

		persistentConnReqs: make(map[string]*connmgr.ConnReq),

		peersByID:  make(map[int32]*peer),
		peersByPub: make(map[string]*peer),

		newPeers:  make(chan *peer, 10),
		donePeers: make(chan *peer, 10),

		broadcastRequests: make(chan *broadcastReq),
		sendRequests:      make(chan *sendReq),

		queries: make(chan interface{}),
		quit:    make(chan struct{}),
	}

	// If the debug HTLC flag is on, then we invoice a "master debug"
	// invoice which all outgoing payments will be sent and all incoming
	// HTLCs with the debug R-Hash immediately settled.
	if cfg.DebugHTLC {
		kiloCoin := btcutil.Amount(btcutil.SatoshiPerBitcoin * 1000)
		s.invoices.AddDebugInvoice(kiloCoin, *debugPre)
		srvrLog.Debugf("Debug HTLC invoice inserted, preimage=%x, hash=%x",
			debugPre[:], debugHash[:])
	}

	// TODO(roasbeef): add --externalip flag?
	selfAddr, ok := listeners[0].Addr().(*net.TCPAddr)
	if !ok {
		return nil, fmt.Errorf("default listener must be TCP")
	}

	chanGraph := chanDB.ChannelGraph()
	self := &channeldb.LightningNode{
		LastUpdate: time.Now(),
		Address:    selfAddr,
		PubKey:     privKey.PubKey(),
		// TODO(roasbeef): make alias configurable
		Alias: hex.EncodeToString(serializedPubKey[:10]),
	}
	if err := chanGraph.SetSourceNode(self); err != nil {
		return nil, err
	}

	s.chanRouter, err = routing.New(routing.Config{
		Graph:        chanGraph,
		Chain:        bio,
		Notifier:     notifier,
		Broadcast:    s.broadcastMessage,
		SendMessages: s.sendToPeer,
	})
	if err != nil {
		return nil, err
	}

	s.rpcServer = newRpcServer(s)
	s.breachArbiter = newBreachArbiter(wallet, chanDB, notifier, s.htlcSwitch)
	s.fundingMgr = newFundingManager(wallet, s.breachArbiter)

	// TODO(roasbeef): introduce closure and config system to decouple the
	// initialization above ^

	// Create the connection manager which will be responsible for
	// maintaining persistent outbound connections and also accepting new
	// incoming connections
	cmgr, err := connmgr.New(&connmgr.Config{
		Listeners:      listeners,
		OnAccept:       s.inboundPeerConnected,
		RetryDuration:  time.Second * 5,
		TargetOutbound: 100,
		GetNewAddress:  nil,
		Dial:           noiseDial(s.identityPriv),
		OnConnection:   s.outboundPeerConnected,
	})
	if err != nil {
		return nil, err
	}
	s.connMgr = cmgr

	// In order to promote liveness of our active channels, instruct the
	// connection manager to attempt to establish and maintain persistent
	// connections to all our direct channel counterparties.
	linkNodes, err := s.chanDB.FetchAllLinkNodes()
	if err != nil && err != channeldb.ErrLinkNodesNotFound {
		return nil, err
	}
	for _, node := range linkNodes {
		// Create a wrapper address which couples the IP and the pubkey
		// so the brontide authenticated connection can be established.
		lnAddr := &lnwire.NetAddress{
			IdentityKey: node.IdentityPub,
			Address:     node.Addresses[0],
		}
		pubStr := string(node.IdentityPub.SerializeCompressed())
		srvrLog.Debugf("Attempting persistent connection to channel "+
			"peer %v", lnAddr)

		// Send the persistent connection request to the connection
		// manager, saving the request itself so we can cancel/restart
		// the process as needed.
		connReq := &connmgr.ConnReq{
			Addr:      lnAddr,
			Permanent: true,
		}
		s.persistentConnReqs[pubStr] = connReq
		go s.connMgr.Connect(connReq)
	}

	return s, nil
}

// Start starts the main daemon server, all requested listeners, and any helper
// goroutines.
func (s *server) Start() error {
	// Already running?
	if atomic.AddInt32(&s.started, 1) != 1 {
		return nil
	}

{'fraction_non_alphanumeric': 0.10301263362487852, 'fraction_numerical': 0.006802721088435374, 'mean_word_length': 3.9529803776379118, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6285084', 'n_tokens_mistral': 4795, 'n_tokens_neox': 4487, 'n_words': 1187}