Merge pull request #9258 from yyforyongyu/fix-notification

chainntnfs: fix missing notifications

Author: Roasbeef

chainntnfs/bitcoindnotify/bitcoind.go

@@ -665,8 +665,14 @@ func (b *BitcoindNotifier) handleBlockConnected(block chainntnfs.BlockEpoch) err
 	// satisfy any client requests based upon the new block.
 	b.bestBlock = block
 
+	err = b.txNotifier.NotifyHeight(uint32(block.Height))
+	if err != nil {
+		return fmt.Errorf("unable to notify height: %w", err)
+	}
+
 	b.notifyBlockEpochs(block.Height, block.Hash, block.BlockHeader)
-	return b.txNotifier.NotifyHeight(uint32(block.Height))
+
+	return nil
 }
 
 // notifyBlockEpochs notifies all registered block epoch clients of the newly

chainntnfs/btcdnotify/btcd.go

@@ -725,11 +725,16 @@ func (b *BtcdNotifier) handleBlockConnected(epoch chainntnfs.BlockEpoch) error {
 	// satisfy any client requests based upon the new block.
 	b.bestBlock = epoch
 
+	err = b.txNotifier.NotifyHeight(uint32(epoch.Height))
+	if err != nil {
+		return fmt.Errorf("unable to notify height: %w", err)
+	}
+
 	b.notifyBlockEpochs(
 		epoch.Height, epoch.Hash, epoch.BlockHeader,
 	)
 
-	return b.txNotifier.NotifyHeight(uint32(epoch.Height))
+	return nil
 }
 
 // notifyBlockEpochs notifies all registered block epoch clients of the newly

chainntnfs/interface.go

@@ -258,6 +258,9 @@ type ConfirmationEvent struct {
 // channels.
 func NewConfirmationEvent(numConfs uint32, cancel func()) *ConfirmationEvent {
 	return &ConfirmationEvent{
+		// We cannot rely on the subscriber to immediately read from
+		// the channel so we need to create a larger buffer to avoid
+		// blocking the notifier.
 		Confirmed:    make(chan *TxConfirmation, 1),
 		Updates:      make(chan uint32, numConfs),
 		NegativeConf: make(chan int32, 1),

chainntnfs/neutrinonotify/neutrino.go

@@ -689,10 +689,16 @@ func (n *NeutrinoNotifier) handleBlockConnected(newBlock *filteredBlock) error {
 	n.bestBlock.Height = int32(newBlock.height)
 	n.bestBlock.BlockHeader = newBlock.header
 
+	err = n.txNotifier.NotifyHeight(newBlock.height)
+	if err != nil {
+		return fmt.Errorf("unable to notify height: %w", err)
+	}
+
 	n.notifyBlockEpochs(
 		int32(newBlock.height), &newBlock.hash, newBlock.header,
 	)
-	return n.txNotifier.NotifyHeight(newBlock.height)
+
+	return nil
 }
 
 // getFilteredBlock is a utility to retrieve the full filtered block from a block epoch.

chainntnfs/txnotifier.go

@@ -244,20 +244,25 @@ type ConfNtfn struct {
 	// notification is to be sent.
 	NumConfirmations uint32
 
-	// Event contains references to the channels that the notifications are to
-	// be sent over.
+	// Event contains references to the channels that the notifications are
+	// to be sent over.
 	Event *ConfirmationEvent
 
 	// HeightHint is the minimum height in the chain that we expect to find
 	// this txid.
 	HeightHint uint32
 
-	// dispatched is false if the confirmed notification has not been sent yet.
+	// dispatched is false if the confirmed notification has not been sent
+	// yet.
 	dispatched bool
 
 	// includeBlock is true if the dispatched notification should also have
 	// the block included with it.
 	includeBlock bool
+
+	// numConfsLeft is the number of confirmations left to be sent to the
+	// subscriber.
+	numConfsLeft uint32
 }
 
 // HistoricalConfDispatch parametrizes a manual rescan for a particular
@@ -589,6 +594,7 @@ func (n *TxNotifier) newConfNtfn(txid *chainhash.Hash,
 		}),
 		HeightHint:   heightHint,
 		includeBlock: opts.includeBlock,
+		numConfsLeft: numConfs,
 	}, nil
 }
 
