lnd.xprv/discovery/gossiper.go
Olaoluwa Osuntokun b3ac3492a0
discovery: if unable to find gossipSyncer for peer, create one
In this commit we fix an existing bug caused by a scheduling race
condition. We'll now ensure that if we get a gossip message from a peer
before we create an instance for it, then we create one on the spot so
we can service the message. Before this commit, we would drop the first
message, and therefore never sync up with the peer at all, causing them
to miss channel announcements.
2018-05-31 17:54:45 -07:00

2519 lines
80 KiB
Go

package discovery
import (
"bytes"
"encoding/binary"
"fmt"
"runtime"
"sync"
"sync/atomic"
"time"
"github.com/coreos/bbolt"
"github.com/davecgh/go-spew/spew"
"github.com/go-errors/errors"
"github.com/lightningnetwork/lnd/chainntnfs"
"github.com/lightningnetwork/lnd/channeldb"
"github.com/lightningnetwork/lnd/lnwallet"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/multimutex"
"github.com/lightningnetwork/lnd/routing"
"github.com/roasbeef/btcd/btcec"
"github.com/roasbeef/btcd/chaincfg/chainhash"
"github.com/roasbeef/btcd/wire"
)
var (
// messageStoreKey is a key used to create a top level bucket in
// the gossiper database, used for storing messages that are to
// be sent to peers. Currently this is used for reliably sending
// AnnounceSignatures messages, by persisting them until a send
// operation has succeeded.
messageStoreKey = []byte("message-store")
)
// networkMsg couples a routing related wire message with the peer that
// originally sent it.
type networkMsg struct {
peer *btcec.PublicKey
msg lnwire.Message
isRemote bool
err chan error
}
// chanPolicyUpdateRequest is a request that is sent to the server when a caller
// wishes to update the channel policy (fees e.g.) for a particular set of
// channels. New ChannelUpdate messages will be crafted to be sent out during
// the next broadcast epoch and the fee updates committed to the lower layer.
type chanPolicyUpdateRequest struct {
targetChans []wire.OutPoint
newSchema routing.ChannelPolicy
errResp chan error
}
// Config defines the configuration for the service. ALL elements within the
// configuration MUST be non-nil for the service to carry out its duties.
type Config struct {
// ChainHash is a hash that indicates which resident chain of the
// AuthenticatedGossiper. Any announcements that don't match this
// chain hash will be ignored.
//
// TODO(roasbeef): eventually make into map so can de-multiplex
// incoming announcements
// * also need to do same for Notifier
ChainHash chainhash.Hash
// Router is the subsystem which is responsible for managing the
// topology of lightning network. After incoming channel, node, channel
// updates announcements are validated they are sent to the router in
// order to be included in the LN graph.
Router routing.ChannelGraphSource
// ChanSeries is an interfaces that provides access to a time series
// view of the current known channel graph. Each gossipSyncer enabled
// peer will utilize this in order to create and respond to channel
// graph time series queries.
ChanSeries ChannelGraphTimeSeries
// Notifier is used for receiving notifications of incoming blocks.
// With each new incoming block found we process previously premature
// announcements.
//
// TODO(roasbeef): could possibly just replace this with an epoch
// channel.
Notifier chainntnfs.ChainNotifier
// Broadcast broadcasts a particular set of announcements to all peers
// that the daemon is connected to. If supplied, the exclude parameter
// indicates that the target peer should be excluded from the
// broadcast.
Broadcast func(skips map[routing.Vertex]struct{},
msg ...lnwire.Message) error
// SendToPeer is a function which allows the service to send a set of
// messages to a particular peer identified by the target public key.
SendToPeer func(target *btcec.PublicKey, msg ...lnwire.Message) error
// NotifyWhenOnline is a function that allows the gossiper to be
// notified when a certain peer comes online, allowing it to
// retry sending a peer message.
NotifyWhenOnline func(peer *btcec.PublicKey, connectedChan chan<- struct{})
// ProofMatureDelta the number of confirmations which is needed before
// exchange the channel announcement proofs.
ProofMatureDelta uint32
// TrickleDelay the period of trickle timer which flushes to the
// network the pending batch of new announcements we've received since
// the last trickle tick.
TrickleDelay time.Duration
// RetransmitDelay is the period of a timer which indicates that we
// should check if we need re-broadcast any of our personal channels.
RetransmitDelay time.Duration
// DB is a global boltdb instance which is needed to pass it in waiting
// proof storage to make waiting proofs persistent.
DB *channeldb.DB
// AnnSigner is an instance of the MessageSigner interface which will
// be used to manually sign any outgoing channel updates. The signer
// implementation should be backed by the public key of the backing
// Lightning node.
//
// TODO(roasbeef): extract ann crafting + sign from fundingMgr into
// here?
AnnSigner lnwallet.MessageSigner
}
// AuthenticatedGossiper is a subsystem which is responsible for receiving
// announcements, validating them and applying the changes to router, syncing
// lightning network with newly connected nodes, broadcasting announcements
// after validation, negotiating the channel announcement proofs exchange and
// handling the premature announcements. All outgoing announcements are
// expected to be properly signed as dictated in BOLT#7, additionally, all
// incoming message are expected to be well formed and signed. Invalid messages
// will be rejected by this struct.
type AuthenticatedGossiper struct {
// Parameters which are needed to properly handle the start and stop of
// the service.
started uint32
stopped uint32
quit chan struct{}
wg sync.WaitGroup
// cfg is a copy of the configuration struct that the gossiper service
// was initialized with.
cfg *Config
// newBlocks is a channel in which new blocks connected to the end of
// the main chain are sent over.
newBlocks <-chan *chainntnfs.BlockEpoch
// prematureAnnouncements maps a block height to a set of network
// messages which are "premature" from our PoV. A message is premature
// if it claims to be anchored in a block which is beyond the current
// main chain tip as we know it. Premature network messages will be
// processed once the chain tip as we know it extends to/past the
// premature height.
//
// TODO(roasbeef): limit premature networkMsgs to N
prematureAnnouncements map[uint32][]*networkMsg
// prematureChannelUpdates is a map of ChannelUpdates we have received
// that wasn't associated with any channel we know about. We store
// them temporarily, such that we can reprocess them when a
// ChannelAnnouncement for the channel is received.
prematureChannelUpdates map[uint64][]*networkMsg
pChanUpdMtx sync.Mutex
// waitingProofs is a persistent storage of partial channel proof
// announcement messages. We use it to buffer half of the material
// needed to reconstruct a full authenticated channel announcement.
// Once we receive the other half the channel proof, we'll be able to
// properly validate it and re-broadcast it out to the network.
waitingProofs *channeldb.WaitingProofStore
// networkMsgs is a channel that carries new network broadcasted
// message from outside the gossiper service to be processed by the
// networkHandler.
networkMsgs chan *networkMsg
// chanPolicyUpdates is a channel that requests to update the
// forwarding policy of a set of channels is sent over.
chanPolicyUpdates chan *chanPolicyUpdateRequest
// bestHeight is the height of the block at the tip of the main chain
// as we know it.
bestHeight uint32
// selfKey is the identity public key of the backing Lightning node.
selfKey *btcec.PublicKey
// channelMtx is used to restrict the database access to one
// goroutine per channel ID. This is done to ensure that when
// the gossiper is handling an announcement, the db state stays
// consistent between when the DB is first read until it's written.
channelMtx *multimutex.Mutex
rejectMtx sync.RWMutex
recentRejects map[uint64]struct{}
// peerSyncers keeps track of all the gossip syncers we're maintain for
// peers that understand this mode of operation. When we go to send out
// new updates, for all peers in the map, we'll send the messages
// directly to their gossiper, rather than broadcasting them. With this
// change, we ensure we filter out all updates properly.
syncerMtx sync.RWMutex
peerSyncers map[routing.Vertex]*gossipSyncer
sync.Mutex
}
// New creates a new AuthenticatedGossiper instance, initialized with the
// passed configuration parameters.
func New(cfg Config, selfKey *btcec.PublicKey) (*AuthenticatedGossiper, error) {
storage, err := channeldb.NewWaitingProofStore(cfg.DB)
if err != nil {
return nil, err
}
return &AuthenticatedGossiper{
selfKey: selfKey,
cfg: &cfg,
networkMsgs: make(chan *networkMsg),
quit: make(chan struct{}),
chanPolicyUpdates: make(chan *chanPolicyUpdateRequest),
prematureAnnouncements: make(map[uint32][]*networkMsg),
prematureChannelUpdates: make(map[uint64][]*networkMsg),
waitingProofs: storage,
channelMtx: multimutex.NewMutex(),
recentRejects: make(map[uint64]struct{}),
peerSyncers: make(map[routing.Vertex]*gossipSyncer),
}, nil
}
// SynchronizeNode sends a message to the service indicating it should
// synchronize lightning topology state with the target node. This method is to
// be utilized when a node connections for the first time to provide it with
// the latest topology update state. In order to accomplish this, (currently)
// the entire network graph is read from disk, then serialized to the format
// defined within the current wire protocol. This cache of graph data is then
// sent directly to the target node.
func (d *AuthenticatedGossiper) SynchronizeNode(pub *btcec.PublicKey) error {
// TODO(roasbeef): need to also store sig data in db
// * will be nice when we switch to pairing sigs would only need one ^_^
// We'll collate all the gathered routing messages into a single slice
// containing all the messages to be sent to the target peer.
var announceMessages []lnwire.Message
makeNodeAnn := func(n *channeldb.LightningNode) (*lnwire.NodeAnnouncement, error) {
alias, _ := lnwire.NewNodeAlias(n.Alias)
wireSig, err := lnwire.NewSigFromRawSignature(n.AuthSigBytes)
if err != nil {
return nil, err
}
return &lnwire.NodeAnnouncement{
Signature: wireSig,
Timestamp: uint32(n.LastUpdate.Unix()),
Addresses: n.Addresses,
NodeID: n.PubKeyBytes,
Features: n.Features.RawFeatureVector,
RGBColor: n.Color,
Alias: alias,
}, nil
}
// We'll use this map to ensure we don't send the same node
// announcement more than one time as one node may have many channel
// anns we'll need to send.
nodePubsSent := make(map[routing.Vertex]struct{})
// As peers are expecting channel announcements before node
// announcements, we first retrieve the initial announcement, as well as
// the latest channel update announcement for both of the directed edges
// that make up each channel, and queue these to be sent to the peer.
var (
numEdges uint32
numNodes uint32
)
if err := d.cfg.Router.ForEachChannel(func(chanInfo *channeldb.ChannelEdgeInfo,
e1, e2 *channeldb.ChannelEdgePolicy) error {
// First, using the parameters of the channel, along with the
// channel authentication proof, we'll create re-create the
// original authenticated channel announcement. If the channel
// also has known validated nodes, then we'll send that as
// well.
if chanInfo.AuthProof != nil {
chanAnn, e1Ann, e2Ann, err := CreateChanAnnouncement(
chanInfo.AuthProof, chanInfo, e1, e2,
)
if err != nil {
return err
}
announceMessages = append(announceMessages, chanAnn)
if e1Ann != nil {
announceMessages = append(announceMessages, e1Ann)
// If this edge has a validated node
// announcement, that we haven't yet sent, then
// we'll send that as well.
nodePub := e1.Node.PubKeyBytes
hasNodeAnn := e1.Node.HaveNodeAnnouncement
if _, ok := nodePubsSent[nodePub]; !ok && hasNodeAnn {
nodeAnn, err := makeNodeAnn(e1.Node)
if err != nil {
return err
}
announceMessages = append(
announceMessages, nodeAnn,
)
nodePubsSent[nodePub] = struct{}{}
numNodes++
}
}
if e2Ann != nil {
announceMessages = append(announceMessages, e2Ann)
// If this edge has a validated node
// announcement, that we haven't yet sent, then
// we'll send that as well.
nodePub := e2.Node.PubKeyBytes
hasNodeAnn := e2.Node.HaveNodeAnnouncement
if _, ok := nodePubsSent[nodePub]; !ok && hasNodeAnn {
nodeAnn, err := makeNodeAnn(e2.Node)
if err != nil {
return err
}
announceMessages = append(
announceMessages, nodeAnn,
)
nodePubsSent[nodePub] = struct{}{}
numNodes++
}
}
numEdges++
}
return nil
}); err != nil && err != channeldb.ErrGraphNoEdgesFound {
log.Errorf("unable to sync infos with peer: %v", err)
return err
}
log.Infof("Syncing channel graph state with %x, sending %v "+
"vertexes and %v edges", pub.SerializeCompressed(),
numNodes, numEdges)
// With all the announcement messages gathered, send them all in a
// single batch to the target peer.
return d.cfg.SendToPeer(pub, announceMessages...)
}
// PropagateChanPolicyUpdate signals the AuthenticatedGossiper to update the
// channel forwarding policies for the specified channels. If no channels are
// specified, then the update will be applied to all outgoing channels from the
// source node. Policy updates are done in two stages: first, the
// AuthenticatedGossiper ensures the update has been committed by dependent
// sub-systems, then it signs and broadcasts new updates to the network.
func (d *AuthenticatedGossiper) PropagateChanPolicyUpdate(
newSchema routing.ChannelPolicy, chanPoints ...wire.OutPoint) error {
errChan := make(chan error, 1)
policyUpdate := &chanPolicyUpdateRequest{
targetChans: chanPoints,
newSchema: newSchema,
errResp: errChan,
}
select {
case d.chanPolicyUpdates <- policyUpdate:
return <-errChan
case <-d.quit:
return fmt.Errorf("AuthenticatedGossiper shutting down")
}
}
// Start spawns network messages handler goroutine and registers on new block
// notifications in order to properly handle the premature announcements.
func (d *AuthenticatedGossiper) Start() error {
if !atomic.CompareAndSwapUint32(&d.started, 0, 1) {
return nil
}
log.Info("Authenticated Gossiper is starting")
// First we register for new notifications of newly discovered blocks.
// We do this immediately so we'll later be able to consume any/all
// blocks which were discovered.
blockEpochs, err := d.cfg.Notifier.RegisterBlockEpochNtfn()
if err != nil {
return err
}
d.newBlocks = blockEpochs.Epochs
height, err := d.cfg.Router.CurrentBlockHeight()
if err != nil {
return err
}
d.bestHeight = height
// In case we had an AnnounceSignatures ready to be sent when the
// gossiper was last shut down, we must continue on our quest to
// deliver this message to our peer such that they can craft the
// full channel proof.
if err := d.resendAnnounceSignatures(); err != nil {
return err
}
d.wg.Add(1)
go d.networkHandler()
return nil
}
// Stop signals any active goroutines for a graceful closure.
func (d *AuthenticatedGossiper) Stop() {
if !atomic.CompareAndSwapUint32(&d.stopped, 0, 1) {
return
}
log.Info("Authenticated Gossiper is stopping")
d.syncerMtx.RLock()
for _, syncer := range d.peerSyncers {
syncer.Stop()
}
d.syncerMtx.RUnlock()
close(d.quit)
d.wg.Wait()
}
// TODO(roasbeef): need method to get current gossip timestamp?
// * using mtx, check time rotate forward is needed?
// ProcessRemoteAnnouncement sends a new remote announcement message along with
// the peer that sent the routing message. The announcement will be processed
// then added to a queue for batched trickled announcement to all connected
// peers. Remote channel announcements should contain the announcement proof
// and be fully validated.
func (d *AuthenticatedGossiper) ProcessRemoteAnnouncement(msg lnwire.Message,
src *btcec.PublicKey) chan error {
nMsg := &networkMsg{
msg: msg,
isRemote: true,
peer: src,
err: make(chan error, 1),
}
select {
case d.networkMsgs <- nMsg:
case <-d.quit:
nMsg.err <- errors.New("gossiper has shut down")
}
return nMsg.err
}
// ProcessLocalAnnouncement sends a new remote announcement message along with
// the peer that sent the routing message. The announcement will be processed
// then added to a queue for batched trickled announcement to all connected
// peers. Local channel announcements don't contain the announcement proof and
// will not be fully validated. Once the channel proofs are received, the
// entire channel announcement and update messages will be re-constructed and
// broadcast to the rest of the network.
func (d *AuthenticatedGossiper) ProcessLocalAnnouncement(msg lnwire.Message,
src *btcec.PublicKey) chan error {
nMsg := &networkMsg{
msg: msg,
isRemote: false,
peer: src,
err: make(chan error, 1),
}
select {
case d.networkMsgs <- nMsg:
case <-d.quit:
nMsg.err <- errors.New("gossiper has shut down")
}
return nMsg.err
}
// channelUpdateID is a unique identifier for ChannelUpdate messages, as
// channel updates can be identified by the (ShortChannelID, Flags)
// tuple.
type channelUpdateID struct {
// channelID represents the set of data which is needed to
// retrieve all necessary data to validate the channel existence.
channelID lnwire.ShortChannelID
// Flags least-significant bit must be set to 0 if the creating node
// corresponds to the first node in the previously sent channel
// announcement and 1 otherwise.
flags lnwire.ChanUpdateFlag
}
// msgWithSenders is a wrapper struct around a message, and the set of peers
// that originally sent us this message. Using this struct, we can ensure that
// we don't re-send a message to the peer that sent it to us in the first
// place.
type msgWithSenders struct {
// msg is the wire message itself.
msg lnwire.Message
// sender is the set of peers that sent us this message.
senders map[routing.Vertex]struct{}
}
// mergeSyncerMap is used to merge the set of senders of a particular message
// with peers that we have an active gossipSyncer with. We do this to ensure
// that we don't broadcast messages to any peers that we have active gossip
// syncers for.
func (m *msgWithSenders) mergeSyncerMap(syncers map[routing.Vertex]struct{}) {
for peerPub := range syncers {
m.senders[peerPub] = struct{}{}
}
}
// deDupedAnnouncements de-duplicates announcements that have been added to the
// batch. Internally, announcements are stored in three maps
// (one each for channel announcements, channel updates, and node
// announcements). These maps keep track of unique announcements and ensure no
// announcements are duplicated. We keep the three message types separate, such
// that we can send channel announcements first, then channel updates, and
// finally node announcements when it's time to broadcast them.
type deDupedAnnouncements struct {
// channelAnnouncements are identified by the short channel id field.
channelAnnouncements map[lnwire.ShortChannelID]msgWithSenders
// channelUpdates are identified by the channel update id field.
channelUpdates map[channelUpdateID]msgWithSenders
// nodeAnnouncements are identified by the Vertex field.
nodeAnnouncements map[routing.Vertex]msgWithSenders
sync.Mutex
}
// Reset operates on deDupedAnnouncements to reset the storage of
// announcements.
func (d *deDupedAnnouncements) Reset() {
d.Lock()
defer d.Unlock()
d.reset()
}
// reset is the private version of the Reset method. We have this so we can
// call this method within method that are already holding the lock.
func (d *deDupedAnnouncements) reset() {
// Storage of each type of announcement (channel announcements, channel
// updates, node announcements) is set to an empty map where the
// appropriate key points to the corresponding lnwire.Message.
d.channelAnnouncements = make(map[lnwire.ShortChannelID]msgWithSenders)
d.channelUpdates = make(map[channelUpdateID]msgWithSenders)
d.nodeAnnouncements = make(map[routing.Vertex]msgWithSenders)
}
// addMsg adds a new message to the current batch. If the message is already
// present in the current batch, then this new instance replaces the latter,
// and the set of senders is updated to reflect which node sent us this
// message.
func (d *deDupedAnnouncements) addMsg(message networkMsg) {
// Depending on the message type (channel announcement, channel update,
// or node announcement), the message is added to the corresponding map
// in deDupedAnnouncements. Because each identifying key can have at
// most one value, the announcements are de-duplicated, with newer ones
// replacing older ones.
switch msg := message.msg.(type) {
// Channel announcements are identified by the short channel id field.
case *lnwire.ChannelAnnouncement:
deDupKey := msg.ShortChannelID
sender := routing.NewVertex(message.peer)
mws, ok := d.channelAnnouncements[deDupKey]
if !ok {
mws = msgWithSenders{
msg: msg,
senders: make(map[routing.Vertex]struct{}),
}
mws.senders[sender] = struct{}{}
d.channelAnnouncements[deDupKey] = mws
return
}
mws.msg = msg
mws.senders[sender] = struct{}{}
d.channelAnnouncements[deDupKey] = mws
// Channel updates are identified by the (short channel id, flags)
// tuple.
case *lnwire.ChannelUpdate:
sender := routing.NewVertex(message.peer)
deDupKey := channelUpdateID{
msg.ShortChannelID,
msg.Flags,
}
oldTimestamp := uint32(0)
mws, ok := d.channelUpdates[deDupKey]
if ok {
// If we already have seen this message, record its
// timestamp.
oldTimestamp = mws.msg.(*lnwire.ChannelUpdate).Timestamp
}
// If we already had this message with a strictly newer
// timestamp, then we'll just discard the message we got.
if oldTimestamp > msg.Timestamp {
return
}
// If the message we just got is newer than what we previously
// have seen, or this is the first time we see it, then we'll
// add it to our map of announcements.
if oldTimestamp < msg.Timestamp {
mws = msgWithSenders{
msg: msg,
senders: make(map[routing.Vertex]struct{}),
}
// We'll mark the sender of the message in the
// senders map.
mws.senders[sender] = struct{}{}
d.channelUpdates[deDupKey] = mws
return
}
// Lastly, if we had seen this exact message from before, with
// the same timestamp, we'll add the sender to the map of
// senders, such that we can skip sending this message back in
// the next batch.
mws.msg = msg
mws.senders[sender] = struct{}{}
d.channelUpdates[deDupKey] = mws
// Node announcements are identified by the Vertex field. Use the
// NodeID to create the corresponding Vertex.
case *lnwire.NodeAnnouncement:
sender := routing.NewVertex(message.peer)
deDupKey := routing.Vertex(msg.NodeID)
// We do the same for node announcements as we did for channel
// updates, as they also carry a timestamp.
oldTimestamp := uint32(0)
mws, ok := d.nodeAnnouncements[deDupKey]
if ok {
oldTimestamp = mws.msg.(*lnwire.NodeAnnouncement).Timestamp
}
// Discard the message if it's old.
if oldTimestamp > msg.Timestamp {
return
}
// Replace if it's newer.
if oldTimestamp < msg.Timestamp {
mws = msgWithSenders{
msg: msg,
senders: make(map[routing.Vertex]struct{}),
}
mws.senders[sender] = struct{}{}
d.nodeAnnouncements[deDupKey] = mws
return
}
// Add to senders map if it's the same as we had.
mws.msg = msg
mws.senders[sender] = struct{}{}
d.nodeAnnouncements[deDupKey] = mws
}
}
// AddMsgs is a helper method to add multiple messages to the announcement
// batch.
func (d *deDupedAnnouncements) AddMsgs(msgs ...networkMsg) {
d.Lock()
defer d.Unlock()
for _, msg := range msgs {
d.addMsg(msg)
}
}
// Emit returns the set of de-duplicated announcements to be sent out during
// the next announcement epoch, in the order of channel announcements, channel
// updates, and node announcements. Each message emitted, contains the set of
// peers that sent us the message. This way, we can ensure that we don't waste
// bandwidth by re-sending a message to the peer that sent it to us in the
// first place. Additionally, the set of stored messages are reset.
func (d *deDupedAnnouncements) Emit() []msgWithSenders {
d.Lock()
defer d.Unlock()
// Get the total number of announcements.
numAnnouncements := len(d.channelAnnouncements) + len(d.channelUpdates) +
len(d.nodeAnnouncements)
// Create an empty array of lnwire.Messages with a length equal to
// the total number of announcements.
msgs := make([]msgWithSenders, 0, numAnnouncements)
// Add the channel announcements to the array first.
for _, message := range d.channelAnnouncements {
msgs = append(msgs, message)
}
// Then add the channel updates.
for _, message := range d.channelUpdates {
msgs = append(msgs, message)
}
// Finally add the node announcements.
for _, message := range d.nodeAnnouncements {
msgs = append(msgs, message)
}
d.reset()
// Return the array of lnwire.messages.
return msgs
}
// resendAnnounceSignatures will inspect the messageStore database bucket for
// AnnounceSignatures messages that we recently tried to send to a peer. If the
// associated channels still not have the full channel proofs assembled, we
// will try to resend them. If we have the full proof, we can safely delete the
// message from the messageStore.
func (d *AuthenticatedGossiper) resendAnnounceSignatures() error {
type msgTuple struct {
peer *btcec.PublicKey
msg *lnwire.AnnounceSignatures
dbKey []byte
}
// Fetch all the AnnounceSignatures messages that was added to the
// database.
//
// TODO(halseth): database access should be abstracted
// behind interface.
var msgsResend []msgTuple
if err := d.cfg.DB.View(func(tx *bolt.Tx) error {
bucket := tx.Bucket(messageStoreKey)
if bucket == nil {
return nil
}
if err := bucket.ForEach(func(k, v []byte) error {
// The database value represents the encoded
// AnnounceSignatures message.
r := bytes.NewReader(v)
msg := &lnwire.AnnounceSignatures{}
if err := msg.Decode(r, 0); err != nil {
return err
}
// The first 33 bytes of the database key is the peer's
// public key.
peer, err := btcec.ParsePubKey(k[:33], btcec.S256())
if err != nil {
return err
}
t := msgTuple{peer, msg, k}
// Add the message to the slice, such that we can
// resend it after the database transaction is over.
msgsResend = append(msgsResend, t)
return nil
}); err != nil {
return err
}
return nil
}); err != nil {
return err
}
// deleteMsg removes the message associated with the passed msgTuple
// from the messageStore.
deleteMsg := func(t msgTuple) error {
log.Debugf("Deleting message for chanID=%v from "+
"messageStore", t.msg.ChannelID)
if err := d.cfg.DB.Update(func(tx *bolt.Tx) error {
bucket := tx.Bucket(messageStoreKey)
if bucket == nil {
return fmt.Errorf("bucket " +
"unexpectedly did not exist")
}
return bucket.Delete(t.dbKey[:])
}); err != nil {
return fmt.Errorf("Failed deleting message "+
"from database: %v", err)
}
return nil
}
// We now iterate over these messages, resending those that we don't
// have the full proof for, deleting the rest.
for _, t := range msgsResend {
// Check if the full channel proof exists in our graph.
chanInfo, _, _, err := d.cfg.Router.GetChannelByID(
t.msg.ShortChannelID)
if err != nil {
// If the channel cannot be found, it is most likely a
// leftover message for a channel that was closed. In
// this case we delete it from the message store.
log.Warnf("unable to fetch channel info for "+
"chanID=%v from graph: %v. Will delete local"+
"proof from database",
t.msg.ChannelID, err)
if err := deleteMsg(t); err != nil {
return err
}
continue
}
// 1. If the full proof does not exist in the graph, it means
// that we haven't received the remote proof yet (or that we
// crashed before able to assemble the full proof). Since the
// remote node might think they have delivered their proof to
// us, we will resend _our_ proof to trigger a resend on their
// part: they will then be able to assemble and send us the
// full proof.
if chanInfo.AuthProof == nil {
err := d.sendAnnSigReliably(t.msg, t.peer)
if err != nil {
return err
}
continue
}
// 2. If the proof does exist in the graph, we have
// successfully received the remote proof and assembled the
// full proof. In this case we can safely delete the local
// proof from the database. In case the remote hasn't been able
// to assemble the full proof yet (maybe because of a crash),
// we will send them the full proof if we notice that they
// retry sending their half proof.
if chanInfo.AuthProof != nil {
log.Debugf("Deleting message for chanID=%v from "+
"messageStore", t.msg.ChannelID)
if err := deleteMsg(t); err != nil {
return err
}
}
}
return nil
}
// findGossipSyncer is a utility method used by the gossiper to locate the
// gossip syncer for an inbound message so we can properly dispatch the
// incoming message. If a gossip syncer isn't found, then one will be created
// for the target peer.
func (d *AuthenticatedGossiper) findGossipSyncer(pub *btcec.PublicKey) *gossipSyncer {
target := routing.NewVertex(pub)
// First, we'll try to find an existing gossiper for this peer.
d.syncerMtx.RLock()
syncer, ok := d.peerSyncers[target]
d.syncerMtx.RUnlock()
// If one exists, then we'll return it directly.
if ok {
return syncer
}
// Otherwise, we'll obtain the mutex, then check again if a gossiper
// was added after we dropped the read mutex.
d.syncerMtx.Lock()
syncer, ok = d.peerSyncers[target]
if ok {
d.syncerMtx.Unlock()
return syncer
}
// At this point, a syncer doesn't yet exist, so we'll create a new one
// for the peer and return it to the caller.
syncer = newGossiperSyncer(gossipSyncerCfg{
chainHash: d.cfg.ChainHash,
syncChanUpdates: true,
channelSeries: d.cfg.ChanSeries,
encodingType: lnwire.EncodingSortedPlain,
sendToPeer: func(msgs ...lnwire.Message) error {
return d.cfg.SendToPeer(pub, msgs...)
},
})
copy(syncer.peerPub[:], pub.SerializeCompressed())
d.peerSyncers[target] = syncer
syncer.Start()
d.syncerMtx.Unlock()
return syncer
}
// networkHandler is the primary goroutine that drives this service. The roles
// of this goroutine includes answering queries related to the state of the
// network, syncing up newly connected peers, and also periodically
// broadcasting our latest topology state to all connected peers.
//
// NOTE: This MUST be run as a goroutine.
func (d *AuthenticatedGossiper) networkHandler() {
defer d.wg.Done()
// Initialize empty deDupedAnnouncements to store announcement batch.
announcements := deDupedAnnouncements{}
announcements.Reset()
retransmitTimer := time.NewTicker(d.cfg.RetransmitDelay)
defer retransmitTimer.Stop()
trickleTimer := time.NewTicker(d.cfg.TrickleDelay)
defer trickleTimer.Stop()
// To start, we'll first check to see if there are any stale channels
// that we need to re-transmit.
if err := d.retransmitStaleChannels(); err != nil {
log.Errorf("unable to rebroadcast stale channels: %v",
err)
}
// We'll use this validation to ensure that we process jobs in their
// dependency order during parallel validation.
validationBarrier := routing.NewValidationBarrier(
runtime.NumCPU()*4, d.quit,
)
for {
select {
// A new policy update has arrived. We'll commit it to the
// sub-systems below us, then craft, sign, and broadcast a new
// ChannelUpdate for the set of affected clients.
case policyUpdate := <-d.chanPolicyUpdates:
// First, we'll now create new fully signed updates for
// the affected channels and also update the underlying
// graph with the new state.
newChanUpdates, err := d.processChanPolicyUpdate(policyUpdate)
if err != nil {
log.Errorf("Unable to craft policy updates: %v",
err)
policyUpdate.errResp <- err
continue
}
// Finally, with the updates committed, we'll now add
// them to the announcement batch to be flushed at the
// start of the next epoch.
announcements.AddMsgs(newChanUpdates...)
policyUpdate.errResp <- nil
case announcement := <-d.networkMsgs:
switch msg := announcement.msg.(type) {
// Channel announcement signatures are amongst the only
// messages that we'll process serially.
case *lnwire.AnnounceSignatures:
emittedAnnouncements := d.processNetworkAnnouncement(
announcement,
)
if emittedAnnouncements != nil {
announcements.AddMsgs(
emittedAnnouncements...,
)
}
continue
// If a peer is updating its current update horizon,
// then we'll dispatch that directly to the proper
// gossipSyncer.
case *lnwire.GossipTimestampRange:
syncer := d.findGossipSyncer(announcement.peer)
// If we've found the message target, then
// we'll dispatch the message directly to it.
err := syncer.ApplyGossipFilter(msg)
if err != nil {
log.Warnf("unable to apply gossip "+
"filter for peer=%x: %v",
announcement.peer.SerializeCompressed(), err)
}
continue
// For messages in the known set of channel series
// queries, we'll dispatch the message directly to the
// peer, and skip the main processing loop.
case *lnwire.QueryShortChanIDs,
*lnwire.QueryChannelRange,
*lnwire.ReplyChannelRange,
*lnwire.ReplyShortChanIDsEnd:
syncer := d.findGossipSyncer(announcement.peer)
syncer.ProcessQueryMsg(announcement.msg)
continue
}
// If this message was recently rejected, then we won't
// attempt to re-process it.
if d.isRecentlyRejectedMsg(announcement.msg) {
announcement.err <- fmt.Errorf("recently rejected")
continue
}
// We'll set up any dependent, and wait until a free
// slot for this job opens up, this allow us to not
// have thousands of goroutines active.
validationBarrier.InitJobDependencies(announcement.msg)
d.wg.Add(1)
go func() {
defer d.wg.Done()
defer validationBarrier.CompleteJob()
// If this message has an existing dependency,
// then we'll wait until that has been fully
// validated before we proceed.
err := validationBarrier.WaitForDependants(
announcement.msg,
)
if err != nil {
if err != routing.ErrVBarrierShuttingDown {
log.Warnf("unexpected error "+
"during validation "+
"barrier shutdown: %v",
err)
}
return
}
// Process the network announcement to determine if
// this is either a new announcement from our PoV
// or an edges to a prior vertex/edge we previously
// proceeded.
emittedAnnouncements := d.processNetworkAnnouncement(
announcement,
)
// If this message had any dependencies, then
// we can now signal them to continue.
validationBarrier.SignalDependants(announcement.msg)
// If the announcement was accepted, then add the
// emitted announcements to our announce batch to
// be broadcast once the trickle timer ticks gain.
if emittedAnnouncements != nil {
// TODO(roasbeef): exclude peer that sent
announcements.AddMsgs(
emittedAnnouncements...,
)
}
}()
// A new block has arrived, so we can re-process the previously
// premature announcements.
case newBlock, ok := <-d.newBlocks:
// If the channel has been closed, then this indicates
// the daemon is shutting down, so we exit ourselves.
if !ok {
return
}
// Once a new block arrives, we updates our running
// track of the height of the chain tip.
blockHeight := uint32(newBlock.Height)
atomic.StoreUint32(&d.bestHeight, blockHeight)
// Next we check if we have any premature announcements
// for this height, if so, then we process them once
// more as normal announcements.
d.Lock()
numPremature := len(d.prematureAnnouncements[uint32(newBlock.Height)])
d.Unlock()
if numPremature != 0 {
log.Infof("Re-processing %v premature "+
"announcements for height %v",
numPremature, blockHeight)
}
d.Lock()
for _, ann := range d.prematureAnnouncements[uint32(newBlock.Height)] {
emittedAnnouncements := d.processNetworkAnnouncement(ann)
if emittedAnnouncements != nil {
announcements.AddMsgs(
emittedAnnouncements...,
)
}
}
delete(d.prematureAnnouncements, blockHeight)
d.Unlock()
// The trickle timer has ticked, which indicates we should
// flush to the network the pending batch of new announcements
// we've received since the last trickle tick.
case <-trickleTimer.C:
// Emit the current batch of announcements from
// deDupedAnnouncements.
announcementBatch := announcements.Emit()
// If the current announcements batch is nil, then we
// have no further work here.
if len(announcementBatch) == 0 {
continue
}
// For the set of peers that have an active gossip
// syncers, we'll collect their pubkeys so we can avoid
// sending them the full message blast below.
d.syncerMtx.RLock()
syncerPeers := map[routing.Vertex]struct{}{}
for peerPub := range d.peerSyncers {
syncerPeers[peerPub] = struct{}{}
}
d.syncerMtx.RUnlock()
log.Infof("Broadcasting batch of %v new announcements",
len(announcementBatch))
// We'll first attempt to filter out this new message
// for all peers that have active gossip syncers
// active.
d.syncerMtx.RLock()
for _, syncer := range d.peerSyncers {
syncer.FilterGossipMsgs(announcementBatch...)
}
d.syncerMtx.RUnlock()
// Next, If we have new things to announce then
// broadcast them to all our immediately connected
// peers.
for _, msgChunk := range announcementBatch {
// With the syncers taken care of, we'll merge
// the sender map with the set of syncers, so
// we don't send out duplicate messages.
msgChunk.mergeSyncerMap(syncerPeers)
err := d.cfg.Broadcast(
msgChunk.senders, msgChunk.msg,
)
if err != nil {
log.Errorf("unable to send batch "+
"announcements: %v", err)
continue
}
}
// The retransmission timer has ticked which indicates that we
// should check if we need to prune or re-broadcast any of our
// personal channels. This addresses the case of "zombie" channels and
// channel advertisements that have been dropped, or not properly
// propagated through the network.
case <-retransmitTimer.C:
if err := d.retransmitStaleChannels(); err != nil {
log.Errorf("unable to rebroadcast stale "+
"channels: %v", err)
}
// The gossiper has been signalled to exit, to we exit our
// main loop so the wait group can be decremented.
case <-d.quit:
return
}
}
}
// TODO(roasbeef): d/c peers that send uupdates not on our chain
// InitPeerSyncState is called by outside sub-systems when a connection is
// established to a new peer that understands how to perform channel range
// queries. We'll allocate a new gossip syncer for it, and start any goroutines
// needed to handle new queries. The recvUpdates bool indicates if we should
// continue to receive real-time updates from the remote peer once we've synced
// channel state.
func (d *AuthenticatedGossiper) InitSyncState(peer *btcec.PublicKey, recvUpdates bool) {
d.syncerMtx.Lock()
defer d.syncerMtx.Unlock()
// If we already have a syncer, then we'll exit early as we don't want
// to override it.
nodeID := routing.NewVertex(peer)
if _, ok := d.peerSyncers[nodeID]; ok {
return
}
log.Infof("Creating new gossipSyncer for peer=%x",
peer.SerializeCompressed())
syncer := newGossiperSyncer(gossipSyncerCfg{
chainHash: d.cfg.ChainHash,
syncChanUpdates: recvUpdates,
channelSeries: d.cfg.ChanSeries,
encodingType: lnwire.EncodingSortedPlain,
sendToPeer: func(msgs ...lnwire.Message) error {
return d.cfg.SendToPeer(peer, msgs...)
},
})
copy(syncer.peerPub[:], peer.SerializeCompressed())
d.peerSyncers[nodeID] = syncer
syncer.Start()
}
// PruneSyncState is called by outside sub-systems once a peer that we were
// previously connected to has been disconnected. In this case we can stop the
// existing gossipSyncer assigned to the peer and free up resources.
func (d *AuthenticatedGossiper) PruneSyncState(peer *btcec.PublicKey) {
d.syncerMtx.Lock()
defer d.syncerMtx.Unlock()
log.Infof("Removing gossipSyncer for peer=%x",
peer.SerializeCompressed())
vertex := routing.NewVertex(peer)
syncer, ok := d.peerSyncers[routing.NewVertex(peer)]
if !ok {
return
}
syncer.Stop()
delete(d.peerSyncers, vertex)
return
}
// isRecentlyRejectedMsg returns true if we recently rejected a message, and
// false otherwise, This avoids expensive reprocessing of the message.
func (d *AuthenticatedGossiper) isRecentlyRejectedMsg(msg lnwire.Message) bool {
d.rejectMtx.RLock()
defer d.rejectMtx.RUnlock()
switch m := msg.(type) {
case *lnwire.ChannelUpdate:
_, ok := d.recentRejects[m.ShortChannelID.ToUint64()]
return ok
case *lnwire.ChannelAnnouncement:
_, ok := d.recentRejects[m.ShortChannelID.ToUint64()]
return ok
default:
return false
}
}
// retransmitStaleChannels examines all outgoing channels that the source node
// is known to maintain to check to see if any of them are "stale". A channel
// is stale iff, the last timestamp of its rebroadcast is older then
// broadcastInterval.
func (d *AuthenticatedGossiper) retransmitStaleChannels() error {
// Iterate over all of our channels and check if any of them fall
// within the prune interval or re-broadcast interval.
type updateTuple struct {
info *channeldb.ChannelEdgeInfo
edge *channeldb.ChannelEdgePolicy
}
var edgesToUpdate []updateTuple
err := d.cfg.Router.ForAllOutgoingChannels(func(
info *channeldb.ChannelEdgeInfo,
edge *channeldb.ChannelEdgePolicy) error {
// If there's no auth proof attached to this edge, it means
// that it is a private channel not meant to be announced to
// the greater network, so avoid sending channel updates for
// this channel to not leak its
// existence.
if info.AuthProof == nil {
log.Debugf("Skipping retransmission of channel "+
"without AuthProof: %v", info.ChannelID)
return nil
}
const broadcastInterval = time.Hour * 24
timeElapsed := time.Since(edge.LastUpdate)
// If it's been a full day since we've re-broadcasted the
// channel, add the channel to the set of edges we need to
// update.
if timeElapsed >= broadcastInterval {
edgesToUpdate = append(edgesToUpdate, updateTuple{
info: info,
edge: edge,
})
}
return nil
})
if err != nil {
return fmt.Errorf("error while retrieving outgoing "+
"channels: %v", err)
}
var signedUpdates []lnwire.Message
for _, chanToUpdate := range edgesToUpdate {
// Re-sign and update the channel on disk and retrieve our
// ChannelUpdate to broadcast.
chanAnn, chanUpdate, err := d.updateChannel(
chanToUpdate.info, chanToUpdate.edge,
)
if err != nil {
return fmt.Errorf("unable to update channel: %v", err)
}
// If we have a valid announcement to transmit, then we'll send
// that along with the update.
if chanAnn != nil {
signedUpdates = append(signedUpdates, chanAnn)
}
signedUpdates = append(signedUpdates, chanUpdate)
}
// If we don't have any channels to re-broadcast, then we'll exit
// early.
if len(signedUpdates) == 0 {
return nil
}
log.Infof("Retransmitting %v outgoing channels", len(edgesToUpdate))
// With all the wire announcements properly crafted, we'll broadcast
// our known outgoing channels to all our immediate peers.
if err := d.cfg.Broadcast(nil, signedUpdates...); err != nil {
return fmt.Errorf("unable to re-broadcast channels: %v", err)
}
return nil
}
// processChanPolicyUpdate generates a new set of channel updates with the new
// channel policy applied for each specified channel identified by its channel
// point. In the case that no channel points are specified, then the update
// will be applied to all channels. Finally, the backing ChannelGraphSource is
// updated with the latest information reflecting the applied updates.
//
// TODO(roasbeef): generalize into generic for any channel update
func (d *AuthenticatedGossiper) processChanPolicyUpdate(
policyUpdate *chanPolicyUpdateRequest) ([]networkMsg, error) {
// First, we'll construct a set of all the channels that need to be
// updated.
chansToUpdate := make(map[wire.OutPoint]struct{})
for _, chanPoint := range policyUpdate.targetChans {
chansToUpdate[chanPoint] = struct{}{}
}
haveChanFilter := len(chansToUpdate) != 0
if haveChanFilter {
log.Infof("Updating routing policies for chan_points=%v",
spew.Sdump(chansToUpdate))
} else {
log.Infof("Updating routing policies for all chans")
}
type edgeWithInfo struct {
info *channeldb.ChannelEdgeInfo
edge *channeldb.ChannelEdgePolicy
}
var edgesToUpdate []edgeWithInfo
// Next, we'll loop over all the outgoing channels the router knows of.
// If we have a filter then we'll only collected those channels,
// otherwise we'll collect them all.
err := d.cfg.Router.ForAllOutgoingChannels(func(
info *channeldb.ChannelEdgeInfo,
edge *channeldb.ChannelEdgePolicy) error {
// If we have a channel filter, and this channel isn't a part
// of it, then we'll skip it.
if _, ok := chansToUpdate[info.ChannelPoint]; !ok && haveChanFilter {
return nil
}
// Now that we know we should update this channel, we'll update
// its set of policies.
edge.FeeBaseMSat = policyUpdate.newSchema.BaseFee
edge.FeeProportionalMillionths = lnwire.MilliSatoshi(
policyUpdate.newSchema.FeeRate,
)
edge.TimeLockDelta = uint16(policyUpdate.newSchema.TimeLockDelta)
edgesToUpdate = append(edgesToUpdate, edgeWithInfo{
info: info,
edge: edge,
})
return nil
})
if err != nil {
return nil, err
}
// With the set of edges we need to update retrieved, we'll now re-sign
// them, and insert them into the database.
var chanUpdates []networkMsg
for _, edgeInfo := range edgesToUpdate {
// Now that we've collected all the channels we need to update,
// we'll Re-sign and update the backing ChannelGraphSource, and
// retrieve our ChannelUpdate to broadcast.
_, chanUpdate, err := d.updateChannel(
edgeInfo.info, edgeInfo.edge,
)
if err != nil {
return nil, err
}
// We set ourselves as the source of this message to indicate
// that we shouldn't skip any peers when sending this message.
chanUpdates = append(chanUpdates, networkMsg{
peer: d.selfKey,
msg: chanUpdate,
})
}
return chanUpdates, nil
}
// processRejectedEdge examines a rejected edge to see if we can extract any
// new announcements from it. An edge will get rejected if we already added
// the same edge without AuthProof to the graph. If the received announcement
// contains a proof, we can add this proof to our edge. We can end up in this
// situation in the case where we create a channel, but for some reason fail
// to receive the remote peer's proof, while the remote peer is able to fully
// assemble the proof and craft the ChannelAnnouncement.
func (d *AuthenticatedGossiper) processRejectedEdge(chanAnnMsg *lnwire.ChannelAnnouncement,
proof *channeldb.ChannelAuthProof) ([]networkMsg, error) {
// First, we'll fetch the state of the channel as we know if from the
// database.
chanInfo, e1, e2, err := d.cfg.Router.GetChannelByID(
chanAnnMsg.ShortChannelID,
)
if err != nil {
return nil, err
}
// The edge is in the graph, and has a proof attached, then we'll just
// reject it as normal.
if chanInfo.AuthProof != nil {
return nil, nil
}
// Otherwise, this means that the edge is within the graph, but it
// doesn't yet have a proper proof attached. If we did not receive
// the proof such that we now can add it, there's nothing more we
// can do.
if proof == nil {
return nil, nil
}
// We'll then create then validate the new fully assembled
// announcement.
chanAnn, e1Ann, e2Ann, err := CreateChanAnnouncement(
proof, chanInfo, e1, e2,
)
if err != nil {
return nil, err
}
err = ValidateChannelAnn(chanAnn)
if err != nil {
err := errors.Errorf("assembled channel announcement proof "+
"for shortChanID=%v isn't valid: %v",
chanAnnMsg.ShortChannelID, err)
log.Error(err)
return nil, err
}
// If everything checks out, then we'll add the fully assembled proof
// to the database.
err = d.cfg.Router.AddProof(chanAnnMsg.ShortChannelID, proof)
if err != nil {
err := errors.Errorf("unable add proof to shortChanID=%v: %v",
chanAnnMsg.ShortChannelID, err)
log.Error(err)
return nil, err
}
// As we now have a complete channel announcement for this channel,
// we'll construct the announcement so they can be broadcast out to all
// our peers.
announcements := make([]networkMsg, 0, 3)
announcements = append(announcements, networkMsg{
msg: chanAnn,
peer: d.selfKey,
})
if e1Ann != nil {
announcements = append(announcements, networkMsg{
msg: e1Ann,
peer: d.selfKey,
})
}
if e2Ann != nil {
announcements = append(announcements, networkMsg{
msg: e2Ann,
peer: d.selfKey,
})
}
return announcements, nil
}
// processNetworkAnnouncement processes a new network relate authenticated
// channel or node announcement or announcements proofs. If the announcement
// didn't affect the internal state due to either being out of date, invalid,
// or redundant, then nil is returned. Otherwise, the set of announcements will
// be returned which should be broadcasted to the rest of the network.
func (d *AuthenticatedGossiper) processNetworkAnnouncement(nMsg *networkMsg) []networkMsg {
isPremature := func(chanID lnwire.ShortChannelID, delta uint32) bool {
// TODO(roasbeef) make height delta 6
// * or configurable
bestHeight := atomic.LoadUint32(&d.bestHeight)
return chanID.BlockHeight+delta > bestHeight
}
var announcements []networkMsg
switch msg := nMsg.msg.(type) {
// A new node announcement has arrived which either presents new
// information about a node in one of the channels we know about, or a
// updating previously advertised information.
case *lnwire.NodeAnnouncement:
timestamp := time.Unix(int64(msg.Timestamp), 0)
if nMsg.isRemote {
// We'll quickly ask the router if it already has a
// newer update for this node so we can skip validating
// signatures if not required.
if d.cfg.Router.IsStaleNode(msg.NodeID, timestamp) {
nMsg.err <- nil
return nil
}
if err := ValidateNodeAnn(msg); err != nil {
err := errors.Errorf("unable to validate "+
"node announcement: %v", err)
log.Error(err)
nMsg.err <- err
return nil
}
}
features := lnwire.NewFeatureVector(msg.Features, lnwire.GlobalFeatures)
node := &channeldb.LightningNode{
HaveNodeAnnouncement: true,
LastUpdate: timestamp,
Addresses: msg.Addresses,
PubKeyBytes: msg.NodeID,
Alias: msg.Alias.String(),
AuthSigBytes: msg.Signature.ToSignatureBytes(),
Features: features,
Color: msg.RGBColor,
}
if err := d.cfg.Router.AddNode(node); err != nil {
if routing.IsError(err, routing.ErrOutdated,
routing.ErrIgnored) {
log.Debug(err)
} else {
log.Error(err)
}
nMsg.err <- err
return nil
}
// Node announcement was successfully proceeded and know it
// might be broadcast to other connected nodes.
announcements = append(announcements, networkMsg{
msg: msg,
peer: nMsg.peer,
})
nMsg.err <- nil
// TODO(roasbeef): get rid of the above
return announcements
// A new channel announcement has arrived, this indicates the
// *creation* of a new channel within the network. This only advertises
// the existence of a channel and not yet the routing policies in
// either direction of the channel.
case *lnwire.ChannelAnnouncement:
// We'll ignore any channel announcements that target any chain
// other than the set of chains we know of.
if !bytes.Equal(msg.ChainHash[:], d.cfg.ChainHash[:]) {
log.Errorf("Ignoring ChannelAnnouncement from "+
"chain=%v, gossiper on chain=%v", msg.ChainHash,
d.cfg.ChainHash)
d.rejectMtx.Lock()
d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{}
d.rejectMtx.Unlock()
return nil
}
// If the advertised inclusionary block is beyond our knowledge
// of the chain tip, then we'll put the announcement in limbo
// to be fully verified once we advance forward in the chain.
if nMsg.isRemote && isPremature(msg.ShortChannelID, 0) {
blockHeight := msg.ShortChannelID.BlockHeight
log.Infof("Announcement for chan_id=(%v), is premature: "+
"advertises height %v, only height %v is known",
msg.ShortChannelID.ToUint64(),
msg.ShortChannelID.BlockHeight,
atomic.LoadUint32(&d.bestHeight))
d.Lock()
d.prematureAnnouncements[blockHeight] = append(
d.prematureAnnouncements[blockHeight],
nMsg,
)
d.Unlock()
return nil
}
// At this point, we'll now ask the router if this is a stale
// update. If so we can skip all the processing below.
if d.cfg.Router.IsKnownEdge(msg.ShortChannelID) {
nMsg.err <- nil
return nil
}
// If this is a remote channel announcement, then we'll validate
// all the signatures within the proof as it should be well
// formed.
var proof *channeldb.ChannelAuthProof
if nMsg.isRemote {
if err := ValidateChannelAnn(msg); err != nil {
err := errors.Errorf("unable to validate "+
"announcement: %v", err)
d.rejectMtx.Lock()
d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{}
d.rejectMtx.Unlock()
log.Error(err)
nMsg.err <- err
return nil
}
// If the proof checks out, then we'll save the proof
// itself to the database so we can fetch it later when
// gossiping with other nodes.
proof = &channeldb.ChannelAuthProof{
NodeSig1Bytes: msg.NodeSig1.ToSignatureBytes(),
NodeSig2Bytes: msg.NodeSig2.ToSignatureBytes(),
BitcoinSig1Bytes: msg.BitcoinSig1.ToSignatureBytes(),
BitcoinSig2Bytes: msg.BitcoinSig2.ToSignatureBytes(),
}
}
// With the proof validate (if necessary), we can now store it
// within the database for our path finding and syncing needs.
var featureBuf bytes.Buffer
if err := msg.Features.Encode(&featureBuf); err != nil {
log.Errorf("unable to encode features: %v", err)
nMsg.err <- err
return nil
}
edge := &channeldb.ChannelEdgeInfo{
ChannelID: msg.ShortChannelID.ToUint64(),
ChainHash: msg.ChainHash,
NodeKey1Bytes: msg.NodeID1,
NodeKey2Bytes: msg.NodeID2,
BitcoinKey1Bytes: msg.BitcoinKey1,
BitcoinKey2Bytes: msg.BitcoinKey2,
AuthProof: proof,
Features: featureBuf.Bytes(),
}
// We will add the edge to the channel router. If the nodes
// present in this channel are not present in the database, a
// partial node will be added to represent each node while we
// wait for a node announcement.
//
// Before we add the edge to the database, we obtain
// the mutex for this channel ID. We do this to ensure
// no other goroutine has read the database and is now
// making decisions based on this DB state, before it
// writes to the DB.
d.channelMtx.Lock(msg.ShortChannelID.ToUint64())
defer d.channelMtx.Unlock(msg.ShortChannelID.ToUint64())
if err := d.cfg.Router.AddEdge(edge); err != nil {
// If the edge was rejected due to already being known,
// then it may be that case that this new message has a
// fresh channel proof, so we'll check.
if routing.IsError(err, routing.ErrOutdated,
routing.ErrIgnored) {
// Attempt to process the rejected message to
// see if we get any new announcements.
anns, rErr := d.processRejectedEdge(msg, proof)
if rErr != nil {
d.rejectMtx.Lock()
d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{}
d.rejectMtx.Unlock()
nMsg.err <- rErr
return nil
}
// If while processing this rejected edge, we
// realized there's a set of announcements we
// could extract, then we'll return those
// directly.
if len(anns) != 0 {
nMsg.err <- nil
return anns
}
// Otherwise, this is just a regular rejected
// edge.
log.Debugf("Router rejected channel "+
"edge: %v", err)
} else {
log.Tracef("Router rejected channel "+
"edge: %v", err)
}
nMsg.err <- err
return nil
}
// If we earlier received any ChannelUpdates for this channel,
// we can now process them, as the channel is added to the
// graph.
shortChanID := msg.ShortChannelID.ToUint64()
var channelUpdates []*networkMsg
d.pChanUpdMtx.Lock()
for _, cu := range d.prematureChannelUpdates[shortChanID] {
channelUpdates = append(channelUpdates, cu)
}
// Now delete the premature ChannelUpdates, since we added them
// all to the queue of network messages.
delete(d.prematureChannelUpdates, shortChanID)
d.pChanUpdMtx.Unlock()
// Launch a new goroutine to handle each ChannelUpdate, this to
// ensure we don't block here, as we can handle only one
// announcement at a time.
for _, cu := range channelUpdates {
go func(nMsg *networkMsg) {
switch msg := nMsg.msg.(type) {
case *lnwire.ChannelUpdate:
// We can safely wait for the error to
// be returned, as in case of shutdown,
// the gossiper will return an error.
var err error
if nMsg.isRemote {
err = <-d.ProcessRemoteAnnouncement(
msg, nMsg.peer)
} else {
err = <-d.ProcessLocalAnnouncement(
msg, nMsg.peer)
}
if err != nil {
log.Errorf("Failed reprocessing"+
" ChannelUpdate for "+
"shortChanID=%v: %v",
msg.ShortChannelID.ToUint64(),
err)
return
}
// We don't expect any other message type than
// ChannelUpdate to be in this map.
default:
log.Errorf("Unsupported message type "+
"found among ChannelUpdates: %T", msg)
}
}(cu)
}
// Channel announcement was successfully proceeded and know it
// might be broadcast to other connected nodes if it was
// announcement with proof (remote).
if proof != nil {
announcements = append(announcements, networkMsg{
msg: msg,
peer: nMsg.peer,
})
}
nMsg.err <- nil
return announcements
// A new authenticated channel edge update has arrived. This indicates
// that the directional information for an already known channel has
// been updated.
case *lnwire.ChannelUpdate:
// We'll ignore any channel announcements that target any chain
// other than the set of chains we know of.
if !bytes.Equal(msg.ChainHash[:], d.cfg.ChainHash[:]) {
log.Errorf("Ignoring ChannelUpdate from "+
"chain=%v, gossiper on chain=%v", msg.ChainHash,
d.cfg.ChainHash)
d.rejectMtx.Lock()
d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{}
d.rejectMtx.Unlock()
return nil
}
blockHeight := msg.ShortChannelID.BlockHeight
shortChanID := msg.ShortChannelID.ToUint64()
// If the advertised inclusionary block is beyond our knowledge
// of the chain tip, then we'll put the announcement in limbo
// to be fully verified once we advance forward in the chain.
if nMsg.isRemote && isPremature(msg.ShortChannelID, 0) {
log.Infof("Update announcement for "+
"short_chan_id(%v), is premature: advertises "+
"height %v, only height %v is known",
shortChanID, blockHeight,
atomic.LoadUint32(&d.bestHeight))
d.Lock()
d.prematureAnnouncements[blockHeight] = append(
d.prematureAnnouncements[blockHeight],
nMsg,
)
d.Unlock()
return nil
}
// Before we perform any of the expensive checks below, we'll
// make sure that the router doesn't already have a fresher
// announcement for this edge.
timestamp := time.Unix(int64(msg.Timestamp), 0)
if d.cfg.Router.IsStaleEdgePolicy(
msg.ShortChannelID, timestamp, msg.Flags,
) {
nMsg.err <- nil
return nil
}
// Get the node pub key as far as we don't have it in channel
// update announcement message. We'll need this to properly
// verify message signature.
//
// We make sure to obtain the mutex for this channel ID
// before we access the database. This ensures the state
// we read from the database has not changed between this
// point and when we call UpdateEdge() later.
d.channelMtx.Lock(msg.ShortChannelID.ToUint64())
defer d.channelMtx.Unlock(msg.ShortChannelID.ToUint64())
chanInfo, _, _, err := d.cfg.Router.GetChannelByID(msg.ShortChannelID)
if err != nil {
switch err {
case channeldb.ErrGraphNotFound:
fallthrough
case channeldb.ErrGraphNoEdgesFound:
fallthrough
case channeldb.ErrEdgeNotFound:
// If the edge corresponding to this
// ChannelUpdate was not found in the graph,
// this might be a channel in the process of
// being opened, and we haven't processed our
// own ChannelAnnouncement yet, hence it is not
// found in the graph. This usually gets
// resolved after the channel proofs are
// exchanged and the channel is broadcasted to
// the rest of the network, but in case this
// is a private channel this won't ever happen.
// Because of this, we temporarily add it to a
// map, and reprocess it after our own
// ChannelAnnouncement has been processed.
d.pChanUpdMtx.Lock()
d.prematureChannelUpdates[shortChanID] = append(
d.prematureChannelUpdates[shortChanID],
nMsg,
)
d.pChanUpdMtx.Unlock()
log.Debugf("Got ChannelUpdate for edge not "+
"found in graph(shortChanID=%v), "+
"saving for reprocessing later",
shortChanID)
// If the node supports it, we may try to
// request the chan ann from it.
go func() {
reqErr := d.maybeRequestChanAnn(
msg.ShortChannelID,
)
if reqErr != nil {
log.Errorf("unable to request ann "+
"for chan_id=%v: %v", shortChanID,
reqErr)
}
}()
nMsg.err <- nil
return nil
default:
err := errors.Errorf("unable to validate "+
"channel update short_chan_id=%v: %v",
shortChanID, err)
log.Error(err)
nMsg.err <- err
d.rejectMtx.Lock()
d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{}
d.rejectMtx.Unlock()
return nil
}
}
// The least-significant bit in the flag on the channel update
// announcement tells us "which" side of the channels directed
// edge is being updated.
var pubKey *btcec.PublicKey
switch {
case msg.Flags&lnwire.ChanUpdateDirection == 0:
pubKey, _ = chanInfo.NodeKey1()
case msg.Flags&lnwire.ChanUpdateDirection == 1:
pubKey, _ = chanInfo.NodeKey2()
}
// Validate the channel announcement with the expected public
// key, In the case of an invalid channel , we'll return an
// error to the caller and exit early.
if err := ValidateChannelUpdateAnn(pubKey, msg); err != nil {
rErr := errors.Errorf("unable to validate channel "+
"update announcement for short_chan_id=%v: %v",
spew.Sdump(msg.ShortChannelID), err)
log.Error(rErr)
nMsg.err <- rErr
return nil
}
update := &channeldb.ChannelEdgePolicy{
SigBytes: msg.Signature.ToSignatureBytes(),
ChannelID: shortChanID,
LastUpdate: timestamp,
Flags: msg.Flags,
TimeLockDelta: msg.TimeLockDelta,
MinHTLC: msg.HtlcMinimumMsat,
FeeBaseMSat: lnwire.MilliSatoshi(msg.BaseFee),
FeeProportionalMillionths: lnwire.MilliSatoshi(msg.FeeRate),
}
if err := d.cfg.Router.UpdateEdge(update); err != nil {
if routing.IsError(err, routing.ErrOutdated, routing.ErrIgnored) {
log.Debug(err)
} else {
d.rejectMtx.Lock()
d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{}
d.rejectMtx.Unlock()
log.Error(err)
}
nMsg.err <- err
return nil
}
// If this is a local ChannelUpdate without an AuthProof, it
// means it is an update to a channel that is not (yet)
// supposed to be announced to the greater network. However,
// our channel counter party will need to be given the update,
// so we'll try sending the update directly to the remote peer.
if !nMsg.isRemote && chanInfo.AuthProof == nil {
// Get our peer's public key.
var remotePeer *btcec.PublicKey
switch {
case msg.Flags&lnwire.ChanUpdateDirection == 0:
remotePeer, _ = chanInfo.NodeKey2()
case msg.Flags&lnwire.ChanUpdateDirection == 1:
remotePeer, _ = chanInfo.NodeKey1()
}
// Send ChannelUpdate directly to remotePeer.
// TODO(halseth): make reliable send?
if err = d.cfg.SendToPeer(remotePeer, msg); err != nil {
log.Errorf("unable to send channel update "+
"message to peer %x: %v",
remotePeer.SerializeCompressed(), err)
}
}
// Channel update announcement was successfully processed and
// now it can be broadcast to the rest of the network. However,
// we'll only broadcast the channel update announcement if it
// has an attached authentication proof.
if chanInfo.AuthProof != nil {
announcements = append(announcements, networkMsg{
msg: msg,
peer: nMsg.peer,
})
}
nMsg.err <- nil
return announcements
// A new signature announcement has been received. This indicates
// willingness of nodes involved in the funding of a channel to
// announce this new channel to the rest of the world.
case *lnwire.AnnounceSignatures:
needBlockHeight := msg.ShortChannelID.BlockHeight + d.cfg.ProofMatureDelta
shortChanID := msg.ShortChannelID.ToUint64()
prefix := "local"
if nMsg.isRemote {
prefix = "remote"
}
log.Infof("Received new %v channel announcement: %v", prefix,
spew.Sdump(msg))
// By the specification, channel announcement proofs should be
// sent after some number of confirmations after channel was
// registered in bitcoin blockchain. Therefore, we check if the
// proof is premature. If so we'll halt processing until the
// expected announcement height. This allows us to be tolerant
// to other clients if this constraint was changed.
if isPremature(msg.ShortChannelID, d.cfg.ProofMatureDelta) {
d.Lock()
d.prematureAnnouncements[needBlockHeight] = append(
d.prematureAnnouncements[needBlockHeight],
nMsg,
)
d.Unlock()
log.Infof("Premature proof announcement, "+
"current block height lower than needed: %v <"+
" %v, add announcement to reprocessing batch",
atomic.LoadUint32(&d.bestHeight), needBlockHeight)
return nil
}
// Ensure that we know of a channel with the target channel ID
// before proceeding further.
//
// We must acquire the mutex for this channel ID before getting
// the channel from the database, to ensure what we read does
// not change before we call AddProof() later.
d.channelMtx.Lock(msg.ShortChannelID.ToUint64())
defer d.channelMtx.Unlock(msg.ShortChannelID.ToUint64())
chanInfo, e1, e2, err := d.cfg.Router.GetChannelByID(
msg.ShortChannelID)
if err != nil {
// TODO(andrew.shvv) this is dangerous because remote
// node might rewrite the waiting proof.
proof := channeldb.NewWaitingProof(nMsg.isRemote, msg)
if err := d.waitingProofs.Add(proof); err != nil {
err := errors.Errorf("unable to store "+
"the proof for short_chan_id=%v: %v",
shortChanID, err)
log.Error(err)
nMsg.err <- err
return nil
}
log.Infof("Orphan %v proof announcement with "+
"short_chan_id=%v, adding"+
"to waiting batch", prefix, shortChanID)
nMsg.err <- nil
return nil
}
isFirstNode := bytes.Equal(nMsg.peer.SerializeCompressed(),
chanInfo.NodeKey1Bytes[:])
isSecondNode := bytes.Equal(nMsg.peer.SerializeCompressed(),
chanInfo.NodeKey2Bytes[:])
// Ensure that channel that was retrieved belongs to the peer
// which sent the proof announcement.
if !(isFirstNode || isSecondNode) {
err := errors.Errorf("channel that was received not "+
"belongs to the peer which sent the proof, "+
"short_chan_id=%v", shortChanID)
log.Error(err)
nMsg.err <- err
return nil
}
// If proof was sent by a local sub-system, then we'll
// send the announcement signature to the remote node
// so they can also reconstruct the full channel
// announcement.
if !nMsg.isRemote {
var remotePeer *btcec.PublicKey
if isFirstNode {
remotePeer, _ = chanInfo.NodeKey2()
} else {
remotePeer, _ = chanInfo.NodeKey1()
}
// Since the remote peer might not be online
// we'll call a method that will attempt to
// deliver the proof when it comes online.
if err := d.sendAnnSigReliably(msg, remotePeer); err != nil {
err := errors.Errorf("unable to send reliably "+
"to remote for short_chan_id=%v: %v",
shortChanID, err)
log.Error(err)
nMsg.err <- err
return nil
}
}
// Check if we already have the full proof for this channel.
if chanInfo.AuthProof != nil {
// If we already have the fully assembled proof, then
// the peer sending us their proof has probably not
// received our local proof yet. So be kind and send
// them the full proof.
if nMsg.isRemote {
peerID := nMsg.peer.SerializeCompressed()
log.Debugf("Got AnnounceSignatures for " +
"channel with full proof.")
d.wg.Add(1)
go func() {
defer d.wg.Done()
log.Debugf("Received half proof for "+
"channel %v with existing "+
"full proof. Sending full "+
"proof to peer=%x",
msg.ChannelID,
peerID)
chanAnn, _, _, err := CreateChanAnnouncement(
chanInfo.AuthProof, chanInfo, e1, e2,
)
if err != nil {
log.Errorf("unable to gen ann: %v", err)
return
}
err = d.cfg.SendToPeer(nMsg.peer, chanAnn)
if err != nil {
log.Errorf("Failed sending "+
"full proof to "+
"peer=%x: %v",
peerID, err)
return
}
log.Debugf("Full proof sent to peer=%x"+
" for chanID=%v", peerID, msg.ChannelID)
}()
}
log.Debugf("Already have proof for channel "+
"with chanID=%v", msg.ChannelID)
nMsg.err <- nil
return nil
}
// Check that we received the opposite proof. If so, then we're
// now able to construct the full proof, and create the channel
// announcement. If we didn't receive the opposite half of the
// proof than we should store it this one, and wait for
// opposite to be received.
proof := channeldb.NewWaitingProof(nMsg.isRemote, msg)
oppositeProof, err := d.waitingProofs.Get(proof.OppositeKey())
if err != nil && err != channeldb.ErrWaitingProofNotFound {
err := errors.Errorf("unable to get "+
"the opposite proof for short_chan_id=%v: %v",
shortChanID, err)
log.Error(err)
nMsg.err <- err
return nil
}
if err == channeldb.ErrWaitingProofNotFound {
if err := d.waitingProofs.Add(proof); err != nil {
err := errors.Errorf("unable to store "+
"the proof for short_chan_id=%v: %v",
shortChanID, err)
log.Error(err)
nMsg.err <- err
return nil
}
log.Infof("1/2 of channel ann proof received for "+
"short_chan_id=%v, waiting for other half",
shortChanID)
nMsg.err <- nil
return nil
}
// We now have both halves of the channel announcement proof,
// then we'll reconstruct the initial announcement so we can
// validate it shortly below.
var dbProof channeldb.ChannelAuthProof
if isFirstNode {
dbProof.NodeSig1Bytes = msg.NodeSignature.ToSignatureBytes()
dbProof.NodeSig2Bytes = oppositeProof.NodeSignature.ToSignatureBytes()
dbProof.BitcoinSig1Bytes = msg.BitcoinSignature.ToSignatureBytes()
dbProof.BitcoinSig2Bytes = oppositeProof.BitcoinSignature.ToSignatureBytes()
} else {
dbProof.NodeSig1Bytes = oppositeProof.NodeSignature.ToSignatureBytes()
dbProof.NodeSig2Bytes = msg.NodeSignature.ToSignatureBytes()
dbProof.BitcoinSig1Bytes = oppositeProof.BitcoinSignature.ToSignatureBytes()
dbProof.BitcoinSig2Bytes = msg.BitcoinSignature.ToSignatureBytes()
}
chanAnn, e1Ann, e2Ann, err := CreateChanAnnouncement(&dbProof, chanInfo, e1, e2)
if err != nil {
log.Error(err)
nMsg.err <- err
return nil
}
// With all the necessary components assembled validate the
// full channel announcement proof.
if err := ValidateChannelAnn(chanAnn); err != nil {
err := errors.Errorf("channel announcement proof "+
"for short_chan_id=%v isn't valid: %v",
shortChanID, err)
log.Error(err)
nMsg.err <- err
return nil
}
// If the channel was returned by the router it means that
// existence of funding point and inclusion of nodes bitcoin
// keys in it already checked by the router. In this stage we
// should check that node keys are attest to the bitcoin keys
// by validating the signatures of announcement. If proof is
// valid then we'll populate the channel edge with it, so we
// can announce it on peer connect.
err = d.cfg.Router.AddProof(msg.ShortChannelID, &dbProof)
if err != nil {
err := errors.Errorf("unable add proof to the "+
"channel chanID=%v: %v", msg.ChannelID, err)
log.Error(err)
nMsg.err <- err
return nil
}
if err := d.waitingProofs.Remove(proof.OppositeKey()); err != nil {
err := errors.Errorf("unable remove opposite proof "+
"for the channel with chanID=%v: %v", msg.ChannelID, err)
log.Error(err)
nMsg.err <- err
return nil
}
// Proof was successfully created and now can announce the
// channel to the remain network.
log.Infof("Fully valid channel proof for short_chan_id=%v "+
"constructed, adding to next ann batch",
shortChanID)
// Assemble the necessary announcements to add to the next
// broadcasting batch.
announcements = append(announcements, networkMsg{
msg: chanAnn,
peer: nMsg.peer,
})
if e1Ann != nil {
announcements = append(announcements, networkMsg{
msg: e1Ann,
peer: nMsg.peer,
})
}
if e2Ann != nil {
announcements = append(announcements, networkMsg{
msg: e2Ann,
peer: nMsg.peer,
})
}
nMsg.err <- nil
return announcements
default:
nMsg.err <- errors.New("wrong type of the announcement")
return nil
}
}
// sendAnnSigReliably will try to send the provided local AnnounceSignatures
// to the remote peer, waiting for it to come online if necessary. This
// method returns after adding the message to persistent storage, such
// that the caller knows that the message will be delivered at one point.
func (d *AuthenticatedGossiper) sendAnnSigReliably(
msg *lnwire.AnnounceSignatures, remotePeer *btcec.PublicKey) error {
// We first add this message to the database, such that in case
// we do not succeed in sending it to the peer, we'll fetch it
// from the DB next time we start, and retry. We use the peer ID
// + shortChannelID as key, as there possibly is more than one
// channel opening in progress to the same peer.
var key [41]byte
copy(key[:33], remotePeer.SerializeCompressed())
binary.BigEndian.PutUint64(key[33:], msg.ShortChannelID.ToUint64())
err := d.cfg.DB.Update(func(tx *bolt.Tx) error {
bucket, err := tx.CreateBucketIfNotExists(messageStoreKey)
if err != nil {
return err
}
// Encode the AnnounceSignatures message.
var b bytes.Buffer
if err := msg.Encode(&b, 0); err != nil {
return err
}
// Add the encoded message to the database using the peer
// + shortChanID as key.
return bucket.Put(key[:], b.Bytes())
})
if err != nil {
return err
}
// We have succeeded adding the message to the database. We now launch
// a goroutine that will keep on trying sending the message to the
// remote peer until it succeeds, or the gossiper shuts down. In case
// of success, the message will be removed from the database.
d.wg.Add(1)
go func() {
defer d.wg.Done()
for {
log.Debugf("Sending AnnounceSignatures for channel "+
"%v to remote peer %x", msg.ChannelID,
remotePeer.SerializeCompressed())
err := d.cfg.SendToPeer(remotePeer, msg)
if err == nil {
// Sending succeeded, we can
// continue the flow.
break
}
log.Errorf("unable to send AnnounceSignatures message "+
"to peer(%x): %v. Will retry when online.",
remotePeer.SerializeCompressed(), err)
connected := make(chan struct{})
d.cfg.NotifyWhenOnline(remotePeer, connected)
select {
case <-connected:
log.Infof("peer %x reconnected. Retry sending" +
" AnnounceSignatures.")
// Retry sending.
case <-d.quit:
log.Infof("Gossiper shutting down, did not send" +
" AnnounceSignatures.")
return
}
}
log.Infof("Sent channel announcement proof to remote peer: %x",
remotePeer.SerializeCompressed())
}()
// This method returns after the message has been added to the database,
// such that the caller don't have to wait until the message is actually
// delivered, but can be assured that it will be delivered eventually
// when this method returns.
return nil
}
// updateChannel creates a new fully signed update for the channel, and updates
// the underlying graph with the new state.
func (d *AuthenticatedGossiper) updateChannel(info *channeldb.ChannelEdgeInfo,
edge *channeldb.ChannelEdgePolicy) (*lnwire.ChannelAnnouncement, *lnwire.ChannelUpdate, error) {
var err error
// Make sure timestamp is always increased, such that our update gets
// propagated.
timestamp := time.Now().Unix()
if timestamp <= edge.LastUpdate.Unix() {
timestamp = edge.LastUpdate.Unix() + 1
}
edge.LastUpdate = time.Unix(timestamp, 0)
chanUpdate := &lnwire.ChannelUpdate{
ChainHash: info.ChainHash,
ShortChannelID: lnwire.NewShortChanIDFromInt(edge.ChannelID),
Timestamp: uint32(timestamp),
Flags: edge.Flags,
TimeLockDelta: edge.TimeLockDelta,
HtlcMinimumMsat: edge.MinHTLC,
BaseFee: uint32(edge.FeeBaseMSat),
FeeRate: uint32(edge.FeeProportionalMillionths),
}
chanUpdate.Signature, err = lnwire.NewSigFromRawSignature(edge.SigBytes)
if err != nil {
return nil, nil, err
}
// With the update applied, we'll generate a new signature over a
// digest of the channel announcement itself.
sig, err := SignAnnouncement(d.cfg.AnnSigner, d.selfKey, chanUpdate)
if err != nil {
return nil, nil, err
}
// Next, we'll set the new signature in place, and update the reference
// in the backing slice.
edge.SigBytes = sig.Serialize()
chanUpdate.Signature, err = lnwire.NewSigFromSignature(sig)
if err != nil {
return nil, nil, err
}
// To ensure that our signature is valid, we'll verify it ourself
// before committing it to the slice returned.
err = ValidateChannelUpdateAnn(d.selfKey, chanUpdate)
if err != nil {
return nil, nil, fmt.Errorf("generated invalid channel "+
"update sig: %v", err)
}
// Finally, we'll write the new edge policy to disk.
if err := d.cfg.Router.UpdateEdge(edge); err != nil {
return nil, nil, err
}
// We'll also create the original channel announcement so the two can
// be broadcast along side each other (if necessary), but only if we
// have a full channel announcement for this channel.
var chanAnn *lnwire.ChannelAnnouncement
if info.AuthProof != nil {
chanID := lnwire.NewShortChanIDFromInt(info.ChannelID)
chanAnn = &lnwire.ChannelAnnouncement{
ShortChannelID: chanID,
NodeID1: info.NodeKey1Bytes,
NodeID2: info.NodeKey2Bytes,
ChainHash: info.ChainHash,
BitcoinKey1: info.BitcoinKey1Bytes,
Features: lnwire.NewRawFeatureVector(),
BitcoinKey2: info.BitcoinKey2Bytes,
}
chanAnn.NodeSig1, err = lnwire.NewSigFromRawSignature(
info.AuthProof.NodeSig1Bytes,
)
if err != nil {
return nil, nil, err
}
chanAnn.NodeSig2, err = lnwire.NewSigFromRawSignature(
info.AuthProof.NodeSig2Bytes,
)
if err != nil {
return nil, nil, err
}
chanAnn.BitcoinSig1, err = lnwire.NewSigFromRawSignature(
info.AuthProof.BitcoinSig1Bytes,
)
if err != nil {
return nil, nil, err
}
chanAnn.BitcoinSig2, err = lnwire.NewSigFromRawSignature(
info.AuthProof.BitcoinSig2Bytes,
)
if err != nil {
return nil, nil, err
}
}
return chanAnn, chanUpdate, err
}
// maybeRequestChanAnn will attempt to request the full channel announcement
// for a particular short chan ID. We do this in the case that we get a channel
// update, yet don't already have a channel announcement for it.
func (d *AuthenticatedGossiper) maybeRequestChanAnn(cid lnwire.ShortChannelID) error {
d.syncerMtx.Lock()
defer d.syncerMtx.Unlock()
for nodeID, syncer := range d.peerSyncers {
// If this syncer is already at the terminal state, then we'll
// chose it to request the fully channel update.
if syncer.SyncState() == chansSynced {
pub, err := btcec.ParsePubKey(nodeID[:], btcec.S256())
if err != nil {
return err
}
log.Debugf("attempting to request chan ann for "+
"chan_id=%v from node=%x", cid, nodeID[:])
return d.cfg.SendToPeer(pub, &lnwire.QueryShortChanIDs{
ChainHash: d.cfg.ChainHash,
EncodingType: lnwire.EncodingSortedPlain,
ShortChanIDs: []lnwire.ShortChannelID{cid},
})
}
}
log.Debugf("unable to find peer to request chan ann for chan_id=%v "+
"from", cid)
return nil
}