mvcc: avoid negative watcher count metrics

The watch count metrics are not robust to duplicate cancellations. These
cause the count to be decremented twice, leading eventually to negative
counts. We are seeing this in production. The duplicate cancellations
themselves are not themselves a big problem (except performance), but
they are caused by the new proactive cancellation logic (#11850). As it
turns out, w.closingc seems to receive two messages for a cancellation.
I have added a fix which ensures that we won't send duplicate cancel
requests.

clientv3/watch.go

@@ -650,7 +650,11 @@ func (w *watchGrpcStream) run() {
 			return
 
 		case ws := <-w.closingc:
-			if ws.id != -1 {
+			if ws.id == -1 {
+				// this stream hasn't actually started, don't attempt to cancel
+			} else if _, ok := w.substreams[ws.id]; !ok {
+				// this is a duplicate cancellation, the substream no longer exists
+			} else {
 				// client is closing an established watch; close it on the server proactively instead of waiting
 				// to close when the next message arrives
 				cancelSet[ws.id] = struct{}{}

mvcc/watchable_store.go

@@ -153,10 +153,13 @@ func (s *watchableStore) cancelWatcher(wa *watcher) {
 		s.mu.Lock()
 		if s.unsynced.delete(wa) {
 			slowWatcherGauge.Dec()
+			watcherGauge.Dec()
 			break
 		} else if s.synced.delete(wa) {
+			watcherGauge.Dec()
 			break
 		} else if wa.compacted {
+			watcherGauge.Dec()
 			break
 		} else if wa.ch == nil {
 			// already canceled (e.g., cancel/close race)
@@ -177,6 +180,7 @@ func (s *watchableStore) cancelWatcher(wa *watcher) {
 		}
 		if victimBatch != nil {
 			slowWatcherGauge.Dec()
+			watcherGauge.Dec()
 			delete(victimBatch, wa)
 			break
 		}
@@ -186,7 +190,6 @@ func (s *watchableStore) cancelWatcher(wa *watcher) {
 		time.Sleep(time.Millisecond)
 	}
 
-	watcherGauge.Dec()
 	wa.ch = nil
 	s.mu.Unlock()
 }