chainntnfs/txnotifer: prevent dispatching notifications within ConnectTip

In this commit, we modify the TxNotifier's ConnectTip method to no
longer dispatch notifications to any clients who had a request fulfilled
within the height connected. Instead, it will queue the notifications
for dispatch and we add a new method NotifyHeight, which will actually
dispatch them. We do this to allow the users of the TxNotifier to be
more flexible when dispatching notifications.

chainntnfs/txnotifier.go

@@ -887,9 +887,13 @@ func (n *TxNotifier) dispatchSpendDetails(ntfn *SpendNtfn, details *SpendDetail)
 //   confirmation registration for.
 //
 // In the event that the transaction is relevant, a confirmation/spend
-// notification will be dispatched to the relevant clients. Confirmation
+// notification will be queued for dispatch to the relevant clients.
-// notifications will only be dispatched for transactions that have met the
+// Confirmation notifications will only be dispatched for transactions that have
-// required number of confirmations required by the client.
+// met the required number of confirmations required by the client.
+//
+// NOTE: In order to actually dispatch the relevant transaction notifications to
+// clients, NotifyHeight must be called with the same block height in order to
+// maintain correctness.
 func (n *TxNotifier) ConnectTip(blockHash *chainhash.Hash, blockHeight uint32,
 	txns []*btcutil.Tx) error {
 
@@ -1017,14 +1021,24 @@ func (n *TxNotifier) ConnectTip(blockHash *chainhash.Hash, blockHeight uint32,
 		}
 	}
 
-	// Now that we've determined which transactions were confirmed and which
+	// Finally, now that we've determined which transactions were confirmed
-	// outpoints were spent within the new block, we can update their
+	// and which outpoints were spent within the new block, we can update
-	// entries in their respective caches, along with all of our unconfirmed
+	// their entries in their respective caches, along with all of our
-	// transactions and unspent outpoints.
+	// unconfirmed transactions and unspent outpoints.
 	n.updateHints(blockHeight)
 
