multi: make deletion of edge atomic.

When the option SCID is used we need to make sure we update the
channel data in an atomic way so that we don't miss any updates
by our peer.

Author: ziggie1984

funding/manager.go

@@ -532,8 +532,7 @@ type Config struct {
 
 	// DeleteAliasEdge allows the Manager to delete an alias channel edge
 	// from the graph. It also returns our local to-be-deleted policy.
-	DeleteAliasEdge func(scid lnwire.ShortChannelID) (
-		*models.ChannelEdgePolicy, error)
+	DeleteAliasEdge func(aliasScID, newScID lnwire.ShortChannelID) error
 
 	// AliasManager is an implementation of the aliasHandler interface that
 	// abstracts away the handling of many alias functions.
@@ -3723,20 +3722,14 @@ func (f *Manager) annAfterSixConfs(completeChan *channeldb.OpenChannel,
 			// addToGraph. This is because the peer may have
 			// sent us a ChannelUpdate with an alias and we don't
 			// want to relay this.
-			ourPolicy, err := f.cfg.DeleteAliasEdge(baseScid)
+			err = f.cfg.DeleteAliasEdge(
+				baseScid, baseScid,
+			)
 			if err != nil {
 				return fmt.Errorf("failed deleting real edge "+
 					"for alias channel from graph: %v",
 					err)
 			}
-
-			err = f.addToGraph(
-				completeChan, &baseScid, nil, ourPolicy,
-			)
-			if err != nil {
-				return fmt.Errorf("failed to re-add to "+
-					"graph: %v", err)
-			}
 		}
 
 		// Create and broadcast the proofs required to make this channel
@@ -3808,24 +3801,13 @@ func (f *Manager) waitForZeroConfChannel(c *channeldb.OpenChannel) error {
 				"six confirmations: %v", err)
 		}
 
-		// TODO: Make this atomic!
-		ourPolicy, err := f.cfg.DeleteAliasEdge(c.ShortChanID())
+		err := f.cfg.DeleteAliasEdge(
+			c.ShortChanID(), confChan.shortChanID,
+		)
 		if err != nil {
 			return fmt.Errorf("unable to delete alias edge from "+
 				"graph: %v", err)
 		}
-
-		// We'll need to update the graph with the new ShortChannelID
-		// via an addToGraph call. We don't pass in the peer's
-		// alias since we'll be using the confirmed SCID from now on
-		// regardless if it's public or not.
-		err = f.addToGraph(
-			c, &confChan.shortChanID, nil, ourPolicy,
-		)
-		if err != nil {
-			return fmt.Errorf("failed adding confirmed zero-conf "+
-				"SCID to graph: %v", err)
-		}
 	}
 
 	// Since we have now marked down the confirmed SCID, we'll also need to

funding/manager_test.go

