Merge pull request #1105 from halseth/chainntfs-btcd-hostorical-mempool-spends

Notify on confirmed spends during btcd rescans

chainntnfs/btcdnotify/btcd.go

@@ -370,8 +370,8 @@ out:
 				chainntnfs.Log.Error(err)
 			}
 
-		// NOTE: we currently only use txUpdates for mempool spends. It
-		// might get removed entirely in the future.
+		// NOTE: we currently only use txUpdates for mempool spends and
+		// rescan spends. It might get removed entirely in the future.
 		case item := <-b.txUpdates.ChanOut():
 			newSpend := item.(*txUpdate)
 			spendingTx := newSpend.tx
@@ -397,7 +397,10 @@ out:
 					// TODO(roasbeef): after change to
 					// loadfilter, only notify on block
 					// inclusion?
+
+					confirmedSpend := false
 					if newSpend.details != nil {
+						confirmedSpend = true
 						spendDetails.SpendingHeight = newSpend.details.Height
 					} else {
 						spendDetails.SpendingHeight = currentHeight + 1
@@ -409,17 +412,25 @@ out:
 					// the spend within a block.
 					rem := make(map[uint64]*spendNotification)
 					for c, ntfn := range clients {
-						// If this client didn't want
+						// If this is a mempool spend,
+						// and this client didn't want
 						// to be notified on mempool
 						// spends, store it for later.
-						if !ntfn.mempool {
+						if !confirmedSpend && !ntfn.mempool {
 							rem[c] = ntfn
 							continue
 						}
 
-						chainntnfs.Log.Infof("Dispatching "+
+						confStr := "unconfirmed"
+						if confirmedSpend {
+							confStr = "confirmed"
+						}
+
+						chainntnfs.Log.Infof("Dispatching %s "+
 							"spend notification for "+
-							"outpoint=%v", ntfn.targetOutpoint)
+							"outpoint=%v at height %v",
+							confStr, ntfn.targetOutpoint,
+							spendDetails.SpendingHeight)
 						ntfn.spendChan <- spendDetails
 
 						// Close spendChan to ensure that any calls to Cancel will not

chainntnfs/interface_test.go

@@ -1000,6 +1000,13 @@ func testSpendBeforeNtfnRegistration(miner *rpctest.Harness,
 		t.Fatalf("tx not relayed to miner: %v", err)
 	}
 
+	// We create an epoch client we can use to make sure the notifier is
+	// caught up to the mining node's chain.
+	epochClient, err := notifier.RegisterBlockEpochNtfn()
+	if err != nil {
+		t.Fatalf("unable to register for block epoch: %v", err)
+	}
+
 	// Now we mine an additional block, which should include our spend.
 	if _, err := miner.Node.Generate(1); err != nil {
 		t.Fatalf("unable to generate single block: %v", err)
@@ -1010,39 +1017,91 @@ func testSpendBeforeNtfnRegistration(miner *rpctest.Harness,
 		t.Fatalf("unable to get current height: %v", err)
 	}
 
-	// Now, we register to be notified of a spend that has already
-	// happened.  The notifier should dispatch a spend notification
-	// immediately.
-	spentIntent, err := notifier.RegisterSpendNtfn(outpoint,
-		uint32(currentHeight), true)
-	if err != nil {
-		t.Fatalf("unable to register for spend ntfn: %v", err)
+	// checkSpends registers two clients to be notified of a spend that has
+	// already happened. The notifier should dispatch a spend notification
+	// immediately. We register one that also listen for mempool spends,
+	// both should be notified the same way, as the spend is already mined.
+	checkSpends := func() {
+		const numClients = 2
+		spendClients := make([]*chainntnfs.SpendEvent, numClients)
+		for i := 0; i < numClients; i++ {
+			spentIntent, err := notifier.RegisterSpendNtfn(outpoint,
+				uint32(currentHeight), i%2 == 0)
+			if err != nil {
+				t.Fatalf("unable to register for spend ntfn: %v",
+					err)
+			}
+
+			spendClients[i] = spentIntent
+		}
+
+		for _, client := range spendClients {
+			select {
+			case ntfn := <-client.Spend:
+				// We've received the spend nftn. So now verify
+				// all the fields have been set properly.
+				if *ntfn.SpentOutPoint != *outpoint {
+					t.Fatalf("ntfn includes wrong output, "+
+						"reports %v instead of %v",
+						ntfn.SpentOutPoint, outpoint)
+				}
+				if !bytes.Equal(ntfn.SpenderTxHash[:], spenderSha[:]) {
+					t.Fatalf("ntfn includes wrong spender "+
+						"tx sha, reports %v instead of %v",
+						ntfn.SpenderTxHash[:], spenderSha[:])
+				}
+				if ntfn.SpenderInputIndex != 0 {
+					t.Fatalf("ntfn includes wrong spending "+
+						"input index, reports %v, "+
+						"should be %v",
+						ntfn.SpenderInputIndex, 0)
+				}
+				if ntfn.SpendingHeight != currentHeight {
+					t.Fatalf("ntfn has wrong spending "+
+						"height: expected %v, got %v",
+						currentHeight,
+						ntfn.SpendingHeight)
+				}
+			case <-time.After(30 * time.Second):
+				t.Fatalf("spend ntfn never received")
+			}
+		}
 	}
 
-	spentNtfn := make(chan *chainntnfs.SpendDetail)
-	go func() {
-		spentNtfn <- <-spentIntent.Spend
-	}()
-
+	// Wait for the notifier to have caught up to the mined block.
 	select {
-	case ntfn := <-spentNtfn:
-		// We've received the spend nftn. So now verify all the fields
-		// have been set properly.
-		if *ntfn.SpentOutPoint != *outpoint {
-			t.Fatalf("ntfn includes wrong output, reports %v instead of %v",
-				ntfn.SpentOutPoint, outpoint)
+	case _, ok := <-epochClient.Epochs:
+		if !ok {
+			t.Fatalf("epoch channel was closed")
 		}
-		if !bytes.Equal(ntfn.SpenderTxHash[:], spenderSha[:]) {
-			t.Fatalf("ntfn includes wrong spender tx sha, reports %v instead of %v",
-				ntfn.SpenderTxHash[:], spenderSha[:])
-		}
-		if ntfn.SpenderInputIndex != 0 {
-			t.Fatalf("ntfn includes wrong spending input index, reports %v, should be %v",
-				ntfn.SpenderInputIndex, 0)
-		}
-	case <-time.After(30 * time.Second):
-		t.Fatalf("spend ntfn never received")
+	case <-time.After(15 * time.Second):
+		t.Fatalf("did not receive block epoch")
 	}
+
+	// Check that the spend clients gets immediately notified for the spend
+	// in the previous block.
+	checkSpends()
+
+	// Bury the spend even deeper, and do the same check.
+	const numBlocks = 10
+	if _, err := miner.Node.Generate(numBlocks); err != nil {
+		t.Fatalf("unable to generate single block: %v", err)
+	}
+
+	// Wait for the notifier to have caught up with the new blocks.
+	for i := 0; i < numBlocks; i++ {
+		select {
+		case _, ok := <-epochClient.Epochs:
+			if !ok {
+				t.Fatalf("epoch channel was closed")
+			}
+		case <-time.After(15 * time.Second):
+			t.Fatalf("did not receive block epoch")
+		}
+	}
+
+	// The clients should still be notified immediately.
+	checkSpends()
 }
 
 func testCancelSpendNtfn(node *rpctest.Harness,