-	// Next, we'll dispatch an update to all of the notification clients for
+	return nil
-	// our watched transactions with the number of confirmations left at
+}
+
+// NotifyHeight dispatches confirmation and spend notifications to the clients
+// who registered for a notification which has been fulfilled at the passed
+// height.
+func (n *TxNotifier) NotifyHeight(height uint32) error {
+	n.Lock()
+	defer n.Unlock()
+
+	// First, we'll dispatch an update to all of the notification clients
+	// for our watched transactions with the number of confirmations left at
 	// this new height.
 	for _, txHashes := range n.txsByInitialHeight {
 		for txHash := range txHashes {
@@ -1032,7 +1046,7 @@ func (n *TxNotifier) ConnectTip(blockHash *chainhash.Hash, blockHeight uint32,
 			for _, ntfn := range confSet.ntfns {
 				txConfHeight := confSet.details.BlockHeight +
 					ntfn.NumConfirmations - 1
-				numConfsLeft := txConfHeight - blockHeight
+				numConfsLeft := txConfHeight - height
 
 				// Since we don't clear notifications until
 				// transactions are no longer under the risk of
@@ -1054,7 +1068,7 @@ func (n *TxNotifier) ConnectTip(blockHash *chainhash.Hash, blockHeight uint32,
 
 	// Then, we'll dispatch notifications for all the transactions that have
 	// become confirmed at this new block height.
-	for ntfn := range n.ntfnsByConfirmHeight[blockHeight] {
+	for ntfn := range n.ntfnsByConfirmHeight[height] {
 		confSet := n.confNotifications[*ntfn.TxID]
 
 		Log.Infof("Dispatching %v conf notification for %v",
@@ -1067,11 +1081,11 @@ func (n *TxNotifier) ConnectTip(blockHash *chainhash.Hash, blockHeight uint32,
 			return ErrTxNotifierExiting
 		}
 	}
-	delete(n.ntfnsByConfirmHeight, blockHeight)
+	delete(n.ntfnsByConfirmHeight, height)
 
 	// We'll also dispatch spend notifications for all the outpoints that
 	// were spent at this new block height.
-	for op := range n.opsBySpendHeight[blockHeight] {
+	for op := range n.opsBySpendHeight[height] {
 		spendSet := n.spendNotifications[op]
 		for _, ntfn := range spendSet.ntfns {
 			err := n.dispatchSpendDetails(ntfn, spendSet.details)
@@ -1084,8 +1098,8 @@ func (n *TxNotifier) ConnectTip(blockHash *chainhash.Hash, blockHeight uint32,
 	// Finally, we'll clear the entries from our set of notifications for
 	// transactions and outpoints that are no longer under the risk of being
 	// reorged out of the chain.
-	if blockHeight >= n.reorgSafetyLimit {
+	if height >= n.reorgSafetyLimit {
-		matureBlockHeight := blockHeight - n.reorgSafetyLimit
+		matureBlockHeight := height - n.reorgSafetyLimit
 		for tx := range n.txsByInitialHeight[matureBlockHeight] {
 			delete(n.confNotifications, tx)
 		}

chainntnfs/txnotifier_test.go

@@ -165,12 +165,13 @@ func TestTxNotifierFutureConfDispatch(t *testing.T) {
 		Transactions: []*wire.MsgTx{&tx1, &tx2, &tx3},
 	})
 
-	err := n.ConnectTip(
+	err := n.ConnectTip(block1.Hash(), 11, block1.Transactions())
-		block1.Hash(), 11, block1.Transactions(),
-	)
 	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(11); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// We should only receive one update for tx1 since it only requires
 	// one confirmation and it already met it.
@@ -232,6 +233,9 @@ func TestTxNotifierFutureConfDispatch(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(12); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// We should not receive any event notifications for tx1 since it has
 	// already been confirmed.
@@ -388,6 +392,9 @@ func TestTxNotifierHistoricalConfDispatch(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(11); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// We should not receive any event notifications for tx1 since it has
 	// already been confirmed.
@@ -462,6 +469,9 @@ func TestTxNotifierFutureSpendDispatch(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(11); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	expectedSpendDetails := &chainntnfs.SpendDetail{
 		SpentOutPoint:     &ntfn.OutPoint,
@@ -491,6 +501,9 @@ func TestTxNotifierFutureSpendDispatch(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(12); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	select {
 	case <-ntfn.Event.Spend:
@@ -570,6 +583,9 @@ func TestTxNotifierHistoricalSpendDispatch(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(startingHeight + 1); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	select {
 	case <-ntfn.Event.Spend:
@@ -931,6 +947,9 @@ func TestTxNotifierCancelSpend(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(startingHeight + 1); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// The first request should still be active, so we should receive a
 	// spend notification with the correct spending details.
@@ -1024,22 +1043,28 @@ func TestTxNotifierConfReorg(t *testing.T) {
 	block1 := btcutil.NewBlock(&wire.MsgBlock{
 		Transactions: []*wire.MsgTx{&tx1},
 	})
-	err := n.ConnectTip(nil, 8, block1.Transactions())
+	if err := n.ConnectTip(nil, 8, block1.Transactions()); err != nil {
-	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
-	err = n.ConnectTip(nil, 9, nil)
+	if err := n.NotifyHeight(8); err != nil {
-	if err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
+	if err := n.ConnectTip(nil, 9, nil); err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(9); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	block2 := btcutil.NewBlock(&wire.MsgBlock{
 		Transactions: []*wire.MsgTx{&tx2, &tx3},
 	})
-	err = n.ConnectTip(nil, 10, block2.Transactions())
+	if err := n.ConnectTip(nil, 10, block2.Transactions()); err != nil {
-	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(10); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// We should receive two updates for tx1 since it requires two
 	// confirmations and it has already met them.
@@ -1093,20 +1118,23 @@ func TestTxNotifierConfReorg(t *testing.T) {
 
 	// The block that included tx2 and tx3 is disconnected and two next
 	// blocks without them are connected.
-	err = n.DisconnectTip(10)
+	if err := n.DisconnectTip(10); err != nil {
-	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
 
-	err = n.ConnectTip(nil, 10, nil)
+	if err := n.ConnectTip(nil, 10, nil); err != nil {
-	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(10); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
-	err = n.ConnectTip(nil, 11, nil)
+	if err := n.ConnectTip(nil, 11, nil); err != nil {
-	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(11); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	select {
 	case reorgDepth := <-ntfn2.Event.NegativeConf:
@@ -1151,15 +1179,21 @@ func TestTxNotifierConfReorg(t *testing.T) {
 	})
 	block4 := btcutil.NewBlock(&wire.MsgBlock{})
 
-	err = n.ConnectTip(block3.Hash(), 12, block3.Transactions())
+	err := n.ConnectTip(block3.Hash(), 12, block3.Transactions())
 	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(12); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	err = n.ConnectTip(block4.Hash(), 13, block4.Transactions())
 	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(13); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// We should only receive one update for tx2 since it only requires
 	// one confirmation and it already met it.
@@ -1293,12 +1327,13 @@ func TestTxNotifierSpendReorg(t *testing.T) {
 	block1 := btcutil.NewBlock(&wire.MsgBlock{
 		Transactions: []*wire.MsgTx{spendTx1},
 	})
-	err := n.ConnectTip(
+	err := n.ConnectTip(block1.Hash(), startingHeight+1, block1.Transactions())
-		block1.Hash(), startingHeight+1, block1.Transactions(),
-	)
 	if err != nil {
 		t.Fatalf("unable to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(startingHeight + 1); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// We should receive a spend notification for the first outpoint with
 	// its correct spending details.
@@ -1322,12 +1357,13 @@ func TestTxNotifierSpendReorg(t *testing.T) {
 	block2 := btcutil.NewBlock(&wire.MsgBlock{
 		Transactions: []*wire.MsgTx{spendTx2},
 	})
-	err = n.ConnectTip(
+	err = n.ConnectTip(block2.Hash(), startingHeight+2, block2.Transactions())
-		block2.Hash(), startingHeight+2, block2.Transactions(),
-	)
 	if err != nil {
 		t.Fatalf("unable to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(startingHeight + 2); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// We should not receive another spend notification for the first
 	// outpoint.
@@ -1381,6 +1417,9 @@ func TestTxNotifierSpendReorg(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to disconnect block: %v", err)
 	}
+	if err := n.NotifyHeight(startingHeight + 2); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// We shouldn't receive notifications for either of the outpoints.
 	select {
@@ -1403,6 +1442,9 @@ func TestTxNotifierSpendReorg(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(startingHeight + 3); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// We should now receive a spend notification once again for the second
 	// outpoint containing the new spend details.
@@ -1489,12 +1531,13 @@ func TestTxNotifierConfirmHintCache(t *testing.T) {
 		Transactions: []*wire.MsgTx{&txDummy},
 	})
 
-	err = n.ConnectTip(
+	err = n.ConnectTip(block1.Hash(), txDummyHeight, block1.Transactions())
-		block1.Hash(), txDummyHeight, block1.Transactions(),
-	)
 	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(txDummyHeight); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// Since UpdateConfDetails has not been called for either transaction,
 	// the height hints should remain unchanged. This simulates blocks
@@ -1529,12 +1572,13 @@ func TestTxNotifierConfirmHintCache(t *testing.T) {
 		Transactions: []*wire.MsgTx{&tx1},
 	})
 
-	err = n.ConnectTip(
+	err = n.ConnectTip(block2.Hash(), tx1Height, block2.Transactions())
-		block2.Hash(), tx1Height, block2.Transactions(),
-	)
 	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(tx1Height); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// Now that both notifications are waiting at tip for confirmations,
 	// they should have their height hints updated to the latest block
@@ -1563,12 +1607,13 @@ func TestTxNotifierConfirmHintCache(t *testing.T) {
 		Transactions: []*wire.MsgTx{&tx2},
 	})
 
-	err = n.ConnectTip(
+	err = n.ConnectTip(block3.Hash(), tx2Height, block3.Transactions())
-		block3.Hash(), tx2Height, block3.Transactions(),
-	)
 	if err != nil {
 		t.Fatalf("Failed to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(tx2Height); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// The height hint for the first transaction should remain the same.
 	hint, err = hintCache.QueryConfirmHint(tx1Hash)
@@ -1682,6 +1727,9 @@ func TestTxNotifierSpendHintCache(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(dummyHeight); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// Since we haven't called UpdateSpendDetails on any of the test
 	// outpoints, this implies that there is a still a pending historical
@@ -1720,6 +1768,9 @@ func TestTxNotifierSpendHintCache(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(op1Height); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// Both outpoints should have their spend hints reflect the height of
 	// the new block being connected due to the first outpoint being spent
@@ -1749,6 +1800,9 @@ func TestTxNotifierSpendHintCache(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to connect block: %v", err)
 	}
+	if err := n.NotifyHeight(op2Height); err != nil {
+		t.Fatalf("unable to dispatch notifications: %v", err)
+	}
 
 	// Only the second outpoint should have its spend hint updated due to
 	// being spent within the new block. The first outpoint's spend hint