Add recover for watcher

prophet_leader.go

@@ -60,7 +60,7 @@ func (p *defaultProphet) enableLeader() {
 	p.coordinator.start()
 
 	p.wn = newWatcherNotifier(p.rt)
-	go p.wn.start()
+	p.wn.start()
 
 	// now, we are leader
 	atomic.StoreInt64(&p.leaderFlag, 1)

watcher.go

@@ -275,20 +275,32 @@ func (wn *watcherNotifier) stop() {
 }
 
 func (wn *watcherNotifier) start() {
-	for {
-		evt, ok := <-wn.eventC
-		if !ok {
-			log.Infof("watcher notifer exited")
-			return
+	go func() {
+		defer func() {
+			if err := recover(); err == nil {
+				log.Errorf("watcher notify failed with %+v, restart later", err)
+				wn.start()
+			}
+		}()
+
+		for {
+			evt, ok := <-wn.eventC
+			if !ok {
+				log.Infof("watcher notifer exited")
+				return
+			}
+
+			log.Debugf("new event: %+v", evt)
+			wn.RLock()
+			wn.watchers.Range(func(key, value interface{}) bool {
+				wt := value.(*watcher)
+				wt.notify(evt)
+				return true
+			})
+			wn.RUnlock()
 		}
+	}()
 
-		log.Debugf("new event: %+v", evt)
-		wn.watchers.Range(func(key, value interface{}) bool {
-			wt := value.(*watcher)
-			wt.notify(evt)
-			return true
-		})
-	}
 }
 
 type snap struct {