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

Code before:
						t.Fatalf("%s(%d) mismatch (-want +got):\n%s", tc.method, tc.pivot, diff)
					}
				})
			}
		})
	}
}

func TestIterationWithBatching(t *testing.T) {
	rb := newRingBuffer(6)
	events := []*clientv3.Event{
		{Kv: &mvccpb.KeyValue{Key: []byte("key-a"), ModRevision: 5}},
		{Kv: &mvccpb.KeyValue{Key: []byte("key-b-1"), ModRevision: 10}},
		{Kv: &mvccpb.KeyValue{Key: []byte("key-b-2"), ModRevision: 10}},
		{Kv: &mvccpb.KeyValue{Key: []byte("key-b-3"), ModRevision: 10}},
		{Kv: &mvccpb.KeyValue{Key: []byte("key-c"), ModRevision: 12}},
	}

	rb.Append(events)

	tests := []struct {
		name      string
		method    iterMethod
		pivot     int64
		want      []int64
		wantSizes []int
		expected  string
	}{
		{
			name:      "ascending_gte_includes_batched_revision",
			method:    ascendGTE,
			pivot:     10,
			want:      []int64{10, 12},
			wantSizes: []int{3, 1},
			expected:  "Should include rev 10 (with 3 batched events) and rev 12",
		},
		{
			name:      "ascending_lt_stops_before_batched_revision",
			method:    ascendLT,
			pivot:     10,
			want:      []int64{5},
			wantSizes: []int{1},
			expected:  "Should only include rev 5, stopping before the batched rev 10",
		},
		{
			name:      "all_revisions_with_proper_batch_sizes",
			method:    ascendGTE,
			pivot:     0,
			want:      []int64{5, 10, 12},
			wantSizes: []int{1, 3, 1},
			expected:  "Should include all revisions with correct event counts per batch",

Review comment (suggestion):
Compare whole responses, not just single fields.
```suggestion
		want      [][]*client.Event
```

Code after:
		{Kv: &mvccpb.KeyValue{Key: []byte("key-a"), ModRevision: 5}},
		{Kv: &mvccpb.KeyValue{Key: []byte("key-b-1"), ModRevision: 10}},
		{Kv: &mvccpb.KeyValue{Key: []byte("key-b-2"), ModRevision: 10}},
		{Kv: &mvccpb.KeyValue{Key: []byte("key-b-3"), ModRevision: 10}},
		{Kv: &mvccpb.KeyValue{Key: []byte("key-c"), ModRevision: 12}},
	}

	rb.Append(events)

	tests := []struct {
		name   string
		method iterMethod
		pivot  int64
		want   [][]*clientv3.Event
	}{
		{
			name:   "ascending_gte_includes_batched_revision",
			method: ascendGTE,
			pivot:  10,
			want: [][]*clientv3.Event{
				{
					{Kv: &mvccpb.KeyValue{Key: []byte("key-b-1"), ModRevision: 10}},
					{Kv: &mvccpb.KeyValue{Key: []byte("key-b-2"), ModRevision: 10}},
					{Kv: &mvccpb.KeyValue{Key: []byte("key-b-3"), ModRevision: 10}},
				},
				{
					{Kv: &mvccpb.KeyValue{Key: []byte("key-c"), ModRevision: 12}},
				},
			},
		},
		{
			name:   "ascending_lt_stops_before_batched_revision",
			method: ascendLT,
			pivot:  10,
			want: [][]*clientv3.Event{
				{
					{Kv: &mvccpb.KeyValue{Key: []byte("key-a"), ModRevision: 5}},
				},
			},
		},
		{
			name:   "all_revisions_with_proper_batch_sizes",
			method: ascendGTE,
			pivot:  0,
			want: [][]*clientv3.Event{
				{
					{Kv: &mvccpb.KeyValue{Key: []byte("key-a"), ModRevision: 5}},
				},
				{
					{Kv: &mvccpb.KeyValue{Key: []byte("key-b-1"), ModRevision: 10}},
					{Kv: &mvccpb.KeyValue{Key: []byte("key-b-2"), ModRevision: 10}},
Repository: etcd-io/etcd
File: cache/ringbuffer.go (Go)

Code before:
		return
	}

	for n, i := 0, r.tail; n < r.size; n, i = n+1, (i+1)%len(r.buffer) {
		eventBatch := r.buffer[i]

		if eventBatch.events == nil {
			panic(fmt.Sprintf("ringBuffer.AscendLessThan: unexpected nil at %d", i))
		}

		if eventBatch.rev >= pivot {
			return
		}

		if !iter(eventBatch.rev, eventBatch.events) {
			return
		}
	}
}

func (r *ringBuffer) DescendGreaterThan(pivot int64, iter IterFunc) {
	if r.size == 0 {
		return
	}

	for n, i := 0, (r.head-1+len(r.buffer))%len(r.buffer); n < r.size; n, i = n+1, (i-1+len(r.buffer))%len(r.buffer) {
		eventBatch := r.buffer[i]

		if eventBatch.events == nil {
			panic(fmt.Sprintf("ringBuffer.DescendGreaterThan: unexpected nil at %d", i))
		}

		if eventBatch.rev <= pivot {
			return
		}

		if !iter(eventBatch.rev, eventBatch.events) {
			return
		}
	}
}

func (r *ringBuffer) DescendLessOrEqual(pivot int64, iter IterFunc) {
	if r.size == 0 {
		return
	}

	for n, i := 0, (r.head-1+len(r.buffer))%len(r.buffer); n < r.size; n, i = n+1, (i-1+len(r.buffer))%len(r.buffer) {
		eventBatch := r.buffer[i]

		if eventBatch.events == nil {

Review comment (performance):
```suggestion
	for n, i := 0, r.moduloIndex(r.head-1); n < r.size; n, i = n+1, r.moduloIndex(i-1) {
```

Maybe a helper function would be good 
```
func (r *ringBuffer) moduloIndex(index int) int {
  return (index+len(r.buffer))%len(r.buffer)
}
```



Code after:
		return
	}

	for n, i := 0, r.tail; n < r.size; n, i = n+1, (i+1)%len(r.buffer) {
		eventBatch := r.buffer[i]

		if eventBatch.events == nil {
			panic(fmt.Sprintf("ringBuffer.AscendLessThan: unexpected nil at %d", i))
		}

		if eventBatch.rev >= pivot {
			return
		}

		if !iter(eventBatch.rev, eventBatch.events) {
			return
		}
	}
}

func (r *ringBuffer) DescendGreaterThan(pivot int64, iter IterFunc) {
	if r.size == 0 {
		return
	}

	for n, i := 0, r.moduloIndex(r.head-1); n < r.size; n, i = n+1, r.moduloIndex(i-1) {
		eventBatch := r.buffer[i]

		if eventBatch.events == nil {
			panic(fmt.Sprintf("ringBuffer.DescendGreaterThan: unexpected nil at %d", i))
		}

		if eventBatch.rev <= pivot {
			return
		}

		if !iter(eventBatch.rev, eventBatch.events) {
			return
		}
	}
}

func (r *ringBuffer) DescendLessOrEqual(pivot int64, iter IterFunc) {
	if r.size == 0 {
		return
	}

	for n, i := 0, r.moduloIndex(r.head-1); n < r.size; n, i = n+1, r.moduloIndex(i-1) {
		eventBatch := r.buffer[i]

		if eventBatch.events == nil {
Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:

func (c *Cache) getWatch(ctx context.Context) error {
	getResp, err := c.get(ctx)
	if err != nil {
		return err
	}
	return c.watch(ctx, getResp.Header.Revision+1)
}

func (c *Cache) get(ctx context.Context) (*clientv3.GetResponse, error) {
	resp, err := c.kv.Get(ctx, c.prefix, clientv3.WithPrefix())
	if err != nil {
		return nil, err
	}
	c.store.Restore(resp.Kvs, resp.Header.Revision)
	return resp, nil
}

func (c *Cache) watch(ctx context.Context, rev int64) error {
	readyOnce := sync.Once{}
	for {
		storeW := newWatcher(c.cfg.PerWatcherBufferSize*4, nil)
		c.demux.Register(storeW, rev)
		applyErr := make(chan error, 1)
		c.waitGroup.Add(1)
		go c.applyStorage(ctx, storeW, applyErr)

		watchCh := c.watcher.Watch(
			ctx,
			c.prefix,
			clientv3.WithPrefix(),
			clientv3.WithRev(rev),
			clientv3.WithProgressNotify(),
			clientv3.WithCreatedNotify(),
		)

		err := c.watchEvents(ctx, watchCh, applyErr, &readyOnce)
		c.demux.Unregister(storeW)

		if ctx.Err() != nil {
			return ctx.Err()
		}
		if err != nil {
			return err
		}
	}
}

func (c *Cache) applyStorage(ctx context.Context, storeW *watcher, applyErr chan<- error) {
	defer c.waitGroup.Done()
	for {

Review comment (suggestion):
```suggestion
		c.waitGroup.Add(1)
		go func() {
				defer c.waitGroup.Done()
				go c.applyStorage(ctx, storeW, applyErr)
		}()

```

Code after:
}

func (c *Cache) getWatch() error {
	getResp, err := c.get(c.internalCtx)
	if err != nil {
		return err
	}
	return c.watch(getResp.Header.Revision + 1)
}

func (c *Cache) get(ctx context.Context) (*clientv3.GetResponse, error) {
	resp, err := c.kv.Get(ctx, c.prefix, clientv3.WithPrefix())
	if err != nil {
		return nil, err
	}
	c.store.Restore(resp.Kvs, resp.Header.Revision)
	return resp, nil
}

func (c *Cache) watch(rev int64) error {
	readyOnce := sync.Once{}
	for {
		storeW := newWatcher(c.cfg.PerWatcherBufferSize, nil)
		c.demux.Register(storeW, rev)
		applyErr := make(chan error, 1)
		c.waitGroup.Add(1)
		go func() {
			defer c.waitGroup.Done()
			if err := c.applyStorage(storeW); err != nil {
				applyErr <- err
			}
			close(applyErr)
		}()

		watchCh := c.watcher.Watch(
			c.internalCtx,
			c.prefix,
			clientv3.WithPrefix(),
			clientv3.WithRev(rev),
			clientv3.WithProgressNotify(),
			clientv3.WithCreatedNotify(),
		)

		err := c.watchEvents(watchCh, applyErr, &readyOnce)
		c.demux.Unregister(storeW)

		if err != nil {
			return err
		}
	}
}
Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:
		watchCh := c.watcher.Watch(
			ctx,
			c.prefix,
			clientv3.WithPrefix(),
			clientv3.WithRev(rev),
			clientv3.WithProgressNotify(),
			clientv3.WithCreatedNotify(),
		)

		err := c.watchEvents(ctx, watchCh, applyErr, &readyOnce)
		c.demux.Unregister(storeW)

		if ctx.Err() != nil {
			return ctx.Err()
		}
		if err != nil {
			return err
		}
	}
}

func (c *Cache) applyStorage(ctx context.Context, storeW *watcher, applyErr chan<- error) {
	defer c.waitGroup.Done()
	for {
		select {
		case <-ctx.Done():
			return
		case <-c.internalCtx.Done():
			return
		case events, ok := <-storeW.eventQueue:
			if !ok {
				return
			}
			if err := c.store.Apply(events); err != nil {
				select {
				case applyErr <- err:
				default:
				}
				return
			}
		}
	}
}

func (c *Cache) watchEvents(ctx context.Context, watchCh clientv3.WatchChan, applyErr <-chan error, readyOnce *sync.Once) error {
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case resp, ok := <-watchCh:
			if !ok {

Review comment (performance):
Isn't ctx passed here is the same context as c.internalCtx. Recommend to remove ctx from arguments if you plan to have it as method on Cache.

Code after:
			defer c.waitGroup.Done()
			if err := c.applyStorage(storeW); err != nil {
				applyErr <- err
			}
			close(applyErr)
		}()

		watchCh := c.watcher.Watch(
			c.internalCtx,
			c.prefix,
			clientv3.WithPrefix(),
			clientv3.WithRev(rev),
			clientv3.WithProgressNotify(),
			clientv3.WithCreatedNotify(),
		)

		err := c.watchEvents(watchCh, applyErr, &readyOnce)
		c.demux.Unregister(storeW)

		if err != nil {
			return err
		}
	}
}

func (c *Cache) applyStorage(storeW *watcher) error {
	for {
		select {
		case <-c.internalCtx.Done():
			return nil
		case events, ok := <-storeW.eventQueue:
			if !ok {
				return nil
			}
			if err := c.store.Apply(events); err != nil {
				return err
			}
		}
	}
}

func (c *Cache) watchEvents(watchCh clientv3.WatchChan, applyErr <-chan error, readyOnce *sync.Once) error {
	for {
		select {
		case <-c.internalCtx.Done():
			return c.internalCtx.Err()
		case resp, ok := <-watchCh:
			if !ok {
				return nil
			}
			readyOnce.Do(func() { c.ready.Set() })
Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:
		if ctx.Err() != nil {
			return ctx.Err()
		}
		if err != nil {
			return err
		}
	}
}

func (c *Cache) applyStorage(ctx context.Context, storeW *watcher, applyErr chan<- error) {
	defer c.waitGroup.Done()
	for {
		select {
		case <-ctx.Done():
			return
		case <-c.internalCtx.Done():
			return
		case events, ok := <-storeW.eventQueue:
			if !ok {
				return
			}
			if err := c.store.Apply(events); err != nil {
				select {
				case applyErr <- err:
				default:
				}
				return
			}
		}
	}
}

func (c *Cache) watchEvents(ctx context.Context, watchCh clientv3.WatchChan, applyErr <-chan error, readyOnce *sync.Once) error {
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case resp, ok := <-watchCh:
			if !ok {
				return nil
			}
			readyOnce.Do(func() { c.ready.Set() })
			if err := resp.Err(); err != nil {
				c.ready.Reset()
				c.demux.Purge()
				return err
			}
			c.demux.Broadcast(resp.Events)
		case err := <-applyErr:
			c.ready.Reset()
			c.demux.Purge()

Review comment (suggestion):
```suggestion
				applyErr <- err
```
I don't think you should skip error. Even if this should not happen it's bad to imply that sending error is optional.

Code after:
			clientv3.WithProgressNotify(),
			clientv3.WithCreatedNotify(),
		)

		err := c.watchEvents(watchCh, applyErr, &readyOnce)
		c.demux.Unregister(storeW)

		if err != nil {
			return err
		}
	}
}

func (c *Cache) applyStorage(storeW *watcher) error {
	for {
		select {
		case <-c.internalCtx.Done():
			return nil
		case events, ok := <-storeW.eventQueue:
			if !ok {
				return nil
			}
			if err := c.store.Apply(events); err != nil {
				return err
			}
		}
	}
}

func (c *Cache) watchEvents(watchCh clientv3.WatchChan, applyErr <-chan error, readyOnce *sync.Once) error {
	for {
		select {
		case <-c.internalCtx.Done():
			return c.internalCtx.Err()
		case resp, ok := <-watchCh:
			if !ok {
				return nil
			}
			readyOnce.Do(func() { c.ready.Set() })
			if err := resp.Err(); err != nil {
				c.ready.Reset()
				c.demux.Purge()
				return err
			}
			c.demux.Broadcast(resp.Events)
		case err := <-applyErr:
			c.ready.Reset()
			c.demux.Purge()
			return err
		}
	}
Repository: etcd-io/etcd
File: cache/watcher.go (Go)

Code before:
		}
		if len(filtered) == 0 {
			return true
		}
		eventBatch = filtered
	}
	select {
	case w.respQueue <- clientv3.WatchResponse{Events: eventBatch}:
		return true
	default:
		return false
	}
}

func (w *watcher) notifyCompacted(compactRev int64) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if atomic.LoadInt32(&w.stopped) != 0 {
		return
	}
	w.respQueue <- clientv3.WatchResponse{
		Canceled:        true,
		CompactRevision: compactRev,
		CancelReason:    rpctypes.ErrCompacted.Error(),
	}
	atomic.StoreInt32(&w.stopped, 1)
	close(w.respQueue)
	close(w.done)
}

// Stop closes the event channel atomically.
func (w *watcher) Stop() {
	w.mu.Lock()
	defer w.mu.Unlock()
	if atomic.CompareAndSwapInt32(&w.stopped, 0, 1) {
		close(w.respQueue)
		close(w.done)
	}
}

func (w *watcher) Done() <-chan struct{} { return w.done }


Review comment (suggestion):
Something to maybe cleanup in the future (not needed in this PR): we don't need atomics when we're reading and writing to `w.stopped` under a lock anyway. 

Code after:
		}
		eventBatch = filtered
	}
	select {
	case w.eventQueue <- eventBatch:
		return true
	default:
		return false
	}
}

