chainntnfs/bitcoindnotify: handle spend notification registration w/ TxNotifier

In this commit, we modify the logic within RegisterSpendNtfn for the
BitcoindNotifier to account for the recent changes made to the
TxNotifier. Since it is now able to handle spend notification
registration and dispatch, we can bypass all the current logic within
the BitcoindNotifier and interact directly with the TxNotifier instead.

The most notable changes include the following:

  1. We'll only attempt a historical rescan if the TxNotifier tells us
  so.

  2. We'll dispatch the historical rescan within the main goroutine to
  prevent WaitGroup panics, due to the asynchronous nature of the
  notifier.

chainntnfs/bitcoindnotify/bitcoind.go

@@ -12,7 +12,6 @@ import (
 	"github.com/btcsuite/btcd/wire"
 	"github.com/btcsuite/btcutil"
 	"github.com/btcsuite/btcwallet/chain"
-	"github.com/btcsuite/btcwallet/wtxmgr"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/queue"
 )
@@ -144,6 +143,7 @@ func (b *BitcoindNotifier) Start() error {
 
 	b.txNotifier = chainntnfs.NewTxNotifier(
 		uint32(currentHeight), reorgSafetyLimit, b.confirmHintCache,
+		b.spendHintCache,
 	)
 
 	b.bestBlock = chainntnfs.BlockEpoch{
@@ -259,6 +259,8 @@ out:
 				// included in the active chain. We'll do this
 				// in a goroutine to prevent blocking
 				// potentially long rescans.
+				//
+				// TODO(wilmer): add retry logic if rescan fails?
 				b.wg.Add(1)
 				go func() {
 					defer b.wg.Done()
@@ -286,6 +288,25 @@ out:
 					}
 				}()
 
+			case *chainntnfs.HistoricalSpendDispatch:
+				// In order to ensure we don't block the caller
+				// on what may be a long rescan, we'll launch a
+				// goroutine to do so in the background.
+				//
+				// TODO(wilmer): add retry logic if rescan fails?
+				b.wg.Add(1)
+				go func() {
+					defer b.wg.Done()
+
+					err := b.dispatchSpendDetailsManually(msg)
+					if err != nil {
+						chainntnfs.Log.Errorf("Rescan to "+
+							"determine the spend "+
+							"details of %v failed: %v",
+							msg.OutPoint, err)
+					}
+				}()
+
 			case *blockEpochRegistration:
 				chainntnfs.Log.Infof("New block epoch subscription")
 				b.blockEpochClients[msg.epochID] = msg
@@ -383,7 +404,23 @@ out:
 				b.bestBlock = newBestBlock
 
 			case chain.RelevantTx:
-				b.handleRelevantTx(item, b.bestBlock.Height)
+				// We only care about notifying on confirmed
+				// spends, so if this is a mempool spend, we can
+				// ignore it and wait for the spend to appear in
+				// on-chain.
+				if item.Block == nil {
+					continue
+				}
+
+				tx := &item.TxRecord.MsgTx
+				err := b.txNotifier.ProcessRelevantSpendTx(
+					tx, item.Block.Height,
+				)
+				if err != nil {
+					chainntnfs.Log.Errorf("Unable to "+
+						"process transaction %v: %v",
+						tx.TxHash(), err)
+				}
 			}
 
 		case <-b.quit:
@@ -393,55 +430,6 @@ out:
 	b.wg.Done()
 }
 
