func (s *watchableStore) moveVictims() (moved int) {
	s.mu.Lock()
	victims := s.victims
	s.victims = nil
	s.mu.Unlock()

	var newVictim watcherBatch
	for _, wb := range victims {
		// try to send responses again
		for w, eb := range wb {
			// watcher has observed the store up to, but not including, w.minRev
			rev := w.minRev - 1
			if w.send(WatchResponse{WatchID: w.id, Events: eb.evs, Revision: rev}) {
				pendingEventsGauge.Add(float64(len(eb.evs)))
			} else {
				if newVictim == nil {
					newVictim = make(watcherBatch)
				}
				newVictim[w] = eb
				continue
			}
			moved++
		}

		// assign completed victim watchers to unsync/sync
		s.mu.Lock()
		s.store.revMu.RLock()
		curRev := s.store.currentRev
		for w, eb := range wb {
			if newVictim != nil && newVictim[w] != nil {
				// couldn't send watch response; stays victim
				continue
			}
			w.victim = false
			if eb.moreRev != 0 {
				w.minRev = eb.moreRev
			}
			if w.minRev <= curRev {
				s.unsynced.add(w)
			} else {
				slowWatcherGauge.Dec()
				s.synced.add(w)
			}
		}
		s.store.revMu.RUnlock()
		s.mu.Unlock()
	}

	if len(newVictim) > 0 {
		s.mu.Lock()
		s.victims = append(s.victims, newVictim)
		s.mu.Unlock()
	}

	return moved
}
func (s *watchableStore) moveVictims() (moved int) {
	s.mu.Lock()
	victims := s.victims
	s.victims = nil
	s.mu.Unlock()

	var newVictim watcherBatch
	for _, wb := range victims {
		
		for w, eb := range wb {
			
			rev := w.minRev - 1
			if w.send(WatchResponse{WatchID: w.id, Events: eb.evs, Revision: rev}) {
				pendingEventsGauge.Add(float64(len(eb.evs)))
			} else {
				if newVictim == nil {
					newVictim = make(watcherBatch)
				}
				newVictim[w] = eb
				continue
			}
			moved++
		}

		
		s.mu.Lock()
		s.store.revMu.RLock()
		curRev := s.store.currentRev
		for w, eb := range wb {
			if newVictim != nil && newVictim[w] != nil {
				
				continue
			}
			w.victim = false
			if eb.moreRev != 0 {
				w.minRev = eb.moreRev
			}
			if w.minRev <= curRev {
				s.unsynced.add(w)
			} else {
				slowWatcherGauge.Dec()
				s.synced.add(w)
			}
		}
		s.store.revMu.RUnlock()
		s.mu.Unlock()
	}

	if len(newVictim) > 0 {
		s.mu.Lock()
		s.victims = append(s.victims, newVictim)
		s.mu.Unlock()
	}

	return moved
}
func ( s * watchableStore ) moveVictims ( ) ( moved int ) { s . mu . Lock ( ) victims := s . victims s . victims = nil s . mu . Unlock ( ) var newVictim watcherBatch for _ , wb := range victims { // try to send responses again for w , eb := range wb { // watcher has observed the store up to, but not including, w.minRev rev := w . minRev - 1 if w . send ( WatchResponse { WatchID : w . id , Events : eb . evs , Revision : rev } ) { pendingEventsGauge . Add ( float64 ( len ( eb . evs ) ) ) } else { if newVictim == nil { newVictim = make ( watcherBatch ) } newVictim [ w ] = eb continue } moved ++ } // assign completed victim watchers to unsync/sync s . mu . Lock ( ) s . store . revMu . RLock ( ) curRev := s . store . currentRev for w , eb := range wb { if newVictim != nil && newVictim [ w ] != nil { // couldn't send watch response; stays victim continue } w . victim = false if eb . moreRev != 0 { w . minRev = eb . moreRev } if w . minRev <= curRev { s . unsynced . add ( w ) } else { slowWatcherGauge . Dec ( ) s . synced . add ( w ) } } s . store . revMu . RUnlock ( ) s . mu . Unlock ( ) } if len ( newVictim ) > 0 { s . mu . Lock ( ) s . victims = append ( s . victims , newVictim ) s . mu . Unlock ( ) } return moved }
// moveVictims tries to update watches with already pending event data
moveVictims tries to update watches with already pending event data