ronantakizawa/codereview-bench huggingface dataset:
Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:
	stream, w, err := c.newWatchEventStream(ctx, key, requestedRev, opts)
	if err != nil {
		// terminal error → one-shot reply, needs a single buffer slot
		responseChan := make(chan clientv3.WatchResponse, 1)
		if errors.Is(err, rpctypes.ErrCompacted) {
			var compactRev int64
			if oldestBatch := c.demux.PeekOldest(); oldestBatch != nil {
				compactRev = oldestBatch.rev
			}
			responseChan <- clientv3.WatchResponse{CompactRevision: compactRev}
		} else {
			responseChan <- clientv3.WatchResponse{Canceled: true}
		}
		close(responseChan)
		return responseChan
	}

	ctx, cancel := context.WithCancel(ctx)
	go func() { <-w.Done(); cancel() }()

	responseChan := make(chan clientv3.WatchResponse)
	go func() {
		defer cancel()
		defer close(responseChan)

		for batch := range stream {
			select {
			case <-ctx.Done():
				return
			case responseChan <- clientv3.WatchResponse{Events: batch.events}:
			}
		}
	}()
	return responseChan
}

// Ready reports whether the cache has finished its initial load.
func (c *Cache) Ready() bool {
	select {
	case <-c.ready:
		return true
	default:
		return false
	}
}

// WaitReady blocks until the cache is ready or the ctx is cancelled.
func (c *Cache) WaitReady(ctx context.Context) error {
	select {
	case <-c.ready:
		return nil

Review comment (performance):
Renaming everything to `batch` just makes things harder to read. This stream should be for `[]*clientv3.Event`

Code after:
			responseChan <- clientv3.WatchResponse{CompactRevision: compactRev}
		} else {
			responseChan <- clientv3.WatchResponse{Canceled: true}
		}
		close(responseChan)
		return responseChan
	}

	ctx, cancel := context.WithCancel(ctx)
	go func() { <-w.Done(); cancel() }()

	responseChan := make(chan clientv3.WatchResponse)
	go func() {
		defer cancel()
		defer close(responseChan)

		for events := range stream {
			select {
			case <-ctx.Done():
				return
			case responseChan <- clientv3.WatchResponse{Events: events}:
			}
		}
	}()
	return responseChan
}

// Ready reports whether the cache has finished its initial load.
func (c *Cache) Ready() bool {
	select {
	case <-c.ready:
		return true
	default:
		return false
	}
}

// WaitReady blocks until the cache is ready or the ctx is cancelled.
func (c *Cache) WaitReady(ctx context.Context) error {
	select {
	case <-c.ready:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// Close cancels the private context and blocks until all goroutines return.
func (c *Cache) Close() {
	c.stop()
	c.waitGroup.Wait()
Repository: etcd-io/etcd
File: cache/ringbuffer_test.go (Go)

Code before:
	}

	for _, tt := range tests {
		tt := tt
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()
			rb := newRingBuffer(tt.capacity)
			for _, r := range tt.revs {
				rb.Append(makeBatch(r, "k", 11))
			}

			gotBatches := rb.Filter(tt.predicate)
			gotRevs := make([]int64, len(gotBatches))
			for i, b := range gotBatches {
				gotRevs[i] = b.rev
			}

			if diff := cmp.Diff(tt.wantFilteredRevs, gotRevs); diff != "" {
				t.Fatalf("Filter() revisions mismatch (-want +got):\n%s", diff)
			}
		})
	}
}

func TestAtomicOrdered(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name     string
		capacity int
		inputs   []struct {
			rev  int64
			key  string
			size int
		}
		predicate BatchPredicate
		wantRev   []int64
		wantSize  []int
	}{
		{
			name:     "key_filter",
			capacity: 4,
			inputs: []struct {
				rev  int64
				key  string
				size int
			}{
				{100, "foo", 15},
			},
			predicate: func(b *batch) bool {
				for _, ev := range b.events {

Review comment (suggestion):
```suggestion

```
Stop with this `t.Parallel()`, you misunderstood how to use it. Remove all it's usage.

Code after:
			}

			gotBatches := rb.Filter(tt.minRev)
			gotRevs := make([]int64, len(gotBatches))
			for i, b := range gotBatches {
				gotRevs[i] = b[0].Kv.ModRevision
			}

			if diff := cmp.Diff(tt.wantFilteredRevs, gotRevs); diff != "" {
				t.Fatalf("Filter() revisions mismatch (-want +got):\n%s", diff)
			}
		})
	}
}

