@@ -357,12 +357,21 @@ func (s *watchableStore) syncWatchers(evs []mvccpb.Event) (int, []mvccpb.Event)
357
357
curRev := s .store .currentRev
358
358
compactionRev := s .store .compactMainRev
359
359
360
- wg , minRev := s .unsynced .choose (maxWatchersPerSync , curRev , compactionRev )
360
+ minRev := s .unsynced .chooseAll (curRev , compactionRev )
361
+ for _ , batch := range s .victims {
362
+ for w , b := range batch {
363
+ watcherRev := max (w .minRev , b .moreRev )
364
+ if len (b .evs ) > 0 {
365
+ watcherRev = max (watcherRev , b .evs [len (b .evs )- 1 ].Kv .ModRevision )
366
+ }
367
+ minRev = min (minRev , watcherRev )
368
+ }
369
+ }
361
370
evs = rangeEventsWithReuse (s .store .lg , s .store .b , evs , minRev , curRev + 1 )
362
371
363
372
victims := make (watcherBatch )
364
- wb := newWatcherBatch (wg , evs )
365
- for w := range wg .watchers {
373
+ wb := newWatcherBatch (& s . unsynced , evs )
374
+ for w := range s . unsynced .watchers {
366
375
if w .minRev < compactionRev {
367
376
// Skip the watcher that failed to send compacted watch response due to w.ch is full.
368
377
// Next retry of syncWatchers would try to resend the compacted watch response to w.ch
0 commit comments