From 147c7dc94cb4d97bdaf523aa4bfa13a5e1c18824 Mon Sep 17 00:00:00 2001 From: Turtle Date: Wed, 17 Mar 2021 22:31:41 -0400 Subject: [PATCH 1/3] router: fix 'out of order block error' by retrieving missing blocks in router Fixes an issue where an out of order block error occurs in the router. When this occurs, the change uses the chain notifier to catch up on missed blocks and uses those blocks to fully update the routing graph with closed channels. Fixes #4710, #5132 --- routing/router.go | 197 ++++++++++++++++++++++++++++++++++------------ server.go | 1 + 2 files changed, 147 insertions(+), 51 deletions(-) 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/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, From 70dfa34f38596380b09ae1c79767d5a7f96b935f Mon Sep 17 00:00:00 2001 From: Turtle Date: Wed, 17 Mar 2021 22:32:26 -0400 Subject: [PATCH 2/3] router: test that this change catches up router to the latest blocks --- routing/notifications_test.go | 65 +++++++++++++++++---- routing/router_test.go | 107 ++++++++++++++++++++++++++++++---- 2 files changed, 150 insertions(+), 22 deletions(-) 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_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) + } +} From 02c4f6fe69f97cd540bc42da27a1d623b43a1f9a Mon Sep 17 00:00:00 2001 From: Turtle Date: Wed, 18 Aug 2021 23:49:36 -0400 Subject: [PATCH 3/3] Docs: Updates 0.14.0 release notes --- docs/release-notes/release-notes-0.14.0.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/docs/release-notes/release-notes-0.14.0.md b/docs/release-notes/release-notes-0.14.0.md index 6331c8935..159827f7f 100644 --- a/docs/release-notes/release-notes-0.14.0.md +++ b/docs/release-notes/release-notes-0.14.0.md @@ -267,6 +267,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 @@ -274,6 +276,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