-// handleRelevantTx notifies any clients of a relevant transaction.
-func (b *BitcoindNotifier) handleRelevantTx(tx chain.RelevantTx, bestHeight int32) {
-	msgTx := tx.TxRecord.MsgTx
-
-	// We only care about notifying on confirmed spends, so in case this is
-	// a mempool spend, we can continue, and wait for the spend to appear
-	// in chain.
-	if tx.Block == nil {
-		return
-	}
-
-	// First, check if this transaction spends an output
-	// that has an existing spend notification for it.
-	for i, txIn := range msgTx.TxIn {
-		prevOut := txIn.PreviousOutPoint
-
-		// If this transaction indeed does spend an
-		// output which we have a registered
-		// notification for, then create a spend
-		// summary, finally sending off the details to
-		// the notification subscriber.
-		if clients, ok := b.spendNotifications[prevOut]; ok {
-			spenderSha := msgTx.TxHash()
-			spendDetails := &chainntnfs.SpendDetail{
-				SpentOutPoint:     &prevOut,
-				SpenderTxHash:     &spenderSha,
-				SpendingTx:        &msgTx,
-				SpenderInputIndex: uint32(i),
-			}
-			spendDetails.SpendingHeight = tx.Block.Height
-
-			for _, ntfn := range clients {
-				chainntnfs.Log.Infof("Dispatching confirmed "+
-					"spend notification for outpoint=%v "+
-					"at height %v", ntfn.targetOutpoint,
-					spendDetails.SpendingHeight)
-				ntfn.spendChan <- spendDetails
-
-				// Close spendChan to ensure that any calls to
-				// Cancel will not block. This is safe to do
-				// since the channel is buffered, and the
-				// message can still be read by the receiver.
-				close(ntfn.spendChan)
-			}
-			delete(b.spendNotifications, prevOut)
-		}
-	}
-}
-
 // historicalConfDetails looks up whether a transaction is already included in a
 // block in the active chain and, if so, returns details about the confirmation.
 func (b *BitcoindNotifier) historicalConfDetails(txid *chainhash.Hash,
@@ -717,64 +705,69 @@ type spendCancel struct {
 func (b *BitcoindNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint,
 	pkScript []byte, heightHint uint32) (*chainntnfs.SpendEvent, error) {
 
-	// Before proceeding to register the notification, we'll query our
+	// First, we'll construct a spend notification request and hand it off
-	// height hint cache to determine whether a better one exists.
+	// to the txNotifier.
-	if hint, err := b.spendHintCache.QuerySpendHint(*outpoint); err == nil {
+	spendID := atomic.AddUint64(&b.spendClientCounter, 1)
-		if hint > heightHint {
+	cancel := func() {
-			chainntnfs.Log.Debugf("Using height hint %d retrieved "+
+		b.txNotifier.CancelSpend(*outpoint, spendID)
-				"from cache for %v", hint, outpoint)
-			heightHint = hint
-		}
 	}
 
-	// Construct a notification request for the outpoint and send it to the
+	ntfn := &chainntnfs.SpendNtfn{
-	// main event loop.
+		SpendID:    spendID,
-	ntfn := &spendNotification{
+		OutPoint:   *outpoint,
-		targetOutpoint: outpoint,
+		PkScript:   pkScript,
-		spendChan:      make(chan *chainntnfs.SpendDetail, 1),
+		Event:      chainntnfs.NewSpendEvent(cancel),
-		spendID:        atomic.AddUint64(&b.spendClientCounter, 1),
+		HeightHint: heightHint,
 	}
 
-	select {
+	historicalDispatch, err := b.txNotifier.RegisterSpend(ntfn)
-	case <-b.quit:
+	if err != nil {
-		return nil, ErrChainNotifierShuttingDown
+		return nil, err
-	case b.notificationRegistry <- ntfn:
 	}
 
+	// If the txNotifier didn't return any details to perform a historical
+	// scan of the chain, then we can return early as there's nothing left
+	// for us to do.
+	if historicalDispatch == nil {
+		return ntfn.Event, nil
+	}
+
+	// We'll then request the backend to notify us when it has detected the
+	// outpoint as spent.
 	if err := b.chainConn.NotifySpent([]*wire.OutPoint{outpoint}); err != nil {
 		return nil, err
 	}
 
-	// The following conditional checks to ensure that when a spend
+	// In addition to the check above, we'll also check the backend's UTXO
-	// notification is registered, the output hasn't already been spent. If
+	// set to determine whether the outpoint has been spent. If it hasn't,
-	// the output is no longer in the UTXO set, the chain will be rescanned
+	// we can return to the caller as well.
-	// from the point where the output was added. The rescan will dispatch
-	// the notification.
 	txOut, err := b.chainConn.GetTxOut(&outpoint.Hash, outpoint.Index, true)
 	if err != nil {
 		return nil, err
 	}
-
-	// If the output is unspent, then we'll write it to the cache with the
-	// given height hint. This allows us to increase the height hint as the
-	// chain extends and the output remains unspent.
 	if txOut != nil {
-		err := b.spendHintCache.CommitSpendHint(heightHint, *outpoint)
+		// We'll let the txNotifier know the outpoint is still unspent
+		// in order to begin updating its spend hint.
+		err := b.txNotifier.UpdateSpendDetails(*outpoint, nil)
 		if err != nil {
-			// The error is not fatal, so we should not return an
+			return nil, err
-			// error to the caller.
-			chainntnfs.Log.Error("Unable to update spend hint to "+
-				"%d for %v: %v", heightHint, *outpoint, err)
 		}
-	} else {
+
-		// Otherwise, we'll determine when the output was spent.
+		return ntfn.Event, nil
+	}
+
+	// Otherwise, we'll determine when the output was spent by scanning the
+	// chain. We'll begin by determining where to start our historical
+	// rescan.
 	//
-		// First, we'll attempt to retrieve the transaction's block hash
+	// As a minimal optimization, we'll query the backend's transaction
-		// using the backend's transaction index.
+	// index (if enabled) to determine if we have a better rescan starting
+	// height. We can do this as the GetRawTransaction call will return the
+	// hash of the block it was included in within the chain.
 	tx, err := b.chainConn.GetRawTransactionVerbose(&outpoint.Hash)
 	if err != nil {
-			// Avoid returning an error if the transaction was not
+		// Avoid returning an error if the transaction was not found to
-			// found to proceed with fallback methods.
+		// proceed with fallback methods.
 		jsonErr, ok := err.(*btcjson.RPCError)
 		if !ok || jsonErr.Code != btcjson.ErrRPCNoTxInfo {
 			return nil, fmt.Errorf("unable to query for "+
@@ -782,102 +775,50 @@ func (b *BitcoindNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint,
 		}
 	}
 
-		var blockHash *chainhash.Hash
+	// If the transaction index was enabled, we'll use the block's hash to
-		if tx != nil && tx.BlockHash != "" {
+	// retrieve its height and check whether it provides a better starting
-			// If we're able to retrieve a valid block hash from the
+	// point for our rescan.
-			// transaction, then we'll use it as our rescan starting
+	if tx != nil {
-			// point.
+		// If the transaction containing the outpoint hasn't confirmed
-			blockHash, err = chainhash.NewHashFromStr(tx.BlockHash)
+		// on-chain, then there's no need to perform a rescan.
+		if tx.BlockHash == "" {
+			return ntfn.Event, nil
+		}
+
+		blockHash, err := chainhash.NewHashFromStr(tx.BlockHash)
 		if err != nil {
 			return nil, err
 		}
-		} else {
+		blockHeight, err := b.chainConn.GetBlockHeight(blockHash)
-			// Otherwise, we'll attempt to retrieve the hash for the
-			// block at the heightHint.
-			blockHash, err = b.chainConn.GetBlockHash(
-				int64(heightHint),
-			)
-			if err != nil {
-				return nil, fmt.Errorf("unable to retrieve "+
-					"hash for block with height %d: %v",
-					heightHint, err)
-			}
-		}
-
-		// We'll only scan old blocks if the transaction has actually
-		// been included within a block. Otherwise, we'll encounter an
-		// error when scanning for blocks. This can happens in the case
-		// of a race condition, wherein the output itself is unspent,
-		// and only arrives in the mempool after the getxout call.
-		if blockHash != nil {
-			// Rescan all the blocks until the current one.
-			startHeight, err := b.chainConn.GetBlockHeight(
-				blockHash,
-			)
 		if err != nil {
 			return nil, err
 		}
 
-			_, endHeight, err := b.chainConn.GetBestBlock()
+		if uint32(blockHeight) > historicalDispatch.StartHeight {
-			if err != nil {
+			historicalDispatch.StartHeight = uint32(blockHeight)
-				return nil, err
-			}
-
-			// In order to ensure we don't block the caller on what
-			// may be a long rescan, we'll launch a goroutine to do
-			// so in the background.
-			b.wg.Add(1)
-			go func() {
-				defer b.wg.Done()
-
-				err := b.dispatchSpendDetailsManually(
-					*outpoint, startHeight, endHeight,
-				)
-				if err != nil {
-					chainntnfs.Log.Errorf("Rescan for spend "+
-						"notification txout(%x) "+
-						"failed: %v", outpoint, err)
-				}
-			}()
 		}
 	}
 
-	return &chainntnfs.SpendEvent{
+	// Now that we've determined the starting point of our rescan, we can
-		Spend: ntfn.spendChan,
+	// dispatch it.
-		Cancel: func() {
-			cancel := &spendCancel{
-				op:      *outpoint,
-				spendID: ntfn.spendID,
-			}
-
-			// Submit spend cancellation to notification dispatcher.
 	select {
-			case b.notificationCancels <- cancel:
+	case b.notificationRegistry <- historicalDispatch:
-				// Cancellation is being handled, drain the
+		return ntfn.Event, nil
-				// spend chan until it is closed before yielding
-				// to the caller.
-				for {
-					select {
-					case _, ok := <-ntfn.spendChan:
-						if !ok {
-							return
-						}
 	case <-b.quit:
-						return
+		return nil, ErrChainNotifierShuttingDown
 	}
-				}
-			case <-b.quit:
-			}
-		},
-	}, nil
 }
 
 // disaptchSpendDetailsManually attempts to manually scan the chain within the
 // given height range for a transaction that spends the given outpoint. If one
 // is found, it's spending details are sent to the notifier dispatcher, which
 // will then dispatch the notification to all of its clients.
-func (b *BitcoindNotifier) dispatchSpendDetailsManually(op wire.OutPoint,
+func (b *BitcoindNotifier) dispatchSpendDetailsManually(
-	startHeight, endHeight int32) error {
+	historicalDispatchDetails *chainntnfs.HistoricalSpendDispatch) error {
+
+	op := historicalDispatchDetails.OutPoint
+	startHeight := historicalDispatchDetails.StartHeight
+	endHeight := historicalDispatchDetails.EndHeight
 
 	// Begin scanning blocks at every height to determine if the outpoint
 	// was spent.
@@ -890,6 +831,7 @@ func (b *BitcoindNotifier) dispatchSpendDetailsManually(op wire.OutPoint,
 		default:
 		}
 
+		// First, we'll fetch the block for the current height.
 		blockHash, err := b.chainConn.GetBlockHash(int64(height))
 		if err != nil {
 			return fmt.Errorf("unable to retrieve hash for block "+
@@ -901,38 +843,30 @@ func (b *BitcoindNotifier) dispatchSpendDetailsManually(op wire.OutPoint,
 				"%v: %v", blockHash, err)
 		}
 
+		// Then, we'll manually go over every transaction in it and
+		// determine whether it spends the outpoint in question.
 		for _, tx := range block.Transactions {
-			for _, in := range tx.TxIn {
+			for i, txIn := range tx.TxIn {
-				if in.PreviousOutPoint != op {
+				if txIn.PreviousOutPoint != op {
 					continue
 				}
 
-				// If this transaction input spends the
+				// If it does, we'll construct its spend details
-				// outpoint, we'll gather the details of the
+				// and hand them over to the TxNotifier so that
-				// spending transaction and dispatch a spend
+				// it can properly notify its registered
-				// notification to our clients.
+				// clients.
-				relTx := chain.RelevantTx{
+				txHash := tx.TxHash()
-					TxRecord: &wtxmgr.TxRecord{
+				details := &chainntnfs.SpendDetail{
-						MsgTx:    *tx,
+					SpentOutPoint:     &op,
-						Hash:     tx.TxHash(),
+					SpenderTxHash:     &txHash,
-						Received: block.Header.Timestamp,
+					SpendingTx:        tx,
-					},
+					SpenderInputIndex: uint32(i),
-					Block: &wtxmgr.BlockMeta{
+					SpendingHeight:    int32(height),
-						Block: wtxmgr.Block{
-							Hash:   *blockHash,
-							Height: height,
-						},
-						Time: block.Header.Timestamp,
-					},
 				}
 
-				select {
+				return b.txNotifier.UpdateSpendDetails(
-				case b.notificationRegistry <- relTx:
+					op, details,
-				case <-b.quit:
+				)
-					return ErrChainNotifierShuttingDown
-				}
-
-				return nil
 			}
 		}
 	}

chainntnfs/bitcoindnotify/bitcoind_dev.go

@@ -31,6 +31,7 @@ func (b *BitcoindNotifier) UnsafeStart(bestHeight int32, bestHash *chainhash.Has
 
 	b.txNotifier = chainntnfs.NewTxNotifier(
 		uint32(bestHeight), reorgSafetyLimit, b.confirmHintCache,
+		b.spendHintCache,
 	)
 
 	if generateBlocks != nil {