func (w *watcher) Compact(compactRev int64) {
	resp := &clientv3.WatchResponse{
		Canceled:        true,
		CompactRevision: compactRev,
		CancelReason:    rpctypes.ErrCompacted.Error(),
	}
	w.stopOnce.Do(func() {
		w.cancelResp = resp
		close(w.eventQueue)
	})
}

// Stop closes the event channel atomically.
func (w *watcher) Stop() {
	w.stopOnce.Do(func() {
		close(w.eventQueue)
	})
}
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:

	classifyWatchers := func(m map[*watcher]int64) {
		for w, next := range m {
			if isCompaction && next != 0 && next <= compactRev {
				compactList = append(compactList, w)
			} else {
				stopList = append(stopList, w)
			}
			delete(m, w)
		}
	}

	classifyWatchers(d.activeWatchers)
	classifyWatchers(d.laggingWatchers)
	d.activeWatchers = make(map[*watcher]int64)
	d.laggingWatchers = make(map[*watcher]int64)

	d.mu.Unlock()

	for _, w := range compactList {
		w.notifyCompacted(compactRev)
	}
	for _, w := range stopList {
		w.Stop()
	}
}

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

	oldestRev := d.history.PeekOldest()
	if oldestRev == 0 {
		d.mu.Unlock()
		return
	}

	type compactNotice struct {
		w   *watcher
		rev int64
	}
	var compacted []compactNotice

	for w, nextRev := range d.laggingWatchers {
		if nextRev < oldestRev {
			compacted = append(compacted, compactNotice{w: w, rev: 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 {

Review comment (suggestion):
Don't like the idea of doing it outside of lock. This makes it easy to introduce races in code. Like this code racing against resync. 

Have you considered separating identifying which watchers to compact notifying them later? We could send compaction in `resyncLaggingWatchers`, this way all the logic is executed by one loop.

Code after:
	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)
}

// 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.history.RebaseHistory()

	for w, next := range d.activeWatchers {
		if next != 0 && next <= compactRev {
			delete(d.activeWatchers, w)
			d.laggingWatchers[w] = next
		}
	}
}

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

	oldestRev := d.history.PeekOldest()
	if oldestRev == 0 {
		return
	}

	for w, nextRev := range d.laggingWatchers {
		if nextRev < oldestRev {
			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.enqueueEvent(eventBatch) { // buffer overflow: watcher still lagging
				enqueueFailed = true
				return false
			}
			nextRev = rev + 1
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
		}
		sendStart := len(events)
		for i, ev := range events {
			if ev.Kv.ModRevision >= nextRev {
				sendStart = i
				break
			}
		}
		if sendStart == len(events) {
			continue
		}

		if !w.enqueueEvent(events[sendStart:]) { // 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.purge(false, 0) }

// PurgeWithCompaction sends CompactRevision cancels to watchers with too-old next revisions, then stops others to avoid a resync race.
func (d *demux) PurgeWithCompaction(compactRev int64) { d.purge(true, compactRev) }

func (d *demux) purge(isCompaction bool, compactRev int64) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.history.RebaseHistory()

	for w, next := range d.activeWatchers {
		if isCompaction && next != 0 && next <= compactRev {
			delete(d.activeWatchers, w)
			d.laggingWatchers[w] = next
		} else {
			w.Stop()
			delete(d.activeWatchers, w)
		}
	}
	for w, next := range d.laggingWatchers {
		if isCompaction && next != 0 && next <= compactRev {
			d.laggingWatchers[w] = next
		} else {
			w.Stop()
			delete(d.laggingWatchers, w)
		}
	}
}


Review comment (suggestion):
```suggestion
func (d *demux) Compact(compactRev int64) { d.purge(true, compactRev) }
```

Code after:
		}
		sendStart := len(events)
		for i, ev := range events {
			if ev.Kv.ModRevision >= nextRev {
				sendStart = i
				break
			}
		}
		if sendStart == len(events) {
			continue
		}

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

// Purge stops all watchers and rebase history on watch errors
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)
}

