diff --git a/docs/release-notes/release-notes-0.14.0.md b/docs/release-notes/release-notes-0.14.0.md index f36fd6214..167cf06dd 100644 --- a/docs/release-notes/release-notes-0.14.0.md +++ b/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 diff --git a/routing/notifications_test.go b/routing/notifications_test.go index 885a58167..c63aa66e4 100644 --- a/routing/notifications_test.go +++ b/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. diff --git a/routing/router.go b/routing/router.go index 17b5c4e72..8b9cd97da 100644 --- a/routing/router.go +++ b/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 diff --git a/routing/router_test.go b/routing/router_test.go index d7c14969a..b4d49d6c1 100644 --- a/routing/router_test.go +++ b/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) + } +} diff --git a/server.go b/server.go index 8e0498f81..b8d711529 100644 --- a/server.go +++ b/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,