Merge pull request #613 from wpaulino/neutrino-rescan-finished

chain: dispatch RescanFinished notification after BlockConnected

chain/neutrino.go

@@ -28,11 +28,12 @@ type NeutrinoClient struct {
 	// We currently support one rescan/notifiction goroutine per client
 	rescan *neutrino.Rescan
 
-	enqueueNotification chan interface{}
+	enqueueNotification     chan interface{}
-	dequeueNotification chan interface{}
+	dequeueNotification     chan interface{}
-	startTime           time.Time
+	startTime               time.Time
-	lastProgressSent    bool
+	lastProgressSent        bool
-	currentBlock        chan *waddrmgr.BlockStamp
+	lastFilteredBlockHeader *wire.BlockHeader
+	currentBlock            chan *waddrmgr.BlockStamp
 
 	quit       chan struct{}
 	rescanQuit chan struct{}
@@ -339,6 +340,7 @@ func (s *NeutrinoClient) Rescan(startHash *chainhash.Hash, addrs []btcutil.Addre
 	s.scanning = true
 	s.finished = false
 	s.lastProgressSent = false
+	s.lastFilteredBlockHeader = nil
 	s.isRescan = true
 
 	bestBlock, err := s.CS.BestBlock()
@@ -432,6 +434,7 @@ func (s *NeutrinoClient) NotifyReceived(addrs []btcutil.Address) error {
 	// Don't need RescanFinished or RescanProgress notifications.
 	s.finished = true
 	s.lastProgressSent = true
+	s.lastFilteredBlockHeader = nil
 
 	// Rescan with just the specified addresses.
 	newRescan := neutrino.NewRescan(
@@ -495,6 +498,7 @@ func (s *NeutrinoClient) onFilteredBlockConnected(height int32,
 		}
 		ntfn.RelevantTxs = append(ntfn.RelevantTxs, rec)
 	}
+
 	select {
 	case s.enqueueNotification <- ntfn:
 	case <-s.quit:
@@ -503,41 +507,12 @@ func (s *NeutrinoClient) onFilteredBlockConnected(height int32,
 		return
 	}
 
-	// Handle RescanFinished notification if required.
+	s.clientMtx.Lock()
-	bs, err := s.CS.BestBlock()
+	s.lastFilteredBlockHeader = header
-	if err != nil {
+	s.clientMtx.Unlock()
-		log.Errorf("Can't get chain service's best block: %s", err)
-		return
-	}
 
-	if bs.Hash == header.BlockHash() {
+	// Handle RescanFinished notification if required.
-		// Only send the RescanFinished notification once.
+	s.dispatchRescanFinished()
-		s.clientMtx.Lock()
-		if s.finished {
-			s.clientMtx.Unlock()
-			return
-		}
-		// Only send the RescanFinished notification once the
-		// underlying chain service sees itself as current.
-		current := s.CS.IsCurrent() && s.lastProgressSent
-		if current {
-			s.finished = true
-		}
-		s.clientMtx.Unlock()
-		if current {
-			select {
-			case s.enqueueNotification <- &RescanFinished{
-				Hash:   &bs.Hash,
-				Height: bs.Height,
-				Time:   header.Timestamp,
-			}:
-			case <-s.quit:
-				return
-			case <-s.rescanQuit:
-				return
-			}
-		}
-	}
 }
 
 // onBlockDisconnected sends appropriate notifications to the notification
@@ -612,6 +587,57 @@ func (s *NeutrinoClient) onBlockConnected(hash *chainhash.Hash, height int32,
 		case <-s.rescanQuit:
 		}
 	}
+
+	// Check if we're able to dispatch our final RescanFinished notification
+	// after processing this block.
+	s.dispatchRescanFinished()
+}
+
+// dispatchRescanFinished determines whether we're able to dispatch our final
+// RescanFinished notification in order to mark the wallet as synced with the
+// chain. If the notification has already been dispatched, then it won't be done
+// again.
+func (s *NeutrinoClient) dispatchRescanFinished() {
+	bs, err := s.CS.BestBlock()
+	if err != nil {
+		log.Errorf("Can't get chain service's best block: %s", err)
+		return
+	}
+
+	s.clientMtx.Lock()
+	if bs.Hash != s.lastFilteredBlockHeader.BlockHash() {
+		s.clientMtx.Unlock()
+		return
+	}
+
+	// Only send the RescanFinished notification once.
+	if s.finished {
+		s.clientMtx.Unlock()
+		return
+	}
+
+	// Only send the RescanFinished notification once the underlying chain
+	// service sees itself as current.
+	s.finished = s.CS.IsCurrent() && s.lastProgressSent
+	if !s.finished {
+		s.clientMtx.Unlock()
+		return
+	}
+
+	header := s.lastFilteredBlockHeader
+	s.clientMtx.Unlock()
+
+	select {
+	case s.enqueueNotification <- &RescanFinished{
+		Hash:   &bs.Hash,
+		Height: bs.Height,
+		Time:   header.Timestamp,
+	}:
+	case <-s.quit:
+		return
+	case <-s.rescanQuit:
+		return
+	}
 }
 
 // notificationHandler queues and dequeues notifications. There are currently