Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
}

// Compact is called when etcd reports a compaction at compactRev to rebase history;
// it keeps provably-too-old watchers for later cancellation, stops others, and clients should resubscribe.
func (d *demux) Compact(compactRev int64) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.purge()
}

func (d *demux) resyncLaggingWatchers() {
	d.mu.Lock()
	defer d.mu.Unlock()

	if d.minRev == 0 {
		return
	}

	for w, nextRev := range d.laggingWatchers {
		if nextRev < d.minRev {
			w.Compact(nextRev)
			delete(d.laggingWatchers, w)
			continue
		}
		// TODO: re-enable key‐predicate in Filter when non‐zero startRev or performance tuning is needed
		enqueueFailed := false
		d.history.AscendGreaterOrEqual(nextRev, func(rev int64, eventBatch []*clientv3.Event) bool {
			if !w.enqueueResponse(clientv3.WatchResponse{
				Events: eventBatch,
			}) { // buffer overflow: watcher still lagging
				enqueueFailed = true
				return false
			}
			nextRev = rev + 1
			return true
		})
		// Send progress to just resync.
		if !enqueueFailed && d.maxRev > nextRev {
			if w.enqueueResponse(clientv3.WatchResponse{
				Header: etcdserverpb.ResponseHeader{Revision: d.maxRev},
			}) {
				nextRev = d.maxRev + 1
			}
		}

		if !enqueueFailed {
			delete(d.laggingWatchers, w)
			d.activeWatchers[w] = nextRev
		} else {
			d.laggingWatchers[w] = nextRev
		}

Review comment (bug):
I think I should have pointed out before. Please avoid negative logic and double negations like `!enqueueFailed`. Negations complicate logic and make it harder to follow. 
* Use variables with positive connection, `resyncSuccess` over `enqueueFailed`.
* Start if-else statements from positive option, `if resyncSuccess {} else {}` instead of `if !resyncSuccess {} else {}`

Instead do:
```
resyncSuccess := true

if !w.enqueueResponse(resp) {
  resyncSuccess = false
}

if resyncSuccess {
} else {
}
```

Code after:
	for w := range d.activeWatchers {
		w.Stop()
	}
	for w := range d.laggingWatchers {
		w.Stop()
	}
	d.activeWatchers = make(map[*watcher]int64)
	d.laggingWatchers = make(map[*watcher]int64)
}

// Compact is called when etcd reports a compaction at compactRev to rebase history;
// it keeps provably-too-old watchers for later cancellation, stops others, and clients should resubscribe.
func (d *demux) Compact(compactRev int64) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.purge()
}

func (d *demux) resyncLaggingWatchers() {
	d.mu.Lock()
	defer d.mu.Unlock()

	if d.minRev == 0 {
		return
	}

	for w, nextRev := range d.laggingWatchers {
		if nextRev < d.minRev {
			w.Compact(nextRev)
			delete(d.laggingWatchers, w)
			continue
		}
		// TODO: re-enable key‐predicate in Filter when non‐zero startRev or performance tuning is needed
		resyncSuccess := true
		d.history.AscendGreaterOrEqual(nextRev, func(rev int64, eventBatch []*clientv3.Event) bool {
			resp := clientv3.WatchResponse{
				Events: eventBatch,
			}
			if !w.enqueueResponse(resp) { // buffer overflow: watcher still lagging
				resyncSuccess = false
				return false
			}
			nextRev = rev + 1
			return true
		})
		// Send progress to just resync.
		if resyncSuccess {
			resp := clientv3.WatchResponse{
				Header: etcdserverpb.ResponseHeader{Revision: d.maxRev},
			}
			if d.maxRev > nextRev && w.enqueueResponse(resp) {