func TestAtomicOrdered(t *testing.T) {
	tests := []struct {
		name     string
		capacity int
		inputs   []struct {
			rev  int64
			key  string
			size int
		}
		wantRev  []int64
		wantSize []int
	}{
		{
			name:     "unfiltered",
			capacity: 5,
			inputs: []struct {
				rev  int64
				key  string
				size int
			}{
				{5, "a", 1},
				{10, "b", 3},
				{15, "c", 7},
				{20, "d", 11},
			},
			wantRev:  []int64{5, 10, 15, 20},
			wantSize: []int{1, 3, 7, 11},
		},
		{
			name:     "across_wrap",
			capacity: 3,
			inputs: []struct {
				rev  int64
				key  string
				size int
			}{
Repository: etcd-io/etcd
File: cache/predicate.go (Go)

Code before:
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package cache

type Prefix string

func (prefix Prefix) Match(key []byte) bool {
	if prefix == "" {
		return true
	}
	prefixLen := len(prefix)
	return len(key) >= prefixLen && string(key[:prefixLen]) == string(prefix)
}

// AfterRev matches batches whose revision ≥ rev.
func AfterRev(rev int64) BatchPredicate {
	return func(b *batch) bool {
		return b.rev >= rev
	}
}


Review comment (suggestion):
Don't think this predicate makes sense, just remove it and pass revision directly to `Filter` function.

Code after:
// Copyright 2025 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package cache

type Prefix string

func (prefix Prefix) Match(key []byte) bool {
	if prefix == "" {
		return true
	}
	prefixLen := len(prefix)
	return len(key) >= prefixLen && string(key[:prefixLen]) == string(prefix)
}
Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:
	return cache, nil
}

// Watch registers a cache-backed watcher for a given key or prefix.
// It returns a WatchChan that streams WatchResponses containing batches.
func (c *Cache) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan {
	select {
	case <-c.ready:
	case <-ctx.Done():
		emptyWatchChan := make(chan clientv3.WatchResponse)
		close(emptyWatchChan)
		return emptyWatchChan
	}

	op := clientv3.OpGet(key, opts...)
	requestedRev := op.Rev()

	// build the internal Watch event stream (for historic reply + live)
	stream, w, err := c.newWatchEventStream(ctx, key, requestedRev, opts)
	if err != nil {
		// terminal error → one-shot reply, needs a single buffer slot
		responseChan := make(chan clientv3.WatchResponse, 1)
		if errors.Is(err, rpctypes.ErrCompacted) {
			var compactRev int64
			if oldestBatch := c.demux.PeekOldest(); oldestBatch != nil {
				compactRev = oldestBatch.rev
			}
			responseChan <- clientv3.WatchResponse{CompactRevision: compactRev}
		} else {
			responseChan <- clientv3.WatchResponse{Canceled: true}
		}
		close(responseChan)
		return responseChan
	}

	ctx, cancel := context.WithCancel(ctx)
	go func() { <-w.Done(); cancel() }()

	responseChan := make(chan clientv3.WatchResponse)
	go func() {
		defer cancel()
		defer close(responseChan)

		for batch := range stream {
			select {
			case <-ctx.Done():
				return
			case responseChan <- clientv3.WatchResponse{Events: batch.events}:
			}
		}
	}()

Review comment (suggestion):
```suggestion
			compactRev = c.demux.PeekOldest()
```

Code after:
	select {
	case <-c.ready:
	case <-ctx.Done():
		emptyWatchChan := make(chan clientv3.WatchResponse)
		close(emptyWatchChan)
		return emptyWatchChan
	}

	op := clientv3.OpGet(key, opts...)
	requestedRev := op.Rev()

	// build the internal Watch event stream (for historic reply + live)
	stream, w, err := c.newWatchEventStream(ctx, key, requestedRev, opts)
	if err != nil {
		// terminal error → one-shot reply, needs a single buffer slot
		responseChan := make(chan clientv3.WatchResponse, 1)
		if errors.Is(err, rpctypes.ErrCompacted) {
			compactRev := c.demux.PeekOldest()
			responseChan <- clientv3.WatchResponse{CompactRevision: compactRev}
		} else {
			responseChan <- clientv3.WatchResponse{Canceled: true}
		}
		close(responseChan)
		return responseChan
	}

	ctx, cancel := context.WithCancel(ctx)
	go func() { <-w.Done(); cancel() }()

	responseChan := make(chan clientv3.WatchResponse)
	go func() {
		defer cancel()
		defer close(responseChan)

		for events := range stream {
			select {
			case <-ctx.Done():
				return
			case responseChan <- clientv3.WatchResponse{Events: events}:
			}
		}
	}()
	return responseChan
}

// Ready reports whether the cache has finished its initial load.
func (c *Cache) Ready() bool {
	select {
	case <-c.ready:
		return true
	default:
Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:
	if key != c.prefix || !clientv3.IsOptsWithPrefix(opts) || startRev != 0 {
		return nil, nil, ErrUnsupportedWatch
	}

	pred := func(k []byte) bool { return strings.HasPrefix(string(k), key) }

	w := newWatcher(c.cfg.PerWatcherBufferSize, pred)
	c.demux.Register(w, startRev)

	go func() {
		<-ctx.Done()
		c.demux.Unregister(w)
	}()

	return w.batchQueue, w, nil
}

func serveWatchEvents(ctx context.Context, watchCtx *watchCtx) {
	backoff := watchCtx.backoffStart
	for {
		opts := []clientv3.OpOption{
			clientv3.WithPrefix(),
			clientv3.WithProgressNotify(),
			clientv3.WithCreatedNotify(),
		}
		if oldestBatch := watchCtx.cache.demux.PeekOldest(); oldestBatch != nil {
			opts = append(opts,
				clientv3.WithRev(oldestBatch.rev+1))
		}
		watchCh := watchCtx.cache.watcher.Watch(ctx, watchCtx.cache.prefix, opts...)

		if err := readWatchChannel(watchCh, watchCtx.cache, watchCtx.cache.demux, watchCtx.onFirstResponse); err == nil {
			return
		}

		select {
		case <-ctx.Done():
			return
		case <-time.After(backoff):
		}
		if backoff < watchCtx.backoffMax {
			backoff *= 2
		}
	}
}

// readWatchChannel reads an etcd Watch stream into History and enqueueCh, returning nil on cancel or first error.
func readWatchChannel(
	watchChan clientv3.WatchChan,
	cache *Cache,
	demux *demux,

Review comment (performance):
```suggestion
		if oldestRev := watchCtx.cache.demux.PeekOldest(); oldestRev != 0 {
```

Code after:
	go func() {
		<-ctx.Done()
		c.demux.Unregister(w)
	}()

	return w.eventQueue, w, nil
}

func serveWatchEvents(ctx context.Context, watchCtx *watchCtx) {
	backoff := watchCtx.backoffStart
	for {
		opts := []clientv3.OpOption{
			clientv3.WithPrefix(),
			clientv3.WithProgressNotify(),
			clientv3.WithCreatedNotify(),
		}
		if oldestRev := watchCtx.cache.demux.PeekOldest(); oldestRev != 0 {
			opts = append(opts,
				clientv3.WithRev(oldestRev+1))
		}
		watchCh := watchCtx.cache.watcher.Watch(ctx, watchCtx.cache.prefix, opts...)

		if err := readWatchChannel(watchCh, watchCtx.cache, watchCtx.cache.demux, watchCtx.onFirstResponse); err == nil {
			return
		}

		select {
		case <-ctx.Done():
			return
		case <-time.After(backoff):
		}
		if backoff < watchCtx.backoffMax {
			backoff *= 2
		}
	}
}

// readWatchChannel reads an etcd Watch stream into History and enqueueCh, returning nil on cancel or first error.
func readWatchChannel(
	watchChan clientv3.WatchChan,
	cache *Cache,
	demux *demux,
	onFirstResponse func(),
) error {
	for resp := range watchChan {
		onFirstResponse()

		if err := resp.Err(); err != nil {
			if errors.Is(err, rpctypes.ErrCompacted) {
				select {
				case <-cache.ready:
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
type demux struct {
	mu sync.RWMutex
	// activeWatchers & laggingWatchers hold the first revision the watcher still needs (nextRev).
	activeWatchers  map[*watcher]int64
	laggingWatchers map[*watcher]int64
	history         *ringBuffer
	resyncInterval  time.Duration
}

func newDemux(ctx context.Context, wg *sync.WaitGroup, historyWindowSize int, resyncInterval time.Duration) *demux {
	d := &demux{
		activeWatchers:  make(map[*watcher]int64),
		laggingWatchers: make(map[*watcher]int64),
		history:         newRingBuffer(historyWindowSize),
		resyncInterval:  resyncInterval,
	}

	wg.Add(1)
	go func() {
		defer wg.Done()
		d.resyncLoop(ctx)
	}()
	return d
}

// resyncLoop periodically tries to catch lagging watchers up by replaying batches from History.
func (d *demux) resyncLoop(ctx context.Context) {
	timer := time.NewTimer(d.resyncInterval)
	defer timer.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
			d.resyncLaggingWatchers()
			timer.Reset(d.resyncInterval)
		}
	}
}

func (d *demux) Register(w *watcher, startingRev int64) {
	d.mu.Lock()
	defer d.mu.Unlock()

	latestBatch := d.history.PeekLatest()

	// Special case: 0 means “newest”.
	if startingRev == 0 {
		if latestBatch == nil {
			d.activeWatchers[w] = 0

Review comment (suggestion):
```suggestion
// resyncLoop periodically tries to catch lagging watchers up by replaying events from History.
```

Code after:
)

type demux struct {
	mu sync.RWMutex
	// activeWatchers & laggingWatchers hold the first revision the watcher still needs (nextRev).
	activeWatchers  map[*watcher]int64
	laggingWatchers map[*watcher]int64
	history         *ringBuffer
	resyncInterval  time.Duration
}

func NewDemux(ctx context.Context, wg *sync.WaitGroup, historyWindowSize int, resyncInterval time.Duration) *demux {
	d := newDemux(historyWindowSize, resyncInterval)
	wg.Add(1)
	go func() {
		defer wg.Done()
		d.resyncLoop(ctx)
	}()
	return d
}

func newDemux(historyWindowSize int, resyncInterval time.Duration) *demux {
	return &demux{
		activeWatchers:  make(map[*watcher]int64),
		laggingWatchers: make(map[*watcher]int64),
		history:         newRingBuffer(historyWindowSize),
		resyncInterval:  resyncInterval,
	}
}

// resyncLoop periodically tries to catch lagging watchers up by replaying events from History.
func (d *demux) resyncLoop(ctx context.Context) {
	timer := time.NewTimer(d.resyncInterval)
	defer timer.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
			d.resyncLaggingWatchers()
			timer.Reset(d.resyncInterval)
		}
	}
}

func (d *demux) Register(w *watcher, startingRev int64) {
	d.mu.Lock()
	defer d.mu.Unlock()

	latestRev := d.history.PeekLatest()
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
		d.resyncLoop(ctx)
	}()
	return d
}

// resyncLoop periodically tries to catch lagging watchers up by replaying batches from History.
func (d *demux) resyncLoop(ctx context.Context) {
	timer := time.NewTimer(d.resyncInterval)
	defer timer.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
			d.resyncLaggingWatchers()
			timer.Reset(d.resyncInterval)
		}
	}
}

func (d *demux) Register(w *watcher, startingRev int64) {
	d.mu.Lock()
	defer d.mu.Unlock()

	latestBatch := d.history.PeekLatest()

	// Special case: 0 means “newest”.
	if startingRev == 0 {
		if latestBatch == nil {
			d.activeWatchers[w] = 0
			return
		}
		startingRev = latestBatch.rev + 1
	}

	var latestRev int64
	if latestBatch != nil {
		latestRev = latestBatch.rev
	}
	if startingRev <= latestRev {
		d.laggingWatchers[w] = startingRev
	} else {
		d.activeWatchers[w] = startingRev
	}
}

func (d *demux) Unregister(w *watcher) {
	func() {
		d.mu.Lock()
		defer d.mu.Unlock()

Review comment (suggestion):
```suggestion
	latestRev := d.history.PeekLatest()
```

Code after:

func newDemux(historyWindowSize int, resyncInterval time.Duration) *demux {
	return &demux{
		activeWatchers:  make(map[*watcher]int64),
		laggingWatchers: make(map[*watcher]int64),
		history:         newRingBuffer(historyWindowSize),
		resyncInterval:  resyncInterval,
	}
}

// resyncLoop periodically tries to catch lagging watchers up by replaying events from History.
func (d *demux) resyncLoop(ctx context.Context) {
	timer := time.NewTimer(d.resyncInterval)
	defer timer.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
			d.resyncLaggingWatchers()
			timer.Reset(d.resyncInterval)
		}
	}
}

func (d *demux) Register(w *watcher, startingRev int64) {
	d.mu.Lock()
	defer d.mu.Unlock()

	latestRev := d.history.PeekLatest()

	// Special case: 0 means “newest”.
	if startingRev == 0 {
		if latestRev == 0 {
			d.activeWatchers[w] = 0
			return
		}
		startingRev = latestRev + 1
	}

	if startingRev <= latestRev {
		d.laggingWatchers[w] = startingRev
	} else {
		d.activeWatchers[w] = startingRev
	}
}

func (d *demux) Unregister(w *watcher) {
	func() {
		d.mu.Lock()
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
		}
		startingRev = latestBatch.rev + 1
	}

	var latestRev int64
	if latestBatch != nil {
		latestRev = latestBatch.rev
	}
	if startingRev <= latestRev {
		d.laggingWatchers[w] = startingRev
	} else {
		d.activeWatchers[w] = startingRev
	}
}

func (d *demux) Unregister(w *watcher) {
	func() {
		d.mu.Lock()
		defer d.mu.Unlock()
		delete(d.activeWatchers, w)
		delete(d.laggingWatchers, w)
	}()
	w.Stop()
}

func (d *demux) Broadcast(b *batch) {
	d.mu.Lock()
	defer d.mu.Unlock()

	d.history.Append(b)
	for w, nextRev := range d.activeWatchers {
		if b.rev < nextRev {
			continue
		}
		if !w.enqueueBatch(b) { // buffer overflow
			d.laggingWatchers[w] = nextRev
			delete(d.activeWatchers, w)
		} else {
			d.activeWatchers[w] = b.rev + 1
		}
	}
}

// Purge is called when etcd compaction invalidates our cached history, so clients should resubscribe.
func (d *demux) Purge() {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.history.RebaseHistory()
	for w := range d.activeWatchers {
		w.Stop()
	}

Review comment (suggestion):
```suggestion
func (d *demux) Broadcast([]*clientv3.Event) {
```

Code after:
	// Special case: 0 means “newest”.
	if startingRev == 0 {
		if latestRev == 0 {
			d.activeWatchers[w] = 0
			return
		}
		startingRev = latestRev + 1
	}

	if startingRev <= latestRev {
		d.laggingWatchers[w] = startingRev
	} else {
		d.activeWatchers[w] = startingRev
	}
}

func (d *demux) Unregister(w *watcher) {
	func() {
		d.mu.Lock()
		defer d.mu.Unlock()
		delete(d.activeWatchers, w)
		delete(d.laggingWatchers, w)
	}()
	w.Stop()
}

func (d *demux) Broadcast(events []*clientv3.Event) {
	if len(events) == 0 {
		return
	}

	d.mu.Lock()
	defer d.mu.Unlock()

	d.history.Append(events)

	lastRev := events[len(events)-1].Kv.ModRevision
	for w, nextRev := range d.activeWatchers {
		start := len(events)
		for i, ev := range events {
			if ev.Kv.ModRevision >= nextRev {
				start = i
				break
			}
		}

		if start == len(events) {
			continue
		}

		if !w.enqueueEvent(events[start:]) { // overflow → lagging
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
	}
}

// Purge is called when etcd compaction invalidates our cached history, so clients should resubscribe.
func (d *demux) Purge() {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.history.RebaseHistory()
	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)
}

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

	for w, nextRev := range d.laggingWatchers {
		var oldestRev int64
		if oldestBatch := d.history.PeekOldest(); oldestBatch != nil {
			oldestRev = oldestBatch.rev
		}

		if oldestRev != 0 && nextRev < oldestRev {
			w.Stop()
			delete(d.laggingWatchers, w)
			continue
		}
		// TODO: re-enable key‐predicate in Filter when non‐zero startRev or performance tuning is needed
		missedBatches := d.history.Filter(AfterRev(nextRev))

		for _, batch := range missedBatches {
			if !w.enqueueBatch(batch) { // buffer overflow: watcher still lagging
				break
			}
			nextRev = batch.rev + 1
		}

		if len(missedBatches) > 0 && nextRev > missedBatches[len(missedBatches)-1].rev {
			delete(d.laggingWatchers, w)
			d.activeWatchers[w] = nextRev
		} else {
			d.laggingWatchers[w] = nextRev
		}
	}
}

Review comment (suggestion):
```suggestion
		oldestRev = d.history.PeekOldest()
```

Code after:
			if ev.Kv.ModRevision >= nextRev {
				start = i
				break
			}
		}

		if start == len(events) {
			continue
		}

		if !w.enqueueEvent(events[start:]) { // overflow → lagging
			d.laggingWatchers[w] = nextRev
			delete(d.activeWatchers, w)
		} else {
			d.activeWatchers[w] = lastRev + 1
		}
	}
}