// 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.history.RebaseHistory()

	for w, next := range d.activeWatchers {
		if next != 0 && next <= compactRev {
			delete(d.activeWatchers, w)
			d.laggingWatchers[w] = next
		}
	}
}
Repository: etcd-io/etcd
File: cache/watcher.go (Go)

Code before:
// 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

import (
	"sync"
	"sync/atomic"

	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
	clientv3 "go.etcd.io/etcd/client/v3"
)

// watcher holds one client’s buffered stream of events.
type watcher struct {
	respQueue chan clientv3.WatchResponse
	keyPred   KeyPredicate
	stopped   int32
	done      chan struct{} // closed together with Stop()
	mu        sync.Mutex
}

func newWatcher(bufSize int, pred KeyPredicate) *watcher {
	return &watcher{
		respQueue: make(chan clientv3.WatchResponse, bufSize),
		keyPred:   pred,
		done:      make(chan struct{}),
	}
}

// true  -> events delivered (or filtered/duplicate)
// false -> buffer full (caller should mark watcher “lagging”)
func (w *watcher) enqueueEvent(eventBatch []*clientv3.Event) (enqueued bool) {
	if w.keyPred != nil {
		filtered := make([]*clientv3.Event, 0, len(eventBatch))
		for _, event := range eventBatch {
			if w.keyPred(event.Kv.Key) {
				filtered = append(filtered, event)
			}
		}
		if len(filtered) == 0 {
			return true
		}

Review comment (suggestion):
Reading the code again I'm missing why do we even need separate `stopped` variable and separate done channel.

Code after:
// 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

import (
	"sync"

	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
	clientv3 "go.etcd.io/etcd/client/v3"
)

// watcher holds one client’s buffered stream of events.
type watcher struct {
	eventQueue chan []*clientv3.Event
	cancelResp *clientv3.WatchResponse
	keyPred    KeyPredicate
	stopOnce   sync.Once
}

func newWatcher(bufSize int, pred KeyPredicate) *watcher {
	return &watcher{
		eventQueue: make(chan []*clientv3.Event, bufSize),
		keyPred:    pred,
	}
}

// true  -> events delivered (or filtered/duplicate)
// false -> buffer full (caller should mark watcher “lagging”)
func (w *watcher) enqueueEvent(eventBatch []*clientv3.Event) bool {
	if w.keyPred != nil {
		filtered := make([]*clientv3.Event, 0, len(eventBatch))
		for _, event := range eventBatch {
			if w.keyPred(event.Kv.Key) {
				filtered = append(filtered, event)
			}
		}
		if len(filtered) == 0 {
			return true
		}
		eventBatch = filtered
	}
	select {