@@ -664,8 +670,8 @@ func (n *TxNotifier) RegisterConf(txid *chainhash.Hash, pkScript []byte,
 		// already been found, we'll attempt to deliver them immediately
 		// to this client.
 		Log.Debugf("Attempting to dispatch confirmation for %v on "+
-			"registration since rescan has finished",
-			ntfn.ConfRequest)
+			"registration since rescan has finished, conf_id=%v",
+			ntfn.ConfRequest, ntfn.ConfID)
 
 		// The default notification we assigned above includes the
 		// block along with the rest of the details. However not all
@@ -679,9 +685,13 @@ func (n *TxNotifier) RegisterConf(txid *chainhash.Hash, pkScript []byte,
 			confDetails = &confDetailsCopy
 		}
 
-		err := n.dispatchConfDetails(ntfn, confDetails)
-		if err != nil {
-			return nil, err
+		// Deliver the details to the whole conf set where this ntfn
+		// lives in.
+		for _, subscriber := range confSet.ntfns {
+			err := n.dispatchConfDetails(subscriber, confDetails)
+			if err != nil {
+				return nil, err
+			}
 		}
 
 		return &ConfRegistration{
@@ -912,10 +922,16 @@ func (n *TxNotifier) dispatchConfDetails(
 	// If there are no conf details to dispatch or if the notification has
 	// already been dispatched, then we can skip dispatching to this
 	// client.
-	if details == nil || ntfn.dispatched {
-		Log.Debugf("Skipping dispatch of conf details(%v) for "+
-			"request %v, dispatched=%v", details, ntfn.ConfRequest,
-			ntfn.dispatched)
+	if details == nil {
+		Log.Debugf("Skipped dispatching nil conf details for request "+
+			"%v, conf_id=%v", ntfn.ConfRequest, ntfn.ConfID)
+
+		return nil
+	}
+
+	if ntfn.dispatched {
+		Log.Debugf("Skipped dispatched conf details for request %v "+
+			"conf_id=%v", ntfn.ConfRequest, ntfn.ConfID)
 
 		return nil
 	}
@@ -925,16 +941,16 @@ func (n *TxNotifier) dispatchConfDetails(
 	// we'll dispatch a confirmation notification to the caller.
 	confHeight := details.BlockHeight + ntfn.NumConfirmations - 1
 	if confHeight <= n.currentHeight {
-		Log.Debugf("Dispatching %v confirmation notification for %v",
-			ntfn.NumConfirmations, ntfn.ConfRequest)
+		Log.Debugf("Dispatching %v confirmation notification for "+
+			"conf_id=%v, %v", ntfn.NumConfirmations, ntfn.ConfID,
+			ntfn.ConfRequest)
 
 		// We'll send a 0 value to the Updates channel,
 		// indicating that the transaction/output script has already
 		// been confirmed.
-		select {
-		case ntfn.Event.Updates <- 0:
-		case <-n.quit:
-			return ErrTxNotifierExiting
+		err := n.notifyNumConfsLeft(ntfn, 0)
+		if err != nil {
+			return err
 		}
 
 		select {
@@ -944,8 +960,8 @@ func (n *TxNotifier) dispatchConfDetails(
 			return ErrTxNotifierExiting
 		}
 	} else {
-		Log.Debugf("Queueing %v confirmation notification for %v at tip ",
-			ntfn.NumConfirmations, ntfn.ConfRequest)
+		Log.Debugf("Queueing %v confirmation notification for %v at "+
+			"tip", ntfn.NumConfirmations, ntfn.ConfRequest)
 
 		// Otherwise, we'll keep track of the notification
 		// request by the height at which we should dispatch the
@@ -961,10 +977,9 @@ func (n *TxNotifier) dispatchConfDetails(
 		// confirmations are left for the transaction/output script to
 		// be confirmed.
 		numConfsLeft := confHeight - n.currentHeight
-		select {
-		case ntfn.Event.Updates <- numConfsLeft:
-		case <-n.quit:
-			return ErrTxNotifierExiting
+		err := n.notifyNumConfsLeft(ntfn, numConfsLeft)
+		if err != nil {
+			return err
 		}
 	}
 
@@ -1729,10 +1744,9 @@ func (n *TxNotifier) NotifyHeight(height uint32) error {
 					continue
 				}
 
-				select {
-				case ntfn.Event.Updates <- numConfsLeft:
-				case <-n.quit:
-					return ErrTxNotifierExiting
+				err := n.notifyNumConfsLeft(ntfn, numConfsLeft)
+				if err != nil {
+					return err
 				}
 			}
 		}
@@ -1743,8 +1757,9 @@ func (n *TxNotifier) NotifyHeight(height uint32) error {
 	for ntfn := range n.ntfnsByConfirmHeight[height] {
 		confSet := n.confNotifications[ntfn.ConfRequest]
 
-		Log.Debugf("Dispatching %v confirmation notification for %v",
-			ntfn.NumConfirmations, ntfn.ConfRequest)
+		Log.Debugf("Dispatching %v confirmation notification for "+
+			"conf_id=%v, %v", ntfn.NumConfirmations, ntfn.ConfID,
+			ntfn.ConfRequest)
 
 		// The default notification we assigned above includes the
 		// block along with the rest of the details. However not all
@@ -1833,6 +1848,9 @@ func (n *TxNotifier) DisconnectTip(blockHeight uint32) error {
 				default:
 				}
 
+				// We also reset the num of confs update.
+				ntfn.numConfsLeft = ntfn.NumConfirmations
+
 				// Then, we'll check if the current
 				// transaction/output script was included in the
 				// block currently being disconnected. If it
@@ -2069,3 +2087,30 @@ func (n *TxNotifier) TearDown() {
 		}
 	}
 }
+
+// notifyNumConfsLeft sends the number of confirmations left to the
+// notification subscriber through the Event.Updates channel.
+//
+// NOTE: must be used with the TxNotifier's lock held.
+func (n *TxNotifier) notifyNumConfsLeft(ntfn *ConfNtfn, num uint32) error {
+	// If the number left is no less than the recorded value, we can skip
+	// sending it as it means this same value has already been sent before.
+	if num >= ntfn.numConfsLeft {
+		Log.Debugf("Skipped dispatched update (numConfsLeft=%v) for "+
+			"request %v conf_id=%v", num, ntfn.ConfRequest,
+			ntfn.ConfID)
+
+		return nil
+	}
+
+	// Update the number of confirmations left to the notification.
+	ntfn.numConfsLeft = num
+
+	select {
+	case ntfn.Event.Updates <- num:
+	case <-n.quit:
+		return ErrTxNotifierExiting
+	}
+
+	return nil
+}

docs/release-notes/release-notes-0.19.0.md

@@ -54,6 +54,9 @@
 * [Fixed a bug](https://github.com/lightningnetwork/lnd/pull/9275) where the
   peer may block the shutdown process of lnd.
 
+* [Fixed a case](https://github.com/lightningnetwork/lnd/pull/9258) where the
+  confirmation notification may be missed.
+
 # New Features
 ## Functional Enhancements
 ## RPC Additions
@@ -199,6 +202,10 @@ The underlying functionality between those two options remain the same.
 * Oliver Gugger
 * Pins
 * Viktor Tigerström
+<<<<<<< HEAD
 * Yong Yu
 * Ziggie
 
+=======
+* Ziggie
+>>>>>>> 5a6264b6a (docs: update release notes)