// Purge is called when etcd compaction invalidates our cached history, so clients should resubscribe.
func (d *demux) Purge() {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.history.RebaseHistory()
	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)
}

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

	oldestRev := d.history.PeekOldest()
	for w, nextRev := range d.laggingWatchers {
		if oldestRev != 0 && nextRev < oldestRev {
			w.Stop()
			delete(d.laggingWatchers, w)
			continue
		}
		// TODO: re-enable key‐predicate in Filter when non‐zero startRev or performance tuning is needed
		missed := d.history.Filter(nextRev)

		enqueueFailed := false
		for _, eventBatch := range missed {
			if !w.enqueueEvent(eventBatch) { // buffer overflow: watcher still lagging
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:

		if oldestRev != 0 && nextRev < oldestRev {
			w.Stop()
			delete(d.laggingWatchers, w)
			continue
		}
		// TODO: re-enable key‐predicate in Filter when non‐zero startRev or performance tuning is needed
		missedBatches := d.history.Filter(AfterRev(nextRev))

		for _, batch := range missedBatches {
			if !w.enqueueBatch(batch) { // buffer overflow: watcher still lagging
				break
			}
			nextRev = batch.rev + 1
		}

		if len(missedBatches) > 0 && nextRev > missedBatches[len(missedBatches)-1].rev {
			delete(d.laggingWatchers, w)
			d.activeWatchers[w] = nextRev
		} else {
			d.laggingWatchers[w] = nextRev
		}
	}
}

func (d *demux) PeekOldest() *batch {
	d.mu.RLock()
	defer d.mu.RUnlock()
	return d.history.PeekOldest()
}


Review comment (suggestion):
```suggestion
func (d *demux) PeekOldest() int64 {
```

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

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

	oldestRev := d.history.PeekOldest()
	for w, nextRev := range d.laggingWatchers {
		if oldestRev != 0 && nextRev < oldestRev {
			w.Stop()
			delete(d.laggingWatchers, w)
			continue
		}
		// TODO: re-enable key‐predicate in Filter when non‐zero startRev or performance tuning is needed
		missed := d.history.Filter(nextRev)

		enqueueFailed := false
		for _, eventBatch := range missed {
			if !w.enqueueEvent(eventBatch) { // buffer overflow: watcher still lagging
				enqueueFailed = true
				break
			}
			nextRev = eventBatch[0].Kv.ModRevision + 1
		}

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

func (d *demux) PeekOldest() int64 {
	d.mu.RLock()
	defer d.mu.RUnlock()
	return d.history.PeekOldest()
}