@@ -550,10 +550,10 @@ func createTestFundingManager(t *testing.T, privKey *btcec.PrivateKey,
 		NotifyOpenChannelEvent:        evt.NotifyOpenChannelEvent,
 		OpenChannelPredicate:          chainedAcceptor,
 		NotifyPendingOpenChannelEvent: evt.NotifyPendingOpenChannelEvent,
-		DeleteAliasEdge: func(scid lnwire.ShortChannelID) (
-			*models.ChannelEdgePolicy, error) {
+		DeleteAliasEdge: func(
+			aliasScID, newScID lnwire.ShortChannelID) error {
 
-			return nil, nil
+			return nil
 		},
 		AliasManager: aliasMgr,
 		// For unit tests we default to false meaning that no funds

graph/db/graph.go

@@ -1178,6 +1178,67 @@ func (c *ChannelGraph) addChannelEdge(tx kvdb.RwTx,
 	return chanIndex.Put(b.Bytes(), chanKey[:])
 }
 
+// ReAddChannelEdge removes the edge with the given channel ID from the
+// database and adds the new edge to garantee atomicity.
+// This is important for option-scid-alias channel which might over the course
+// of their lifetime change their SCID (e.g. public zero-conf channels).
+func (c *ChannelGraph) ReAddChannelEdge(
+	chanID uint64, newEdge *models.ChannelEdgeInfo,
+	ourPolicy *models.ChannelEdgePolicy) error {
+
+	c.cacheMu.Lock()
+	defer c.cacheMu.Unlock()
+
+	err := kvdb.Update(c.db, func(tx kvdb.RwTx) error {
+		edges := tx.ReadWriteBucket(edgeBucket)
+		if edges == nil {
+			return ErrEdgeNotFound
+		}
+		edgeIndex := edges.NestedReadWriteBucket(edgeIndexBucket)
+		if edgeIndex == nil {
+			return ErrEdgeNotFound
+		}
+		chanIndex := edges.NestedReadWriteBucket(channelPointBucket)
+		if chanIndex == nil {
+			return ErrEdgeNotFound
+		}
+		nodes := tx.ReadWriteBucket(nodeBucket)
+		if nodes == nil {
+			return ErrGraphNodeNotFound
+		}
+
+		var rawChanID [8]byte
+		byteOrder.PutUint64(rawChanID[:], chanID)
+
+		// We don't mark this channel as zombie, because we are readding
+		// it immediately after deleting it below.
+		err := c.delChannelEdgeUnsafe(
+			edges, edgeIndex, chanIndex, nil,
+			rawChanID[:], false, false,
+		)
+		if err != nil {
+			return err
+		}
+
+		// Now we add the channel with the new edge info
+		err = c.addChannelEdge(tx, newEdge)
+
+		// Also add the new channel update from our side.
+		_, err = updateEdgePolicy(tx, ourPolicy, c.graphCache)
+
+		return err
+	}, func() {})
+	if err != nil {
+		return err
+	}
+
+	// Remove the Cache entries.
+	c.rejectCache.remove(chanID)
+	c.chanCache.remove(chanID)
+
+	return nil
+}
+
 // HasChannelEdge returns true if the database knows of a channel edge with the
 // passed channel ID, and false otherwise. If an edge with that ID is found
 // within the graph, then two time stamps representing the last time the edge

server.go

@@ -1382,22 +1382,42 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 
 	// Wrap the DeleteChannelEdges method so that the funding manager can
 	// use it without depending on several layers of indirection.
-	deleteAliasEdge := func(scid lnwire.ShortChannelID) (
-		*models.ChannelEdgePolicy, error) {
+	deleteAliasEdge := func(
+		aliasScID, newScID lnwire.ShortChannelID) error {
 
 		info, e1, e2, err := s.graphDB.FetchChannelEdgesByID(
-			scid.ToUint64(),
+			aliasScID.ToUint64(),
 		)
 		if errors.Is(err, graphdb.ErrEdgeNotFound) {
 			// This is unlikely but there is a slim chance of this
 			// being hit if lnd was killed via SIGKILL and the
 			// funding manager was stepping through the delete
 			// alias edge logic.
-			return nil, nil
+			return nil
 		} else if err != nil {
-			return nil, err
+			return err
 		}
 
+		// The AuthProof is only available when the channel is public
+		// and therefore announced to the broader network after six
+		// confirmation. This happens after readding the edge with the
+		// confirmed scid.
+		newEdgeInfo := models.ChannelEdgeInfo{
+			ChannelID:        newScID.ToUint64(),
+			ChannelPoint:     info.ChannelPoint,
+			Capacity:         info.Capacity,
+			ChainHash:        info.ChainHash,
+			NodeKey1Bytes:    info.NodeKey1Bytes,
+			NodeKey2Bytes:    info.NodeKey2Bytes,
+			BitcoinKey1Bytes: info.BitcoinKey1Bytes,
+			BitcoinKey2Bytes: info.BitcoinKey2Bytes,
+			Features:         info.Features,
+			ExtraOpaqueData:  info.ExtraOpaqueData,
+		}
+
+		// We also readd the channel policy from our side with the new
+		// short channel id.
+		//
 		// Grab our key to find our policy.
 		var ourKey [33]byte
 		copy(ourKey[:], nodeKeyDesc.PubKey.SerializeCompressed())
@@ -1411,13 +1431,33 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 
 		if ourPolicy == nil {
 			// Something is wrong, so return an error.
-			return nil, fmt.Errorf("we don't have an edge")
+			return fmt.Errorf("edge policy not found")
+		}
+
+		// Update the policy data, this invalidates the signature
+		// therefore we need to resign the data.
+		ourPolicy.ChannelID = newEdgeInfo.ChannelID
+		chanUpdate := netann.UnsignedChannelUpdateFromEdge(
+			&newEdgeInfo, ourPolicy)
+
+		data, err := chanUpdate.DataToSign()
+
+		nodeSig, err := cc.MsgSigner.SignMessage(
+			nodeKeyDesc.KeyLocator, data, true,
+		)
+		sig, err := lnwire.NewSigFromSignature(nodeSig)
+		if err != nil {
+			return err
 		}
+		ourPolicy.SetSigBytes(sig.ToSignatureBytes())
 
-		err = s.graphDB.DeleteChannelEdges(
-			false, false, scid.ToUint64(),
+		// Delete the old edge information under the alias SCID and add
+		// the updated data with the new SCID.
+		err = s.graphDB.ReAddChannelEdge(
+			aliasScID.ToUint64(), &newEdgeInfo, ourPolicy,
 		)
-		return ourPolicy, err
+
+		return err
 	}
 
 	// For the reservationTimeout and the zombieSweeperInterval different