[contiv] Watchers may get closed after server restart if db was compacted

db/keyval/etcdv3/bytes_broker_impl.go

@@ -223,25 +223,56 @@ func (db *BytesConnectionEtcd) Watch(resp func(keyval.BytesWatchResp), closeChan
 }
 
 // watchInternal starts the watch subscription for the key.
-func watchInternal(log logging.Logger, watcher clientv3.Watcher, closeCh chan string, key string, resp func(keyval.BytesWatchResp)) error {
-	recvChan := watcher.Watch(context.Background(), key, clientv3.WithPrefix(), clientv3.WithPrevKV())
+func watchInternal(log logging.Logger, watcher clientv3.Watcher, closeCh chan string, prefix string, resp func(keyval.BytesWatchResp)) error {
+	recvChan := watcher.Watch(context.Background(), prefix, clientv3.WithPrefix(), clientv3.WithPrevKV())
 
-	go func() {
-		registeredKey := key
+	go func(registeredKey string) {
+		var compactRev int64
 		for {
 			select {
-			case wresp := <-recvChan:
+			case wresp, ok := <-recvChan:
+				if !ok {
+					log.WithField("prefix", prefix).Warn("Watch recv channel was closed")
+					if compactRev != 0 {
+						recvChan = watcher.Watch(context.Background(), prefix,
+							clientv3.WithPrefix(), clientv3.WithPrevKV(), clientv3.WithRev(compactRev))
+						log.WithFields(logging.Fields{
+							"prefix": prefix,
+							"rev":    compactRev,
+						}).Warn("Watch recv channel was re-created")
+						compactRev = 0
+						continue
+					}
+					return
+				}
+				if wresp.Canceled {
+					log.WithField("prefix", prefix).Warn("Watch was canceled")
+				}
+				err := wresp.Err()
+				if err != nil {
+					log.WithFields(logging.Fields{
+						"prefix": prefix,
+						"err":    err,
+					}).Warn("Watch returned error")
+				}
+				compactRev = wresp.CompactRevision
+				if compactRev != 0 {
+					log.WithFields(logging.Fields{
+						"prefix": prefix,
+						"rev":    compactRev,
+					}).Warn("Watched data were compacted ")
+				}
 				for _, ev := range wresp.Events {
 					handleWatchEvent(log, resp, ev)
 				}
 			case closeVal, ok := <-closeCh:
-				if !ok || registeredKey == closeVal {
-					log.WithField("key", key).Debug("Watch ended")
+				if !ok || closeVal == registeredKey {
+					log.WithField("prefix", prefix).Debug("Watch ended")
 					return
 				}
 			}
 		}
-	}()
+	}(prefix)
 	return nil
 }