multi: add new method ChainArbitrator.RedispatchBlockbeat

This commit adds a new method to enable us resending the blockbeat in
`ChainArbitrator`, which is needed for the channel restore as the chain
watcher and channel arbitrator are added after the start of the chain
arbitrator.

Author: yyforyongyu

chanrestore.go

@@ -8,6 +8,7 @@ import (
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/chanbackup"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/contractcourt"
@@ -286,6 +287,9 @@ func (c *chanDBRestorer) RestoreChansFromSingles(backups ...chanbackup.Single) e
 
 	ltndLog.Infof("Informing chain watchers of new restored channels")
 
+	// Create a slice of channel points.
+	chanPoints := make([]wire.OutPoint, 0, len(channelShells))
+
 	// Finally, we'll need to inform the chain arbitrator of these new
 	// channels so we'll properly watch for their ultimate closure on chain
 	// and sweep them via the DLP.
@@ -294,8 +298,15 @@ func (c *chanDBRestorer) RestoreChansFromSingles(backups ...chanbackup.Single) e
 		if err != nil {
 			return err
 		}
+
+		chanPoints = append(
+			chanPoints, restoredChannel.Chan.FundingOutpoint,
+		)
 	}
 
+	// With all the channels restored, we'll now re-send the blockbeat.
+	c.chainArb.RedispatchBlockbeat(chanPoints)
+
 	return nil
 }
 
@@ -314,7 +325,7 @@ func (s *server) ConnectPeer(nodePub *btcec.PublicKey, addrs []net.Addr) error {
 	// to ensure the new connection is created after this new link/channel
 	// is known.
 	if err := s.DisconnectPeer(nodePub); err != nil {
-		ltndLog.Infof("Peer(%v) is already connected, proceeding "+
+		ltndLog.Infof("Peer(%x) is already connected, proceeding "+
 			"with chan restore", nodePub.SerializeCompressed())
 	}
 

contractcourt/chain_arbitrator.go

@@ -1341,3 +1341,42 @@ func (c *ChainArbitrator) FindOutgoingHTLCDeadline(scid lnwire.ShortChannelID,
 func (c *ChainArbitrator) Name() string {
 	return "ChainArbitrator"
 }
+
+// RedispatchBlockbeat resends the current blockbeat to the channels specified
+// by the chanPoints. It is used when a channel is added to the chain
+// arbitrator after it has been started, e.g., during the channel restore
+// process.
+func (c *ChainArbitrator) RedispatchBlockbeat(chanPoints []wire.OutPoint) {
+	// Get the current blockbeat.
+	beat := c.CurrentBeat()
+
+	// Prepare two sets of consumers.
+	channels := make([]chainio.Consumer, 0, len(chanPoints))
+	watchers := make([]chainio.Consumer, 0, len(chanPoints))
+
+	// Read the active channels in a lock.
+	c.Lock()
+	for _, op := range chanPoints {
+		if channel, ok := c.activeChannels[op]; ok {
+			channels = append(channels, channel)
+		}
+
+		if watcher, ok := c.activeWatchers[op]; ok {
+			watchers = append(watchers, watcher)
+		}
+	}
+	c.Unlock()
+
+	// Iterate all the copied watchers and send the blockbeat to them.
+	err := beat.DispatchConcurrent(watchers)
+	if err != nil {
+		log.Errorf("Notify blockbeat failed: %v", err)
+	}
+
+	// Iterate all the copied channels and send the blockbeat to them.
+	err = beat.DispatchConcurrent(channels)
+	if err != nil {
+		// Shutdown lnd if there's an error processing the block.
+		log.Errorf("Notify blockbeat failed: %v", err)
+	}
+}

itest/lnd_channel_backup_test.go

@@ -1438,9 +1438,6 @@ func assertTimeLockSwept(ht *lntest.HarnessTest, carol, dave *node.HarnessNode,
 	// tx. In addition, Dave will attempt to sweep his anchor output but
 	// fail due to the sweeping tx being uneconomical.
 	expectedTxes := 1
-
-	// Mine a block to trigger the sweeps.
-	ht.MineBlocks(1)
 	ht.AssertNumTxsInMempool(expectedTxes)
 
 	// Carol should consider the channel pending force close (since she is
@@ -1470,11 +1467,10 @@ func assertTimeLockSwept(ht *lntest.HarnessTest, carol, dave *node.HarnessNode,
 	// The commit sweep resolver publishes the sweep tx at defaultCSV-1 and
 	// we already mined one block after the commitment was published, and
 	// one block to trigger Carol's sweeps, so take that into account.
-	ht.MineEmptyBlocks(1)
+	ht.MineEmptyBlocks(2)
 	ht.AssertNumPendingSweeps(dave, 2)
 
 	// Mine a block to trigger the sweeps.
-	ht.MineEmptyBlocks(1)
 	daveSweep := ht.AssertNumTxsInMempool(1)[0]
 	block := ht.MineBlocksAndAssertNumTxes(1, 1)[0]
 	ht.AssertTxInBlock(block, daveSweep)