From 6b6beb4d7dd06d6268450277cfbbd81136c88492 Mon Sep 17 00:00:00 2001
From: "Johan T. Halseth" <johanth@gmail.com>
Date: Thu, 12 Sep 2019 14:34:31 +0200
Subject: [PATCH 1/2] chainntnfs/btcd: start concurrent queues prior to
 connection

We would establish the connection to the chain backend and start getting
block notifications before we had started the concurrent queues, which
would lead to the OnBlockConnected call being blocked, and a deadlock
(since GetBestBlock would never return).

Instead we make sure to start the queues before establishing the
connection, consuming the notifications right away.
---
 chainntnfs/btcdnotify/btcd.go | 38 +++++++++++++++++++++++++++--------
 1 file changed, 30 insertions(+), 8 deletions(-)

diff --git a/chainntnfs/btcdnotify/btcd.go b/chainntnfs/btcdnotify/btcd.go
index cdd1e7789..c393300b1 100644
--- a/chainntnfs/btcdnotify/btcd.go
+++ b/chainntnfs/btcdnotify/btcd.go
@@ -139,17 +139,24 @@ func (b *BtcdNotifier) Start() error {
 		return nil
 	}
 
+	// Start our concurrent queues before starting the chain connection, to
+	// ensure onBlockConnected and onRedeemingTx callbacks won't be
+	// blocked.
+	b.chainUpdates.Start()
+	b.txUpdates.Start()
+
 	// Connect to btcd, and register for notifications on connected, and
 	// disconnected blocks.
 	if err := b.chainConn.Connect(20); err != nil {
-		return err
-	}
-	if err := b.chainConn.NotifyBlocks(); err != nil {
+		b.txUpdates.Stop()
+		b.chainUpdates.Stop()
 		return err
 	}
 
 	currentHash, currentHeight, err := b.chainConn.GetBestBlock()
 	if err != nil {
+		b.txUpdates.Stop()
+		b.chainUpdates.Stop()
 		return err
 	}
 
@@ -163,8 +170,11 @@ func (b *BtcdNotifier) Start() error {
 		Hash:   currentHash,
 	}
 
-	b.chainUpdates.Start()
-	b.txUpdates.Start()
+	if err := b.chainConn.NotifyBlocks(); err != nil {
+		b.txUpdates.Stop()
+		b.chainUpdates.Stop()
+		return err
+	}
 
 	b.wg.Add(1)
 	go b.notificationDispatcher()
@@ -208,10 +218,14 @@ func (b *BtcdNotifier) Stop() error {
 func (b *BtcdNotifier) onBlockConnected(hash *chainhash.Hash, height int32, t time.Time) {
 	// Append this new chain update to the end of the queue of new chain
 	// updates.
-	b.chainUpdates.ChanIn() <- &chainUpdate{
+	select {
+	case b.chainUpdates.ChanIn() <- &chainUpdate{
 		blockHash:   hash,
 		blockHeight: height,
 		connect:     true,
+	}:
+	case <-b.quit:
+		return
 	}
 }
 
@@ -236,10 +250,14 @@ type filteredBlock struct {
 func (b *BtcdNotifier) onBlockDisconnected(hash *chainhash.Hash, height int32, t time.Time) {
 	// Append this new chain update to the end of the queue of new chain
 	// updates.
-	b.chainUpdates.ChanIn() <- &chainUpdate{
+	select {
+	case b.chainUpdates.ChanIn() <- &chainUpdate{
 		blockHash:   hash,
 		blockHeight: height,
 		connect:     false,
+	}:
+	case <-b.quit:
+		return
 	}
 }
 
@@ -247,7 +265,11 @@ func (b *BtcdNotifier) onBlockDisconnected(hash *chainhash.Hash, height int32, t
 func (b *BtcdNotifier) onRedeemingTx(tx *btcutil.Tx, details *btcjson.BlockDetails) {
 	// Append this new transaction update to the end of the queue of new
 	// chain updates.
-	b.txUpdates.ChanIn() <- &txUpdate{tx, details}
+	select {
+	case b.txUpdates.ChanIn() <- &txUpdate{tx, details}:
+	case <-b.quit:
+		return
+	}
 }
 
 // notificationDispatcher is the primary goroutine which handles client

From d195acc632a85363a283de0d68cb45f420308fa5 Mon Sep 17 00:00:00 2001
From: "Johan T. Halseth" <johanth@gmail.com>
Date: Thu, 12 Sep 2019 14:36:38 +0200
Subject: [PATCH 2/2] chainntnfs/neutrino: start concurrent queues prior to
 connection

Similar to what was done for btcd, just to make sure we won't be blocked
on any incoming notifications, start the queues first thing during
startup.
---
 chainntnfs/neutrinonotify/neutrino.go | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/chainntnfs/neutrinonotify/neutrino.go b/chainntnfs/neutrinonotify/neutrino.go
index c7d2e1ad6..d06c0996d 100644
--- a/chainntnfs/neutrinonotify/neutrino.go
+++ b/chainntnfs/neutrinonotify/neutrino.go
@@ -116,12 +116,20 @@ func (n *NeutrinoNotifier) Start() error {
 		return nil
 	}
 
+	// Start our concurrent queues before starting the rescan, to ensure
+	// onFilteredBlockConnected and onRelavantTx callbacks won't be
+	// blocked.
+	n.chainUpdates.Start()
+	n.txUpdates.Start()
+
 	// First, we'll obtain the latest block height of the p2p node. We'll
 	// start the auto-rescan from this point. Once a caller actually wishes
 	// to register a chain view, the rescan state will be rewound
 	// accordingly.
 	startingPoint, err := n.p2pNode.BestBlock()
 	if err != nil {
+		n.txUpdates.Stop()
+		n.chainUpdates.Stop()
 		return err
 	}
 	n.bestBlock.Hash = &startingPoint.Hash
@@ -160,9 +168,6 @@ func (n *NeutrinoNotifier) Start() error {
 	)
 	n.rescanErr = n.chainView.Start()
 
-	n.chainUpdates.Start()
-	n.txUpdates.Start()
-
 	n.wg.Add(1)
 	go n.notificationDispatcher()