Merge pull request #1963 from wpaulino/create-chan-buckets-once

channeldb: ensure channel buckets are only created once

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.
-//
+func fetchChanBucket(tx *bolt.Tx, nodeKey *btcec.PublicKey,
-// 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,
 	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
+	// With the bucket for the node and chain fetched, we can now go down
-	// level, for this channel itself.
+	// 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,
+		chanBucket, err := fetchChanBucket(
-			&c.FundingOutpoint, c.ChainHash)
+			tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash,
-		if err == ErrNoActiveChannels || err == ErrNoChanDBExists {
+		)
+		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,
+		chanBucket, err := fetchChanBucket(
-			&c.FundingOutpoint, c.ChainHash)
+			tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash,
-		if err == ErrNoActiveChannels || err == ErrNoChanDBExists {
+		)
+		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
+// commitment chain. The ChannelDelta returned by this method will always lag one state behind the most current (unrevoked) state of the remote node's
-// 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,
+		chanBucket, err := fetchChanBucket(
-			&c.FundingOutpoint, c.ChainHash)
+			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,
+		chanBucket, err := fetchChanBucket(
-			&c.FundingOutpoint, c.ChainHash)
+			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,
+		chanBucket, err := fetchChanBucket(
-			&c.FundingOutpoint, c.ChainHash)
+			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,
+		chanBucket, err := fetchChanBucket(
-			&c.FundingOutpoint, c.ChainHash)
+			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,
+		chanBucket, err := fetchChanBucket(
-			&c.FundingOutpoint, c.ChainHash)
+			tx, c.IdentityPub, &c.FundingOutpoint, c.ChainHash,
+		)
 		if err != nil {
 			return err
 		}

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

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
+		// To ensure we can route through this channel now that the peer
-		// sending out a new enable update.
+		// 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)
 		}
 	}
@@ -505,9 +507,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 {
+			if err != nil && err != channeldb.ErrEdgeNotFound {
-				peerLog.Errorf("unable to send out active "+
+				srvrLog.Errorf("Unable to enable channel %v: %v",
-					"channel update: %v", err)
+					chanPoint, err)
 			}
 		}
 	}()

server.go

@@ -2844,8 +2844,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)
@@ -2885,6 +2883,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
@@ -3050,8 +3050,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
+				// aren't used for routing within the network by
-				// by other nodes.
+				// other nodes.
 				if c.ChannelFlags&lnwire.FFAnnounceChannel == 0 {
 					continue
 				}
@@ -3110,10 +3110,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",
+							"disable channel %v: %v",
-							err)
+							op, err)
 					}
 				}
 			}