Merge pull request #5315 from orbitalturtle/block_height_discrepancy

Router: fix block height discrepancy

docs/release-notes/release-notes-0.14.0.md

@@ -285,6 +285,8 @@ mode](https://github.com/lightningnetwork/lnd/pull/5564).
 
 [A bug has been fixed when registering for spend notifications in the `txnotifier`. A re-org notification would previously not be dispatched in certain scenarios.](https://github.com/lightningnetwork/lnd/pull/5465)
 
+[Catches up on blocks in the router](https://github.com/lightningnetwork/lnd/pull/5315) in order to fix an "out of order" error that crops up.
+
 ## Documentation 
 
 The [code contribution guidelines have been updated to mention the new
@@ -292,6 +294,7 @@ requirements surrounding updating the release notes for each new
 change](https://github.com/lightningnetwork/lnd/pull/5613). 
 
 # Contributors (Alphabetical Order)
+* Alyssa Hertig
 * Andras Banki-Horvath
 * de6df1re
 * ErikEk

routing/notifications_test.go

@@ -47,6 +47,8 @@ var (
 
 	priv2, _    = btcec.NewPrivateKey(btcec.S256())
 	bitcoinKey2 = priv2.PubKey()
+
+	timeout = time.Second * 5
 )
 
 func createTestNode() (*channeldb.LightningNode, error) {
@@ -122,8 +124,9 @@ func createChannelEdge(ctx *testCtx, bitcoinKey1, bitcoinKey2 []byte,
 }
 
 type mockChain struct {
-	blocks     map[chainhash.Hash]*wire.MsgBlock
-	blockIndex map[uint32]chainhash.Hash
+	blocks           map[chainhash.Hash]*wire.MsgBlock
+	blockIndex       map[uint32]chainhash.Hash
+	blockHeightIndex map[chainhash.Hash]uint32
 
 	utxos map[wire.OutPoint]wire.TxOut
 
@@ -138,10 +141,11 @@ var _ lnwallet.BlockChainIO = (*mockChain)(nil)
 
 func newMockChain(currentHeight uint32) *mockChain {
 	return &mockChain{
-		bestHeight: int32(currentHeight),
-		blocks:     make(map[chainhash.Hash]*wire.MsgBlock),
-		utxos:      make(map[wire.OutPoint]wire.TxOut),
-		blockIndex: make(map[uint32]chainhash.Hash),
+		bestHeight:       int32(currentHeight),
+		blocks:           make(map[chainhash.Hash]*wire.MsgBlock),
+		utxos:            make(map[wire.OutPoint]wire.TxOut),
+		blockIndex:       make(map[uint32]chainhash.Hash),
+		blockHeightIndex: make(map[chainhash.Hash]uint32),
 	}
 }
 
@@ -209,8 +213,10 @@ func (m *mockChain) addBlock(block *wire.MsgBlock, height uint32, nonce uint32)
 	hash := block.Header.BlockHash()
 	m.blocks[hash] = block
 	m.blockIndex[height] = hash
+	m.blockHeightIndex[hash] = height
 	m.Unlock()
 }
+
 func (m *mockChain) GetBlock(blockHash *chainhash.Hash) (*wire.MsgBlock, error) {
 	m.RLock()
 	defer m.RUnlock()
@@ -226,8 +232,10 @@ func (m *mockChain) GetBlock(blockHash *chainhash.Hash) (*wire.MsgBlock, error)
 type mockChainView struct {
 	sync.RWMutex
 
-	newBlocks   chan *chainview.FilteredBlock
-	staleBlocks chan *chainview.FilteredBlock
+	newBlocks           chan *chainview.FilteredBlock
+	staleBlocks         chan *chainview.FilteredBlock
+	notifyBlockAck      chan struct{}
+	notifyStaleBlockAck chan struct{}
 
 	chain lnwallet.BlockChainIO
 
@@ -269,7 +277,7 @@ func (m *mockChainView) UpdateFilter(ops []channeldb.EdgePoint, updateHeight uin
 }
 
 func (m *mockChainView) notifyBlock(hash chainhash.Hash, height uint32,
-	txns []*wire.MsgTx) {
+	txns []*wire.MsgTx, t *testing.T) {
 
 	m.RLock()
 	defer m.RUnlock()
@@ -283,10 +291,23 @@ func (m *mockChainView) notifyBlock(hash chainhash.Hash, height uint32,
 	case <-m.quit:
 		return
 	}
+
+	// Do not ack the block if our notify channel is nil.
+	if m.notifyBlockAck == nil {
+		return
+	}
+
+	select {
+	case m.notifyBlockAck <- struct{}{}:
+	case <-time.After(timeout):
+		t.Fatal("expected block to be delivered")
+	case <-m.quit:
+		return
+	}
 }
 
 func (m *mockChainView) notifyStaleBlock(hash chainhash.Hash, height uint32,
-	txns []*wire.MsgTx) {
+	txns []*wire.MsgTx, t *testing.T) {
 
 	m.RLock()
 	defer m.RUnlock()
@@ -300,6 +321,19 @@ func (m *mockChainView) notifyStaleBlock(hash chainhash.Hash, height uint32,
 	case <-m.quit:
 		return
 	}
+
+	// Do not ack the block if our notify channel is nil.
+	if m.notifyStaleBlockAck == nil {
+		return
+	}
+
+	select {
+	case m.notifyStaleBlockAck <- struct{}{}:
+	case <-time.After(timeout):
+		t.Fatal("expected stale block to be delivered")
+	case <-m.quit:
+		return
+	}
 }
 
 func (m *mockChainView) FilteredBlocks() <-chan *chainview.FilteredBlock {
@@ -317,7 +351,14 @@ func (m *mockChainView) FilterBlock(blockHash *chainhash.Hash) (*chainview.Filte
 		return nil, err
 	}
 
-	filteredBlock := &chainview.FilteredBlock{}
+	chain := m.chain.(*mockChain)
+
+	chain.Lock()
+	filteredBlock := &chainview.FilteredBlock{
+		Hash:   *blockHash,
+		Height: chain.blockHeightIndex[*blockHash],
+	}
+	chain.Unlock()
 	for _, tx := range block.Transactions {
 		for _, txIn := range tx.TxIn {
 			prevOp := txIn.PreviousOutPoint
@@ -895,7 +936,7 @@ func TestChannelCloseNotification(t *testing.T) {
 	}
 	ctx.chain.addBlock(newBlock, blockHeight, blockHeight)
 	ctx.chainView.notifyBlock(newBlock.Header.BlockHash(), blockHeight,
-		newBlock.Transactions)
+		newBlock.Transactions, t)
 
 	// The notification registered above should be sent, if not we'll time
 	// out and mark the test as failed.

routing/router.go

@@ -19,6 +19,7 @@ import (
 	sphinx "github.com/lightningnetwork/lightning-onion"
 	"github.com/lightningnetwork/lnd/amp"
 	"github.com/lightningnetwork/lnd/batch"
+	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/clock"
 	"github.com/lightningnetwork/lnd/htlcswitch"
@@ -290,6 +291,10 @@ type Config struct {
 	// we need in order to properly maintain the channel graph.
 	ChainView chainview.FilteredChainView
 
+	// Notifier is a reference to the ChainNotifier, used to grab
+	// the latest blocks if the router is missing any.
+	Notifier chainntnfs.ChainNotifier
+
 	// Payer is an instance of a PaymentAttemptDispatcher and is used by
 	// the router to send payment attempts onto the network, and receive
 	// their results.
@@ -1140,61 +1145,40 @@ func (r *ChannelRouter) networkHandler() {
 
 			// We'll ensure that any new blocks received attach
 			// directly to the end of our main chain. If not, then
-			// we've somehow missed some blocks. We don't process
-			// this block as otherwise, we may miss on-chain
-			// events.
+			// we've somehow missed some blocks. Here we'll catch
+			// up the chain with the latest blocks.
 			currentHeight := atomic.LoadUint32(&r.bestHeight)
-			if chainUpdate.Height != currentHeight+1 {
-				log.Errorf("out of order block: expecting "+
-					"height=%v, got height=%v", currentHeight+1,
-					chainUpdate.Height)
-				continue
-			}
-
-			// Once a new block arrives, we update our running
-			// track of the height of the chain tip.
-			blockHeight := uint32(chainUpdate.Height)
-			atomic.StoreUint32(&r.bestHeight, blockHeight)
-			log.Infof("Pruning channel graph using block %v (height=%v)",
-				chainUpdate.Hash, blockHeight)
-
-			// We're only interested in all prior outputs that have
-			// been spent in the block, so collate all the
-			// referenced previous outpoints within each tx and
-			// input.
-			var spentOutputs []*wire.OutPoint
-			for _, tx := range chainUpdate.Transactions {
-				for _, txIn := range tx.TxIn {
-					spentOutputs = append(spentOutputs,
-						&txIn.PreviousOutPoint)
+			switch {
+			case chainUpdate.Height == currentHeight+1:
+				err := r.updateGraphWithClosedChannels(
+					chainUpdate,
+				)
+				if err != nil {
+					log.Errorf("unable to prune graph "+
+						"with closed channels: %v", err)
 				}
+
+			case chainUpdate.Height > currentHeight+1:
+				log.Errorf("out of order block: expecting "+
+					"height=%v, got height=%v",
+					currentHeight+1, chainUpdate.Height)
+
+				err := r.getMissingBlocks(currentHeight, chainUpdate)
+				if err != nil {
+					log.Errorf("unable to retrieve missing"+
+						"blocks: %v", err)
+				}
+
+			case chainUpdate.Height < currentHeight+1:
+				log.Errorf("out of order block: expecting "+
+					"height=%v, got height=%v",
+					currentHeight+1, chainUpdate.Height)
+
+				log.Infof("Skipping channel pruning since "+
+					"received block height %v was already"+
+					" processed.", chainUpdate.Height)
 			}
 
-			// With the spent outputs gathered, attempt to prune
-			// the channel graph, also passing in the hash+height
-			// of the block being pruned so the prune tip can be
-			// updated.
-			chansClosed, err := r.cfg.Graph.PruneGraph(spentOutputs,
-				&chainUpdate.Hash, chainUpdate.Height)
-			if err != nil {
-				log.Errorf("unable to prune routing table: %v", err)
-				continue
-			}
-
-			log.Infof("Block %v (height=%v) closed %v channels",
-				chainUpdate.Hash, blockHeight, len(chansClosed))
-
-			if len(chansClosed) == 0 {
-				continue
-			}
-
-			// Notify all currently registered clients of the newly
-			// closed channels.
-			closeSummaries := createCloseSummaries(blockHeight, chansClosed...)
-			r.notifyTopologyChange(&TopologyChange{
-				ClosedChannels: closeSummaries,
-			})
-
 		// A new notification client update has arrived. We're either
 		// gaining a new client, or cancelling notifications for an
 		// existing client.
@@ -1254,6 +1238,117 @@ func (r *ChannelRouter) networkHandler() {
 	}
 }
 
+// getMissingBlocks walks through all missing blocks and updates the graph
+// closed channels accordingly.
+func (r *ChannelRouter) getMissingBlocks(currentHeight uint32,
+	chainUpdate *chainview.FilteredBlock) error {
+
+	outdatedHash, err := r.cfg.Chain.GetBlockHash(int64(currentHeight))
+	if err != nil {
+		return err
+	}
+
+	outdatedBlock := &chainntnfs.BlockEpoch{
+		Height: int32(currentHeight),
+		Hash:   outdatedHash,
+	}
+
+	epochClient, err := r.cfg.Notifier.RegisterBlockEpochNtfn(
+		outdatedBlock,
+	)
+	if err != nil {
+		return err
+	}
+	defer epochClient.Cancel()
+
+	blockDifference := int(chainUpdate.Height - currentHeight)
+
+	// We'll walk through all the outdated blocks and make sure we're able
+	// to update the graph with any closed channels from them.
+	for i := 0; i < blockDifference; i++ {
+		var (
+			missingBlock *chainntnfs.BlockEpoch
+			ok           bool
+		)
+
+		select {
+		case missingBlock, ok = <-epochClient.Epochs:
+			if !ok {
+				return nil
+			}
+
+		case <-r.quit:
+			return nil
+		}
+
+		filteredBlock, err := r.cfg.ChainView.FilterBlock(
+			missingBlock.Hash,
+		)
+		if err != nil {
+			return err
+		}
+
+		err = r.updateGraphWithClosedChannels(
+			filteredBlock,
+		)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// updateGraphWithClosedChannels prunes the channel graph of closed channels
+// that are no longer needed.
+func (r *ChannelRouter) updateGraphWithClosedChannels(
+	chainUpdate *chainview.FilteredBlock) error {
+
+	// Once a new block arrives, we update our running track of the height
+	// of the chain tip.
+	blockHeight := chainUpdate.Height
+
+	atomic.StoreUint32(&r.bestHeight, blockHeight)
+	log.Infof("Pruning channel graph using block %v (height=%v)",
+		chainUpdate.Hash, blockHeight)
+
+	// We're only interested in all prior outputs that have been spent in
+	// the block, so collate all the referenced previous outpoints within
+	// each tx and input.
+	var spentOutputs []*wire.OutPoint
+	for _, tx := range chainUpdate.Transactions {
+		for _, txIn := range tx.TxIn {
+			spentOutputs = append(spentOutputs,
+				&txIn.PreviousOutPoint)
+		}
+	}
+
+	// With the spent outputs gathered, attempt to prune the channel graph,
+	// also passing in the hash+height of the block being pruned so the
+	// prune tip can be updated.
+	chansClosed, err := r.cfg.Graph.PruneGraph(spentOutputs,
+		&chainUpdate.Hash, chainUpdate.Height)
+	if err != nil {
+		log.Errorf("unable to prune routing table: %v", err)
+		return err
+	}
+
+	log.Infof("Block %v (height=%v) closed %v channels", chainUpdate.Hash,
+		blockHeight, len(chansClosed))
+
+	if len(chansClosed) == 0 {
+		return err
+	}
+
+	// Notify all currently registered clients of the newly closed channels.
+	closeSummaries := createCloseSummaries(blockHeight, chansClosed...)
+	r.notifyTopologyChange(&TopologyChange{
+		ClosedChannels: closeSummaries,
+	})
+
+	return nil
+}
+
 // assertNodeAnnFreshness returns a non-nil error if we have an announcement in
 // the database for the passed node with a timestamp newer than the passed
 // timestamp. ErrIgnored will be returned if we already have the node, and

routing/router_test.go

@@ -18,9 +18,11 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
+	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/clock"
 	"github.com/lightningnetwork/lnd/htlcswitch"
+	lnmock "github.com/lightningnetwork/lnd/lntest/mock"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/record"
@@ -44,6 +46,8 @@ type testCtx struct {
 	chain *mockChain
 
 	chainView *mockChainView
+
+	notifier *lnmock.ChainNotifier
 }
 
 func (c *testCtx) getChannelIDFromAlias(t *testing.T, a, b string) uint64 {
@@ -136,11 +140,18 @@ func createTestCtxFromGraphInstanceAssumeValid(t *testing.T,
 		MissionControl:    mc,
 	}
 
+	notifier := &lnmock.ChainNotifier{
+		EpochChan: make(chan *chainntnfs.BlockEpoch),
+		SpendChan: make(chan *chainntnfs.SpendDetail),
+		ConfChan:  make(chan *chainntnfs.TxConfirmation),
+	}
+
 	router, err := New(Config{
 		Graph:              graphInstance.graph,
 		Chain:              chain,
 		ChainView:          chainView,
 		Payer:              &mockPaymentAttemptDispatcherOld{},
+		Notifier:           notifier,
 		Control:            makeMockControlTower(),
 		MissionControl:     mc,
 		SessionSource:      sessionSource,
@@ -171,6 +182,7 @@ func createTestCtxFromGraphInstanceAssumeValid(t *testing.T,
 		channelIDs: graphInstance.channelIDs,
 		chain:      chain,
 		chainView:  chainView,
+		notifier:   notifier,
 	}
 
 	cleanUp := func() {
@@ -1624,7 +1636,7 @@ func TestWakeUpOnStaleBranch(t *testing.T) {
 		ctx.chain.addBlock(block, height, rand.Uint32())
 		ctx.chain.setBestBlock(int32(height))
 		ctx.chainView.notifyBlock(block.BlockHash(), height,
-			[]*wire.MsgTx{})
+			[]*wire.MsgTx{}, t)
 	}
 
 	// Give time to process new blocks
@@ -1656,7 +1668,7 @@ func TestWakeUpOnStaleBranch(t *testing.T) {
 		ctx.chain.addBlock(block, height, rand.Uint32())
 		ctx.chain.setBestBlock(int32(height))
 		ctx.chainView.notifyBlock(block.BlockHash(), height,
-			[]*wire.MsgTx{})
+			[]*wire.MsgTx{}, t)
 	}
 	// Give time to process new blocks
 	time.Sleep(time.Millisecond * 500)
@@ -1833,7 +1845,7 @@ func TestDisconnectedBlocks(t *testing.T) {
 		ctx.chain.addBlock(block, height, rand.Uint32())
 		ctx.chain.setBestBlock(int32(height))
 		ctx.chainView.notifyBlock(block.BlockHash(), height,
-			[]*wire.MsgTx{})
+			[]*wire.MsgTx{}, t)
 	}
 
 	// Give time to process new blocks
@@ -1867,7 +1879,7 @@ func TestDisconnectedBlocks(t *testing.T) {
 		ctx.chain.addBlock(block, height, rand.Uint32())
 		ctx.chain.setBestBlock(int32(height))
 		ctx.chainView.notifyBlock(block.BlockHash(), height,
-			[]*wire.MsgTx{})
+			[]*wire.MsgTx{}, t)
 	}
 	// Give time to process new blocks
 	time.Sleep(time.Millisecond * 500)
@@ -1949,12 +1961,18 @@ func TestDisconnectedBlocks(t *testing.T) {
 	// Create a 15 block fork. We first let the chainView notify the router
 	// about stale blocks, before sending the now connected blocks. We do
 	// this because we expect this order from the chainview.
+	ctx.chainView.notifyStaleBlockAck = make(chan struct{}, 1)
 	for i := len(minorityChain) - 1; i >= 0; i-- {
 		block := minorityChain[i]
 		height := uint32(forkHeight) + uint32(i) + 1
 		ctx.chainView.notifyStaleBlock(block.BlockHash(), height,
-			block.Transactions)
+			block.Transactions, t)
+		<-ctx.chainView.notifyStaleBlockAck
 	}
+
+	time.Sleep(time.Second * 2)
+
+	ctx.chainView.notifyBlockAck = make(chan struct{}, 1)
 	for i := uint32(1); i <= 15; i++ {
 		block := &wire.MsgBlock{
 			Transactions: []*wire.MsgTx{},
@@ -1963,10 +1981,10 @@ func TestDisconnectedBlocks(t *testing.T) {
 		ctx.chain.addBlock(block, height, rand.Uint32())
 		ctx.chain.setBestBlock(int32(height))
 		ctx.chainView.notifyBlock(block.BlockHash(), height,
-			block.Transactions)
+			block.Transactions, t)
+		<-ctx.chainView.notifyBlockAck
 	}
 
-	// Give time to process new blocks
 	time.Sleep(time.Millisecond * 500)
 
 	// chanID2 should not be in the database anymore, since it is not
@@ -2022,7 +2040,7 @@ func TestRouterChansClosedOfflinePruneGraph(t *testing.T) {
 	ctx.chain.addBlock(block102, uint32(nextHeight), rand.Uint32())
 	ctx.chain.setBestBlock(int32(nextHeight))
 	ctx.chainView.notifyBlock(block102.BlockHash(), uint32(nextHeight),
-		[]*wire.MsgTx{})
+		[]*wire.MsgTx{}, t)
 
 	// We'll now create the edges and nodes within the database required
 	// for the ChannelRouter to properly recognize the channel we added
@@ -2075,7 +2093,7 @@ func TestRouterChansClosedOfflinePruneGraph(t *testing.T) {
 		ctx.chain.addBlock(block, uint32(nextHeight), rand.Uint32())
 		ctx.chain.setBestBlock(int32(nextHeight))
 		ctx.chainView.notifyBlock(block.BlockHash(), uint32(nextHeight),
-			[]*wire.MsgTx{})
+			[]*wire.MsgTx{}, t)
 	}
 
 	// At this point, our starting height should be 107.
@@ -2117,7 +2135,7 @@ func TestRouterChansClosedOfflinePruneGraph(t *testing.T) {
 		ctx.chain.addBlock(block, uint32(nextHeight), rand.Uint32())
 		ctx.chain.setBestBlock(int32(nextHeight))
 		ctx.chainView.notifyBlock(block.BlockHash(), uint32(nextHeight),
-			[]*wire.MsgTx{})
+			[]*wire.MsgTx{}, t)
 	}
 
 	// At this point, our starting height should be 112.
@@ -4249,3 +4267,72 @@ func TestSendMPPaymentFailedWithShardsInFlight(t *testing.T) {
 	session.AssertExpectations(t)
 	missionControl.AssertExpectations(t)
 }
+
+// TestBlockDifferenceFix tests if when the router is behind on blocks, the
+// router catches up to the best block head.
+func TestBlockDifferenceFix(t *testing.T) {
+	t.Parallel()
+
+	initialBlockHeight := uint32(0)
+	// Starting height here is set to 0, which is behind where we want to be.
+	ctx, cleanup := createTestCtxSingleNode(t, initialBlockHeight)
+	defer cleanup()
+
+	// Add initial block to our mini blockchain.
+	block := &wire.MsgBlock{
+		Transactions: []*wire.MsgTx{},
+	}
+	ctx.chain.addBlock(block, initialBlockHeight, rand.Uint32())
+
+	// Let's generate a new block of height 5, 5 above where our node is at.
+	newBlock := &wire.MsgBlock{
+		Transactions: []*wire.MsgTx{},
+	}
+	newBlockHeight := uint32(5)
+
+	blockDifference := newBlockHeight - initialBlockHeight
+
+	ctx.chainView.notifyBlockAck = make(chan struct{}, 1)
+
+	ctx.chain.addBlock(newBlock, newBlockHeight, rand.Uint32())
+	ctx.chain.setBestBlock(int32(newBlockHeight))
+	ctx.chainView.notifyBlock(block.BlockHash(), newBlockHeight,
+		[]*wire.MsgTx{}, t)
+
+	<-ctx.chainView.notifyBlockAck
+
+	// At this point, the chain notifier should have noticed that we're
+	// behind on blocks, and will send the n missing blocks that we
+	// need to the client's epochs channel. Let's replicate this
+	// functionality.
+	for i := 0; i < int(blockDifference); i++ {
+		currBlockHeight := int32(i + 1)
+
+		nonce := rand.Uint32()
+
+		newBlock := &wire.MsgBlock{
+			Transactions: []*wire.MsgTx{},
+			Header:       wire.BlockHeader{Nonce: nonce},
+		}
+		ctx.chain.addBlock(newBlock, uint32(currBlockHeight), nonce)
+		currHash := newBlock.Header.BlockHash()
+
+		newEpoch := &chainntnfs.BlockEpoch{
+			Height: currBlockHeight,
+			Hash:   &currHash,
+		}
+
+		ctx.notifier.EpochChan <- newEpoch
+
+		ctx.chainView.notifyBlock(currHash,
+			uint32(currBlockHeight), block.Transactions, t)
+
+		<-ctx.chainView.notifyBlockAck
+	}
+
+	// Then router height should be updated to the latest block.
+	if atomic.LoadUint32(&ctx.router.bestHeight) != newBlockHeight {
+		t.Fatalf("height should have been updated to %v, instead got "+
+			"%v", newBlockHeight, ctx.router.bestHeight)
+	}
+}

server.go

@@ -781,6 +781,7 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 		Graph:               chanGraph,
 		Chain:               cc.ChainIO,
 		ChainView:           cc.ChainView,
+		Notifier:            cc.ChainNotifier,
 		Payer:               s.htlcSwitch,
 		Control:             s.controlTower,
 		MissionControl:      s.missionControl,