From f13c1d278741d3b4c8b4c494f255848371efb86b Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Fri, 21 Sep 2018 17:03:56 -0700 Subject: [PATCH 1/3] channeldb: ensure channel buckets are only created once In this commit, we ensure that we only create the sub-bucket for channels once: at the time of creation. We do this as otherwise it's possible that a method that mutates a channel's state is called after it has already been closed on-chain, leading to the channel bucket being recreated. --- channeldb/channel.go | 126 +++++++++++++++---------------------------- channeldb/error.go | 4 ++ 2 files changed, 47 insertions(+), 83 deletions(-) diff --git a/channeldb/channel.go b/channeldb/channel.go index daaf84439..a284509a8 100644 --- a/channeldb/channel.go +++ b/channeldb/channel.go @@ -495,7 +495,7 @@ func (c *OpenChannel) RefreshShortChanID() error { var sid lnwire.ShortChannelID err := c.Db.View(func(tx *bolt.Tx) error { - chanBucket, err := readChanBucket( + chanBucket, err := fetchChanBucket( tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, ) if err != nil { @@ -521,59 +521,10 @@ func (c *OpenChannel) RefreshShortChanID() error { return nil } -// updateChanBucket is a helper function that returns a writable bucket that a +// fetchChanBucket is a helper function that returns the bucket where a // channel's data resides in given: the public key for the node, the outpoint, // and the chainhash that the channel resides on. -// -// NOTE: This function assumes that all the relevant descendent buckets already -// exist. -func updateChanBucket(tx *bolt.Tx, nodeKey *btcec.PublicKey, - outPoint *wire.OutPoint, chainHash chainhash.Hash) (*bolt.Bucket, error) { - - // First fetch the top level bucket which stores all data related to - // current, active channels. - openChanBucket := tx.Bucket(openChannelBucket) - if openChanBucket == nil { - return nil, ErrNoChanDBExists - } - - // Within this top level bucket, fetch the bucket dedicated to storing - // open channel data specific to the remote node. - nodePub := nodeKey.SerializeCompressed() - nodeChanBucket := openChanBucket.Bucket(nodePub) - if nodeChanBucket == nil { - return nil, ErrNoActiveChannels - } - - // We'll then recurse down an additional layer in order to fetch the - // bucket for this particular chain. - chainBucket, err := nodeChanBucket.CreateBucketIfNotExists(chainHash[:]) - if err != nil { - return nil, ErrNodeNotFound - } - - // With the bucket for the node fetched, we can now go down another - // level, creating the bucket (if it doesn't exist), for this channel - // itself. - var chanPointBuf bytes.Buffer - if err := writeOutpoint(&chanPointBuf, outPoint); err != nil { - return nil, fmt.Errorf("unable to write outpoint: %v", err) - } - chanBucket, err := chainBucket.CreateBucketIfNotExists( - chanPointBuf.Bytes(), - ) - if chanBucket == nil { - return nil, fmt.Errorf("unable to find bucket for "+ - "chan_point=%v", outPoint) - } - - return chanBucket, nil -} - -// readChanBucket is a helper function that returns a readable bucket that a -// channel's data resides in given: the public key for the node, the outpoint, -// and the chainhash that the channel resides on. -func readChanBucket(tx *bolt.Tx, nodeKey *btcec.PublicKey, +func fetchChanBucket(tx *bolt.Tx, nodeKey *btcec.PublicKey, outPoint *wire.OutPoint, chainHash chainhash.Hash) (*bolt.Bucket, error) { // First fetch the top level bucket which stores all data related to @@ -598,16 +549,15 @@ func readChanBucket(tx *bolt.Tx, nodeKey *btcec.PublicKey, return nil, ErrNoActiveChannels } - // With the bucket for the node fetched, we can now go down another - // level, for this channel itself. + // With the bucket for the node and chain fetched, we can now go down + // another level, for this channel itself. var chanPointBuf bytes.Buffer if err := writeOutpoint(&chanPointBuf, outPoint); err != nil { return nil, err } - chanBucket := chainBucket.Bucket(chanPointBuf.Bytes()) if chanBucket == nil { - return nil, ErrNoActiveChannels + return nil, ErrChannelNotFound } return chanBucket, nil @@ -663,7 +613,7 @@ func (c *OpenChannel) MarkAsOpen(openLoc lnwire.ShortChannelID) error { defer c.Unlock() if err := c.Db.Update(func(tx *bolt.Tx) error { - chanBucket, err := updateChanBucket( + chanBucket, err := fetchChanBucket( tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, ) if err != nil { @@ -699,7 +649,7 @@ func (c *OpenChannel) MarkDataLoss(commitPoint *btcec.PublicKey) error { var status ChannelStatus if err := c.Db.Update(func(tx *bolt.Tx) error { - chanBucket, err := updateChanBucket( + chanBucket, err := fetchChanBucket( tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, ) if err != nil { @@ -743,11 +693,14 @@ func (c *OpenChannel) DataLossCommitPoint() (*btcec.PublicKey, error) { var commitPoint *btcec.PublicKey err := c.Db.View(func(tx *bolt.Tx) error { - chanBucket, err := readChanBucket(tx, c.IdentityPub, - &c.FundingOutpoint, c.ChainHash) - if err == ErrNoActiveChannels || err == ErrNoChanDBExists { + chanBucket, err := fetchChanBucket( + tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, + ) + switch err { + case nil: + case ErrNoChanDBExists, ErrNoActiveChannels, ErrChannelNotFound: return ErrNoCommitPoint - } else if err != nil { + default: return err } @@ -791,7 +744,7 @@ func (c *OpenChannel) MarkCommitmentBroadcasted() error { func (c *OpenChannel) putChanStatus(status ChannelStatus) error { if err := c.Db.Update(func(tx *bolt.Tx) error { - chanBucket, err := updateChanBucket( + chanBucket, err := fetchChanBucket( tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, ) if err != nil { @@ -931,7 +884,7 @@ func (c *OpenChannel) UpdateCommitment(newCommitment *ChannelCommitment) error { defer c.Unlock() err := c.Db.Update(func(tx *bolt.Tx) error { - chanBucket, err := updateChanBucket( + chanBucket, err := fetchChanBucket( tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, ) if err != nil { @@ -1352,7 +1305,7 @@ func (c *OpenChannel) AppendRemoteCommitChain(diff *CommitDiff) error { return c.Db.Update(func(tx *bolt.Tx) error { // First, we'll grab the writable bucket where this channel's // data resides. - chanBucket, err := updateChanBucket( + chanBucket, err := fetchChanBucket( tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, ) if err != nil { @@ -1400,11 +1353,14 @@ func (c *OpenChannel) AppendRemoteCommitChain(diff *CommitDiff) error { func (c *OpenChannel) RemoteCommitChainTip() (*CommitDiff, error) { var cd *CommitDiff err := c.Db.View(func(tx *bolt.Tx) error { - chanBucket, err := readChanBucket(tx, c.IdentityPub, - &c.FundingOutpoint, c.ChainHash) - if err == ErrNoActiveChannels || err == ErrNoChanDBExists { + chanBucket, err := fetchChanBucket( + tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, + ) + switch err { + case nil: + case ErrNoChanDBExists, ErrNoActiveChannels, ErrChannelNotFound: return ErrNoPendingCommit - } else if err != nil { + default: return err } @@ -1443,7 +1399,7 @@ func (c *OpenChannel) InsertNextRevocation(revKey *btcec.PublicKey) error { c.RemoteNextRevocation = revKey err := c.Db.Update(func(tx *bolt.Tx) error { - chanBucket, err := updateChanBucket( + chanBucket, err := fetchChanBucket( tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, ) if err != nil { @@ -1473,7 +1429,7 @@ func (c *OpenChannel) AdvanceCommitChainTail(fwdPkg *FwdPkg) error { var newRemoteCommit *ChannelCommitment err := c.Db.Update(func(tx *bolt.Tx) error { - chanBucket, err := updateChanBucket( + chanBucket, err := fetchChanBucket( tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, ) if err != nil { @@ -1642,8 +1598,7 @@ func (c *OpenChannel) RemoveFwdPkg(height uint64) error { // RevocationLogTail returns the "tail", or the end of the current revocation // log. This entry represents the last previous state for the remote node's -// commitment chain. The ChannelDelta returned by this method will always lag -// one state behind the most current (unrevoked) state of the remote node's +// commitment chain. The ChannelDelta returned by this method will always lag one state behind the most current (unrevoked) state of the remote node's // commitment chain. func (c *OpenChannel) RevocationLogTail() (*ChannelCommitment, error) { c.RLock() @@ -1657,8 +1612,9 @@ func (c *OpenChannel) RevocationLogTail() (*ChannelCommitment, error) { var commit ChannelCommitment if err := c.Db.View(func(tx *bolt.Tx) error { - chanBucket, err := readChanBucket(tx, c.IdentityPub, - &c.FundingOutpoint, c.ChainHash) + chanBucket, err := fetchChanBucket( + tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, + ) if err != nil { return err } @@ -1705,8 +1661,9 @@ func (c *OpenChannel) CommitmentHeight() (uint64, error) { err := c.Db.View(func(tx *bolt.Tx) error { // Get the bucket dedicated to storing the metadata for open // channels. - chanBucket, err := readChanBucket(tx, c.IdentityPub, - &c.FundingOutpoint, c.ChainHash) + chanBucket, err := fetchChanBucket( + tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, + ) if err != nil { return err } @@ -1737,8 +1694,9 @@ func (c *OpenChannel) FindPreviousState(updateNum uint64) (*ChannelCommitment, e var commit ChannelCommitment err := c.Db.View(func(tx *bolt.Tx) error { - chanBucket, err := readChanBucket(tx, c.IdentityPub, - &c.FundingOutpoint, c.ChainHash) + chanBucket, err := fetchChanBucket( + tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, + ) if err != nil { return err } @@ -2030,8 +1988,9 @@ func (c *OpenChannel) Snapshot() *ChannelSnapshot { // the local commitment, and the second returned is the remote commitment. func (c *OpenChannel) LatestCommitments() (*ChannelCommitment, *ChannelCommitment, error) { err := c.Db.View(func(tx *bolt.Tx) error { - chanBucket, err := readChanBucket(tx, c.IdentityPub, - &c.FundingOutpoint, c.ChainHash) + chanBucket, err := fetchChanBucket( + tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, + ) if err != nil { return err } @@ -2051,8 +2010,9 @@ func (c *OpenChannel) LatestCommitments() (*ChannelCommitment, *ChannelCommitmen // up to date information required to deliver justice. func (c *OpenChannel) RemoteRevocationStore() (shachain.Store, error) { err := c.Db.View(func(tx *bolt.Tx) error { - chanBucket, err := readChanBucket(tx, c.IdentityPub, - &c.FundingOutpoint, c.ChainHash) + chanBucket, err := fetchChanBucket( + tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash, + ) if err != nil { return err } diff --git a/channeldb/error.go b/channeldb/error.go index e4df0a56c..15cfa840c 100644 --- a/channeldb/error.go +++ b/channeldb/error.go @@ -43,6 +43,10 @@ var ( // specific identity can't be found. ErrNodeNotFound = fmt.Errorf("link node with target identity not found") + // ErrChannelNotFound is returned when we attempt to locate a channel + // for a specific chain, but it is not found. + ErrChannelNotFound = fmt.Errorf("channel not found") + // ErrMetaNotFound is returned when meta bucket hasn't been // created. ErrMetaNotFound = fmt.Errorf("unable to locate meta information") From 99a4952239d7def54b4e279a218b7c8be475a90d Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Fri, 21 Sep 2018 17:04:43 -0700 Subject: [PATCH 2/3] server+peer: check ErrEdgeNotFound when enabling/disabling a channel In this commit, we also check ErrEdgeNotFound when attempting to send an active/inactive channel update for a channel to the network. We do this as it's possible that a channel has confirmed, but it still does not meet the required number of confirmations to be publicly announced. --- peer.go | 6 +++--- server.go | 16 +++++++++------- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/peer.go b/peer.go index d438a4701..659ccd2ee 100644 --- a/peer.go +++ b/peer.go @@ -505,9 +505,9 @@ func (p *peer) loadActiveChannels(chans []*channeldb.OpenChannel) error { // ChannelUpdate. If this channel is already active, // the update won't be sent. err := p.server.announceChanStatus(chanPoint, false) - if err != nil { - peerLog.Errorf("unable to send out active "+ - "channel update: %v", err) + if err != nil && err != channeldb.ErrEdgeNotFound { + srvrLog.Errorf("Unable to enable channel %v: %v", + chanPoint, err) } } }() diff --git a/server.go b/server.go index 4c9ce4c0a..d87f6e92a 100644 --- a/server.go +++ b/server.go @@ -2847,8 +2847,6 @@ func (s *server) announceChanStatus(op wire.OutPoint, disabled bool) error { return nil } - srvrLog.Debugf("Announcing channel(%v) disabled=%v", op, disabled) - // Retrieve the latest update for this channel. We'll use this // as our starting point to send the new update. chanUpdate, err := s.fetchLastChanUpdateByOutPoint(op) @@ -2888,6 +2886,8 @@ func (s *server) announceChanStatus(op wire.OutPoint, disabled bool) error { return err } + srvrLog.Debugf("Announcing channel(%v) disabled=%v", op, disabled) + // Once signed, we'll send the new update to all of our peers. if err := s.applyChannelUpdate(chanUpdate); err != nil { return err @@ -3053,8 +3053,8 @@ func (s *server) watchChannelStatus() { newStatus := make(map[wire.OutPoint]activeStatus) for _, c := range channels { // We'll skip any private channels, as they - // aren't used for routing within the network - // by other nodes. + // aren't used for routing within the network by + // other nodes. if c.ChannelFlags&lnwire.FFAnnounceChannel == 0 { continue } @@ -3113,10 +3113,12 @@ func (s *server) watchChannelStatus() { delete(status, op) err = s.announceChanStatus(op, disable) - if err != nil { + if err != nil && + err != channeldb.ErrEdgeNotFound { + srvrLog.Errorf("Unable to "+ - "disable channel: %v", - err) + "disable channel %v: %v", + op, err) } } } From 828ecfe34379e2f821bb08817394a945ce88fdac Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Fri, 21 Sep 2018 17:13:35 -0700 Subject: [PATCH 3/3] peer: skip sending active channel updates for pending open channels In this commit, we skip sending active channel update for pending open channels as they are not yet part of the graph. --- peer.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/peer.go b/peer.go index 659ccd2ee..c03111d35 100644 --- a/peer.go +++ b/peer.go @@ -488,10 +488,12 @@ func (p *peer) loadActiveChannels(chans []*channeldb.OpenChannel) error { p.activeChannels[chanID] = lnChan p.activeChanMtx.Unlock() - // Only if the channel is public do we need to collect it for - // sending out a new enable update. + // To ensure we can route through this channel now that the peer + // is back online, we'll attempt to send an update to enable it. + // This will only be used for non-pending public channels, as + // they are the only ones capable of routing. chanIsPublic := dbChan.ChannelFlags&lnwire.FFAnnounceChannel != 0 - if chanIsPublic { + if chanIsPublic && !dbChan.IsPending { activePublicChans = append(activePublicChans, *chanPoint) } }